fix(memory): sync new memories into pgvector

2026-07-21 23:27:15 +08:00
parent 6f4a3b53d0
commit c2e2209344
8 changed files with 425 additions and 57 deletions
@@ -60,61 +60,16 @@ function normalizeLegacyMemory(row) {
  };
 }

-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,
-    };
-  }
+async function upsertLegacyMemories({ memories, pgPool, embedMemory, tableName }) {
  if (!pgPool?.query) {
-    throw new Error('backfill apply requires a PostgreSQL pool with query(sql, params)');
+    throw new Error('pgvector sync requires a PostgreSQL pool with query(sql, params)');
  }
  if (typeof embedMemory !== 'function') {
-    throw new Error('backfill apply requires embedMemory(memory) => number[]');
+    throw new Error('pgvector sync requires embedMemory(memory) => number[]');
  }
-
+  const table = quoteIdent(tableName);
  let inserted = 0;
-  for (const memory of batch.memories) {
+  for (const memory of memories) {
    const embedding = normalizeEmbedding(await embedMemory(memory));
    if (!embedding) continue;
    await pgPool.query(
@@ -146,6 +101,109 @@ export async function backfillLegacyMemoriesToPgvector({
    );
    inserted += 1;
  }
+  return inserted;
+}
+
+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 loadLegacyUserMemorySyncBatch(
+  mysqlPool,
+  { userId, sessionId = null, limit = DEFAULT_LIMIT } = {},
+) {
+  if (!mysqlPool?.query) {
+    throw new Error('loadLegacyUserMemorySyncBatch requires a MySQL pool with query(sql, params)');
+  }
+  const resolvedUserId = String(userId ?? '').trim();
+  if (!resolvedUserId) throw new Error('loadLegacyUserMemorySyncBatch requires userId');
+  const resolvedSessionId = String(sessionId ?? '').trim();
+  const resolvedLimit = normalizeLimit(limit);
+  const sessionScope = resolvedSessionId ? ' AND source_session_id = ?' : '';
+  const params = resolvedSessionId
+    ? [resolvedUserId, resolvedSessionId, resolvedLimit]
+    : [resolvedUserId, resolvedLimit];
+  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 user_id = ?${sessionScope}
+     ORDER BY updated_at DESC, id DESC
+     LIMIT ?`,
+    params,
+  );
+  return (rows ?? []).map((row) => normalizeLegacyMemory(row)).filter(Boolean);
+}
+
+export async function syncLegacyUserMemoriesToPgvector({
+  mysqlPool,
+  pgPool,
+  embedMemory,
+  tableName = DEFAULT_TABLE,
+  userId,
+  sessionId = null,
+  limit = DEFAULT_LIMIT,
+} = {}) {
+  const memories = await loadLegacyUserMemorySyncBatch(mysqlPool, { userId, sessionId, limit });
+  const inserted = await upsertLegacyMemories({ memories, pgPool, embedMemory, tableName });
+  return {
+    ok: true,
+    mode: 'user-sync',
+    scanned: memories.length,
+    inserted,
+    userId: String(userId),
+    sessionId: sessionId == null ? null : String(sessionId),
+  };
+}
+
+export async function backfillLegacyMemoriesToPgvector({
+  mysqlPool,
+  pgPool = null,
+  embedMemory = null,
+  tableName = DEFAULT_TABLE,
+  cursor = {},
+  limit = DEFAULT_LIMIT,
+  dryRun = true,
+} = {}) {
+  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,
+    };
+  }
+  const inserted = await upsertLegacyMemories({
+    memories: batch.memories,
+    pgPool,
+    embedMemory,
+    tableName,
+  });

  return {
    ok: true,