memind/memory-v2-pgvector-backfill.test.mjs

import assert from 'node:assert/strict';
import test from 'node:test';
import {
  backfillLegacyMemoriesToPgvector,
  loadLegacyMemoryBackfillBatch,
  loadLegacyUserMemorySyncBatch,
  syncLegacyUserMemoriesToPgvector,
} from './memory-v2-pgvector-backfill.mjs';

function createMysqlPool(rows) {
  const calls = [];
  return {
    calls,
    async query(sql, params) {
      calls.push({ sql, params });
      const [afterUpdatedAt, sameUpdatedAt, afterId, limit] = params;
      const filtered = rows
        .filter((row) => row.status === 'active')
        .filter((row) => row.updated_at > afterUpdatedAt || (row.updated_at === sameUpdatedAt && row.id > afterId))
        .sort((left, right) => left.updated_at - right.updated_at || left.id.localeCompare(right.id))
        .slice(0, limit);
      return [filtered];
    },
  };
}

const legacyRows = [
  {
    id: 'mem-1',
    user_id: 'user-1',
    label: 'fact',
    memory_text: '用户关注 Memory V2 facade',
    evidence_message_id: 'msg-1',
    source_session_id: 'session-1',
    confidence: '0.800',
    status: 'active',
    created_at: 1000,
    updated_at: 2000,
  },
  {
    id: 'mem-2',
    user_id: 'user-1',
    label: 'preference',
    memory_text: '用户喜欢渐进式架构升级',
    evidence_message_id: null,
    source_session_id: 'session-2',
    confidence: '0.700',
    status: 'active',
    created_at: 1100,
    updated_at: 3000,
  },
  {
    id: 'mem-archived',
    user_id: 'user-1',
    label: 'fact',
    memory_text: 'archived should not move',
    status: 'archived',
    created_at: 1200,
    updated_at: 4000,
  },
];

test('loadLegacyMemoryBackfillBatch reads active memories with cursor ordering', async () => {
  const mysqlPool = createMysqlPool(legacyRows);

  const batch = await loadLegacyMemoryBackfillBatch(mysqlPool, {
    cursor: { updatedAt: 2000, id: 'mem-1' },
    limit: 10,
  });

  assert.equal(batch.memories.length, 1);
  assert.equal(batch.memories[0].id, 'mem-2');
  assert.deepEqual(batch.nextCursor, { updatedAt: 3000, id: 'mem-2' });
  assert.equal(batch.hasMore, false);
  assert.match(mysqlPool.calls[0].sql, /FROM h5_user_memory_items/);
  assert.deepEqual(mysqlPool.calls[0].params, [2000, 2000, 'mem-1', 10]);
});

test('backfillLegacyMemoriesToPgvector dry-run does not embed or write', async () => {
  const mysqlPool = createMysqlPool(legacyRows);
  let embedded = false;
  let written = false;

  const result = await backfillLegacyMemoriesToPgvector({
    mysqlPool,
    pgPool: {
      async query() {
        written = true;
      },
    },
    embedMemory: async () => {
      embedded = true;
      return [0.1, 0.2];
    },
    dryRun: true,
    limit: 1,
  });

  assert.equal(result.mode, 'dry-run');
  assert.equal(result.scanned, 1);
  assert.equal(result.inserted, 0);
  assert.equal(result.hasMore, true);
  assert.equal(embedded, false);
  assert.equal(written, false);
});

test('backfillLegacyMemoriesToPgvector apply embeds and upserts rows', async () => {
  const mysqlPool = createMysqlPool(legacyRows);
  const pgCalls = [];

  const result = await backfillLegacyMemoriesToPgvector({
    mysqlPool,
    pgPool: {
      async query(sql, params) {
        pgCalls.push({ sql, params });
        return { rows: [] };
      },
    },
    embedMemory: async (memory) => {
      assert.match(memory.text, /Memory V2|渐进式/);
      return memory.id === 'mem-1' ? [0.1, 0.2] : [0.3, 0.4];
    },
    dryRun: false,
    limit: 10,
  });

  assert.equal(result.mode, 'apply');
  assert.equal(result.scanned, 2);
  assert.equal(result.inserted, 2);
  assert.equal(pgCalls.length, 2);
  assert.match(pgCalls[0].sql, /INSERT INTO "memory_embeddings"/);
  assert.match(pgCalls[0].sql, /ON CONFLICT \(source_memory_id\)/);
  assert.deepEqual(pgCalls[0].params.slice(0, 8), [
    'user-1',
    '用户关注 Memory V2 facade',
    '[0.1,0.2]',
    'fact',
    'mem-1',
    'session-1',
    'msg-1',
    JSON.stringify({ confidence: 0.8 }),
  ]);
  assert.ok(pgCalls[0].params[8] instanceof Date);
  assert.ok(pgCalls[0].params[9] instanceof Date);
});

test('backfillLegacyMemoriesToPgvector apply requires pg pool and embedding function', async () => {
  const mysqlPool = createMysqlPool(legacyRows);

  await assert.rejects(
    () => backfillLegacyMemoriesToPgvector({
      mysqlPool,
      dryRun: false,
      embedMemory: async () => [0.1],
    }),
    /PostgreSQL pool/,
  );
  await assert.rejects(
    () => backfillLegacyMemoriesToPgvector({
      mysqlPool,
      dryRun: false,
      pgPool: { async query() {} },
    }),
    /embedMemory/,
  );
});

test('loadLegacyUserMemorySyncBatch scopes recent memories to one user and session', async () => {
  const calls = [];
  const mysqlPool = {
    async query(sql, params) {
      calls.push({ sql, params });
      return [[legacyRows[1]]];
    },
  };

  const memories = await loadLegacyUserMemorySyncBatch(mysqlPool, {
    userId: 'user-1',
    sessionId: 'session-2',
    limit: 25,
  });

  assert.equal(memories.length, 1);
  assert.equal(memories[0].id, 'mem-2');
  assert.match(calls[0].sql, /user_id = \?/);
  assert.match(calls[0].sql, /source_session_id = \?/);
  assert.match(calls[0].sql, /ORDER BY updated_at DESC/);
  assert.deepEqual(calls[0].params, ['user-1', 'session-2', 25]);
});

test('syncLegacyUserMemoriesToPgvector immediately upserts the scoped memory', async () => {
  const pgCalls = [];
  const mysqlPool = {
    async query() {
      return [[legacyRows[0]]];
    },
  };

  const result = await syncLegacyUserMemoriesToPgvector({
    mysqlPool,
    pgPool: {
      async query(sql, params) {
        pgCalls.push({ sql, params });
        return { rows: [] };
      },
    },
    embedMemory: async () => [0.2, 0.4],
    userId: 'user-1',
    sessionId: 'session-1',
    limit: 10,
  });

  assert.equal(result.mode, 'user-sync');
  assert.equal(result.scanned, 1);
  assert.equal(result.inserted, 1);
  assert.match(pgCalls[0].sql, /INSERT INTO "memory_embeddings"/);
  assert.equal(pgCalls[0].params[4], 'mem-1');
});