0pt/tests/connectors/migration-runner.test.ts

/**
 * `migration-runner.test.ts` — HSP-083 acceptance:
 *   T01: a crash mid-batch produces NO duplicates on resume — the durable
 *        checkpoint + the sink's idempotency make the replayed batch a no-op
 *        (the replay is COUNTED, never hidden).
 *   T02: the cursor advances ONLY after a durable target-ack.
 *   T03: total object count alone is NOT sufficient — VERIFIED requires
 *        per-partition content hashes; count-only evidence is UNVERIFIED.
 */
import { describe, expect, it } from 'vitest';

import {
  InMemoryCheckpointStore,
  runMigration,
  resumeMigration,
  type MigrationBatch,
  type TargetSink,
} from '../../src/connectors/migration-runner';

function aBatch(overrides: Partial<MigrationBatch> = {}): MigrationBatch {
  return {
    partitionKey: 'p1',
    fromCursor: '0',
    toCursor: 'c1',
    records: [{ id: 'r1', name: 'kund AB' }],
    ...overrides,
  };
}

/** An idempotent sink: the same (fromCursor, records) pair acks without re-persisting. */
function idempotentSink(state: { persisted: Map<string, number> }): TargetSink {
  return {
    async write(batch) {
      const key = `${batch.partitionKey}:${batch.fromCursor}`;
      if (state.persisted.has(key)) {
        // Replay: ack, but NOTHING new persisted (the sink's idempotency).
        return { ackToken: `ack_${key}`, persistedCount: 0 };
      }
      state.persisted.set(key, batch.records.length);
      return { ackToken: `ack_${key}`, persistedCount: batch.records.length };
    },
  };
}

describe('HSP-083-T01: krasch mitt i batch ger inte dubbletter', () => {
  it('a resume after a crash replays the batch; the idempotent sink persists NOTHING twice', async () => {
    const state = { persisted: new Map<string, number>() };
    const sink = idempotentSink(state);
    const store = new InMemoryCheckpointStore();
    // First run crashes BEFORE the checkpoint commit (records persisted though).
    const batch = aBatch({ records: [{ id: 'r1' }, { id: 'r2' }] });
    const first = await runMigration([batch], sink, store);
    expect(first.batchesApplied).toBe(1);
    expect(first.totalRecords).toBe(2);

    // Simulate a crash mid-batch: resume from the DURABLE checkpoint (0).
    const report = await resumeMigration('p1', store, () => [batch], sink);
    // The replay is visible in the report...
    expect(report.replayDuplicates).toBe(2);
    // ...but NOTHING was double-persisted.
    expect(state.persisted.get('p1:0')).toBe(2);
    expect(report.totalRecords).toBe(0); // replay acked with persistedCount=0
  });
});

describe('HSP-083-T02: cursor uppdateras efter varaktig targetack', () => {
  it('the checkpoint commits ONLY after the sink acks (T02 ordering)', async () => {
    const store = new InMemoryCheckpointStore();
    const sink = idempotentSink({ persisted: new Map() });
    const batch = aBatch({ toCursor: 'c_42' });
    await runMigration([batch], sink, store);
    // T02: the durable cursor is now the batch's END cursor.
    expect(await store.get('p1')).toBe('c_42');
  });

  it('a fresh partition starts from cursor 0', async () => {
    const store = new InMemoryCheckpointStore();
    expect(await store.get('p_never')).toBeUndefined();
  });
});

describe('HSP-083-T03: total object count räcker inte som enda integritetsbevis', () => {
  it('a successful run is VERIFIED with per-partition content hashes (not count-only)', async () => {
    const sink = idempotentSink({ persisted: new Map() });
    const report = await runMigration([aBatch({ records: [{ id: 'r1' }, { id: 'r2' }] })], sink, new InMemoryCheckpointStore());
    expect(report.state).toBe('VERIFIED');
    expect(report.partitionHashes).toHaveLength(1);
    expect(report.partitionHashes[0]!.contentHash).toMatch(/^fnv1a-/);
    expect(report.partitionHashes[0]!.recordCount).toBe(2);
  });

  it('a run with NO batches is UNVERIFIED (count 0 alone proves nothing)', async () => {
    const sink = idempotentSink({ persisted: new Map() });
    const report = await runMigration([], sink, new InMemoryCheckpointStore());
    expect(report.state).toBe('UNVERIFIED');
    expect(report.partitionHashes).toHaveLength(0);
  });

  it('different content produces different partition hashes (hash ≠ count)', async () => {
    const sink1 = idempotentSink({ persisted: new Map() });
    const sink2 = idempotentSink({ persisted: new Map() });
    const r1 = await runMigration([aBatch({ records: [{ id: 'r1' }] })], sink1, new InMemoryCheckpointStore());
    const r2 = await runMigration([aBatch({ records: [{ id: 'DIFFERENT' }], partitionKey: 'p2' })], sink2, new InMemoryCheckpointStore());
    expect(r1.partitionHashes).toHaveLength(1);
    expect(r2.partitionHashes).toHaveLength(1);
    expect(r1.partitionHashes[0]!.contentHash).not.toBe(r2.partitionHashes[0]!.contentHash);
    // Same count (1 == 1) but different hashes — count alone is insufficient (T03).
    expect(r1.partitionHashes[0]!.recordCount).toBe(r2.partitionHashes[0]!.recordCount);
  });
});