Merge Memory V2 runtime facade

memory-v2-pgvector-backfill.mjs

@@ -0,0 +1,158 @@
+const DEFAULT_LIMIT = 100;
+const DEFAULT_TABLE = 'memory_embeddings';
+const MAX_LIMIT = 1000;
+
+function isSafeIdentifier(value) {
+  return /^[a-zA-Z_][a-zA-Z0-9_]*$/.test(String(value ?? ''));
+}
+
+function quoteIdent(value) {
+  const normalized = String(value ?? '').trim();
+  if (!isSafeIdentifier(normalized)) {
+    throw new Error(`Invalid PostgreSQL identifier: ${normalized}`);
+  }
+  return `"${normalized}"`;
+}
+
+function normalizeLimit(value) {
+  return Math.max(1, Math.min(MAX_LIMIT, Number(value ?? DEFAULT_LIMIT) || DEFAULT_LIMIT));
+}
+
+function normalizeCursor(cursor = {}) {
+  return {
+    updatedAt: Number(cursor.updatedAt ?? 0) || 0,
+    id: String(cursor.id ?? ''),
+  };
+}
+
+function toPgTimestamp(value) {
+  const numeric = Number(value ?? 0);
+  if (!Number.isFinite(numeric) || numeric <= 0) return new Date(0);
+  return new Date(numeric);
+}
+
+function vectorLiteral(embedding) {
+  return `[${embedding.map((item) => Number(item)).join(',')}]`;
+}
+
+function normalizeEmbedding(value) {
+  if (!Array.isArray(value)) return null;
+  const numbers = value.map((item) => Number(item));
+  if (!numbers.length || numbers.some((item) => !Number.isFinite(item))) return null;
+  return numbers;
+}
+
+function normalizeLegacyMemory(row) {
+  const id = String(row?.id ?? '').trim();
+  const userId = String(row?.user_id ?? '').trim();
+  const text = String(row?.memory_text ?? '').trim();
+  if (!id || !userId || !text) return null;
+  return {
+    id,
+    userId,
+    label: String(row?.label ?? 'fact').trim() || 'fact',
+    text,
+    evidenceMessageId: row?.evidence_message_id == null ? null : String(row.evidence_message_id),
+    sourceSessionId: row?.source_session_id == null ? null : String(row.source_session_id),
+    confidence: row?.confidence == null ? null : Number(row.confidence),
+    createdAt: Number(row?.created_at ?? 0) || 0,
+    updatedAt: Number(row?.updated_at ?? 0) || 0,
+  };
+}
+
+export async function loadLegacyMemoryBackfillBatch(mysqlPool, { cursor = {}, limit = DEFAULT_LIMIT } = {}) {
+  if (!mysqlPool?.query) {
+    throw new Error('loadLegacyMemoryBackfillBatch requires a MySQL pool with query(sql, params)');
+  }
+  const resolvedCursor = normalizeCursor(cursor);
+  const resolvedLimit = normalizeLimit(limit);
+  const [rows] = await mysqlPool.query(
+    `SELECT id, user_id, label, memory_text, evidence_message_id, source_session_id,
+            confidence, created_at, updated_at
+     FROM h5_user_memory_items
+     WHERE status = 'active'
+       AND (updated_at > ? OR (updated_at = ? AND id > ?))
+     ORDER BY updated_at ASC, id ASC
+     LIMIT ?`,
+    [resolvedCursor.updatedAt, resolvedCursor.updatedAt, resolvedCursor.id, resolvedLimit],
+  );
+  const memories = (rows ?? []).map((row) => normalizeLegacyMemory(row)).filter(Boolean);
+  const last = memories.at(-1);
+  return {
+    memories,
+    nextCursor: last ? { updatedAt: last.updatedAt, id: last.id } : resolvedCursor,
+    hasMore: memories.length === resolvedLimit,
+  };
+}
+
+export async function backfillLegacyMemoriesToPgvector({
+  mysqlPool,
+  pgPool = null,
+  embedMemory = null,
+  tableName = DEFAULT_TABLE,
+  cursor = {},
+  limit = DEFAULT_LIMIT,
+  dryRun = true,
+} = {}) {
+  const table = quoteIdent(tableName);
+  const batch = await loadLegacyMemoryBackfillBatch(mysqlPool, { cursor, limit });
+  if (dryRun) {
+    return {
+      ok: true,
+      mode: 'dry-run',
+      scanned: batch.memories.length,
+      inserted: 0,
+      nextCursor: batch.nextCursor,
+      hasMore: batch.hasMore,
+    };
+  }
+  if (!pgPool?.query) {
+    throw new Error('backfill apply requires a PostgreSQL pool with query(sql, params)');
+  }
+  if (typeof embedMemory !== 'function') {
+    throw new Error('backfill apply requires embedMemory(memory) => number[]');
+  }
+
+  let inserted = 0;
+  for (const memory of batch.memories) {
+    const embedding = normalizeEmbedding(await embedMemory(memory));
+    if (!embedding) continue;
+    await pgPool.query(
+      `INSERT INTO ${table}
+         (user_id, content, embedding, type, source_memory_id, source_session_id,
+          source_message_id, metadata, created_at, updated_at)
+       VALUES ($1, $2, $3::vector, $4, $5, $6, $7, $8::jsonb, $9, $10)
+       ON CONFLICT (source_memory_id) WHERE source_memory_id IS NOT NULL
+       DO UPDATE SET
+         content = EXCLUDED.content,
+         embedding = EXCLUDED.embedding,
+         type = EXCLUDED.type,
+         source_session_id = EXCLUDED.source_session_id,
+         source_message_id = EXCLUDED.source_message_id,
+         metadata = EXCLUDED.metadata,
+         updated_at = EXCLUDED.updated_at`,
+      [
+        memory.userId,
+        memory.text,
+        vectorLiteral(embedding),
+        memory.label,
+        memory.id,
+        memory.sourceSessionId,
+        memory.evidenceMessageId,
+        JSON.stringify({ confidence: memory.confidence }),
+        toPgTimestamp(memory.createdAt),
+        toPgTimestamp(memory.updatedAt),
+      ],
+    );
+    inserted += 1;
+  }
+
+  return {
+    ok: true,
+    mode: 'apply',
+    scanned: batch.memories.length,
+    inserted,
+    nextCursor: batch.nextCursor,
+    hasMore: batch.hasMore,
+  };
+}