memind/memory-v2-pgvector-backfill.mjs

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

async function upsertLegacyMemories({ memories, pgPool, embedMemory, tableName }) {
  if (!pgPool?.query) {
    throw new Error('pgvector sync requires a PostgreSQL pool with query(sql, params)');
  }
  if (typeof embedMemory !== 'function') {
    throw new Error('pgvector sync requires embedMemory(memory) => number[]');
  }
  const table = quoteIdent(tableName);
  let inserted = 0;
  for (const memory of 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 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,
    mode: 'apply',
    scanned: batch.memories.length,
    inserted,
    nextCursor: batch.nextCursor,
    hasMore: batch.hasMore,
  };
}