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