import { parseSinceArg } from './memory-v2-shadow-audit.mjs'; export const MEMORY_V2_PRODUCT_EVENT_TYPES = Object.freeze({ CANDIDATE_SAVED: 'memory_candidate_saved', PROMOTED: 'memory_promoted', RESOLVED_INJECTED: 'memory_resolved_injected', RECALL_HIT: 'memory_recall_hit', }); const TABLE = 'h5_memory_v2_product_events'; export function buildMemoryV2ProductEventsSchemaSql({ table = TABLE } = {}) { if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new Error('Invalid product events table name'); return `CREATE TABLE IF NOT EXISTS \`${table}\` ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY, event_type VARCHAR(64) NOT NULL, user_id CHAR(36) NULL, session_id VARCHAR(191) NULL, run_id CHAR(36) NULL, candidate_id VARCHAR(64) NULL, data_json JSON NULL, created_at BIGINT NOT NULL, KEY idx_mv2_product_event_type_time (event_type, created_at), KEY idx_mv2_product_event_user_time (user_id, created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`; } export async function ensureMemoryV2ProductEventsSchema(pool, options = {}) { if (!pool?.query) throw new Error('Product events schema requires a MySQL pool'); await pool.query(buildMemoryV2ProductEventsSchemaSql(options)); } export async function recordMemoryV2ProductEvent( pool, { eventType, userId = null, sessionId = null, runId = null, candidateId = null, data = null, createdAt = Date.now(), } = {}, { table = TABLE, now = () => Date.now() } = {}, ) { if (!pool?.query || !eventType) return { recorded: false, reason: 'invalid_input' }; const timestamp = Number(createdAt ?? now()) || now(); await pool.query( `INSERT INTO \`${table}\` (event_type, user_id, session_id, run_id, candidate_id, data_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`, [ String(eventType), userId == null ? null : String(userId), sessionId == null ? null : String(sessionId), runId == null ? null : String(runId), candidateId == null ? null : String(candidateId), data == null ? null : JSON.stringify(data), timestamp, ], ); return { recorded: true, eventType: String(eventType), createdAt: timestamp }; } function parseEventData(raw) { if (raw == null) return null; if (typeof raw === 'object') return raw; try { return JSON.parse(String(raw)); } catch { return null; } } export async function tableExists(pool, tableName) { const [rows] = await pool.query( `SELECT 1 AS ok FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = ? LIMIT 1`, [String(tableName)], ); return rows.length > 0; } async function countProductEvents(pool, { sinceMs, userId, eventType, table = TABLE }) { if (!(await tableExists(pool, table))) return 0; const clauses = ['event_type = ?', 'created_at >= ?']; const params = [eventType, sinceMs]; if (userId) { clauses.push('user_id = ?'); params.push(String(userId)); } const [rows] = await pool.query( `SELECT COUNT(*) AS count FROM \`${table}\` WHERE ${clauses.join(' AND ')}`, params, ); return Number(rows[0]?.count ?? 0); } async function countAgentMemoryEvents(pool, { sinceMs, userId, injectionOnly = false }) { if (!(await tableExists(pool, 'h5_agent_run_events'))) return 0; const clauses = ["e.event_type = 'agent_memory_resolved'", 'e.created_at >= ?']; const params = [sinceMs]; if (userId) { clauses.push('r.user_id = ?'); params.push(String(userId)); } if (injectionOnly) { clauses.push(`JSON_EXTRACT(e.data_json, '$.injectionEnabled') = true`); } const [rows] = await pool.query( `SELECT COUNT(*) AS count FROM h5_agent_run_events e INNER JOIN h5_agent_runs r ON r.id = e.run_id WHERE ${clauses.join(' AND ')}`, params, ); return Number(rows[0]?.count ?? 0); } async function countAgentRecallHits(pool, { sinceMs, userId }) { if (!(await tableExists(pool, 'h5_agent_run_events'))) return 0; const clauses = [ "e.event_type = 'agent_memory_resolved'", 'e.created_at >= ?', `JSON_EXTRACT(e.data_json, '$.injectionEnabled') = true`, `CAST(JSON_UNQUOTE(JSON_EXTRACT(e.data_json, '$.memoryCount')) AS UNSIGNED) > 0`, ]; const params = [sinceMs]; if (userId) { clauses.push('r.user_id = ?'); params.push(String(userId)); } const [rows] = await pool.query( `SELECT COUNT(*) AS count FROM h5_agent_run_events e INNER JOIN h5_agent_runs r ON r.id = e.run_id WHERE ${clauses.join(' AND ')}`, params, ); return Number(rows[0]?.count ?? 0); } async function countCandidatesCreated(pool, { sinceMs, userId }) { if (!(await tableExists(pool, 'h5_memory_v2_candidates'))) return 0; const clauses = ['created_at >= ?']; const params = [sinceMs]; if (userId) { clauses.push('user_id = ?'); params.push(String(userId)); } const [rows] = await pool.query( `SELECT COUNT(*) AS count FROM h5_memory_v2_candidates WHERE ${clauses.join(' AND ')}`, params, ); return Number(rows[0]?.count ?? 0); } async function countMemoryItemsCreated(pool, { sinceMs, userId }) { if (!(await tableExists(pool, 'h5_user_memory_items'))) return 0; const clauses = ['created_at >= ?', "status = 'active'"]; const params = [sinceMs]; if (userId) { clauses.push('user_id = ?'); params.push(String(userId)); } const [rows] = await pool.query( `SELECT COUNT(*) AS count FROM h5_user_memory_items WHERE ${clauses.join(' AND ')}`, params, ); return Number(rows[0]?.count ?? 0); } export async function aggregateMemoryV2ProductMetrics( pool, { since = '7d', userId = null, now = Date.now() } = {}, ) { if (!pool?.query) throw new Error('Product metrics require a MySQL pool'); const window = parseSinceArg(since, now); const sinceMs = window.sinceMs; const [ candidateSavedEvents, promotedEvents, injectedEvents, recallHitEvents, candidateSavedFallback, promotedFallback, injectedFallback, recallHitFallback, ] = await Promise.all([ countProductEvents(pool, { sinceMs, userId, eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.CANDIDATE_SAVED, }), countProductEvents(pool, { sinceMs, userId, eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.PROMOTED, }), countProductEvents(pool, { sinceMs, userId, eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.RESOLVED_INJECTED, }), countProductEvents(pool, { sinceMs, userId, eventType: MEMORY_V2_PRODUCT_EVENT_TYPES.RECALL_HIT, }), countCandidatesCreated(pool, { sinceMs, userId }), countMemoryItemsCreated(pool, { sinceMs, userId }), countAgentMemoryEvents(pool, { sinceMs, userId, injectionOnly: true }), countAgentRecallHits(pool, { sinceMs, userId }), ]); const events = { memory_candidate_saved: candidateSavedEvents || candidateSavedFallback, memory_promoted: promotedEvents || promotedFallback, memory_resolved_injected: injectedEvents || injectedFallback, memory_recall_hit: recallHitEvents || recallHitFallback, }; return { window: { since: window.label, sinceMs, untilMs: now, }, userId: userId ? String(userId) : null, events, sources: { memory_candidate_saved: candidateSavedEvents > 0 ? 'product_events' : 'candidates_table', memory_promoted: promotedEvents > 0 ? 'product_events' : 'memory_items_table', memory_resolved_injected: injectedEvents > 0 ? 'product_events' : 'agent_run_events', memory_recall_hit: recallHitEvents > 0 ? 'product_events' : 'agent_run_events', }, }; }