memind/memory-v2-product-events.mjs

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',
    },
  };
}