import { describe, expect, it } from 'vitest'; import { ArenaApiError } from '../src/arena/session.js'; import { UNCLASSIFIED_ERROR_PHRASE } from '../src/errors/publicPhrase.js'; import { InMemoryRunStore } from '../src/sync/runStore.js'; import { SyncRunner } from '../src/sync/runner.js'; import type { SyncExecutionContext, SyncExecutionResult, SyncExecutor, SyncObjectType, } from '../src/sync/types.js'; class RecordingExecutor implements SyncExecutor { constructor( readonly objectType: SyncObjectType, private readonly log: SyncObjectType[], private readonly failWith?: Error, ) {} async execute(_ctx: SyncExecutionContext): Promise { this.log.push(this.objectType); if (this.failWith) throw this.failWith; return { processed: 1 }; } } function makeExecutors(log: SyncObjectType[], failing?: Partial>) { const types: SyncObjectType[] = ['PRODUCT', 'CHANGE_MASTER', 'BOM', 'DOCUMENTS']; return new Map( types.map((t) => [t, new RecordingExecutor(t, log, failing?.[t])]), ); } describe('SyncRunner', () => { it('moves a run queued → completed, executing objects sequentially in order', async () => { const log: SyncObjectType[] = []; const store = new InMemoryRunStore(); const runner = new SyncRunner(store, makeExecutors(log)); const run = runner.submit('full', ['PRODUCT', 'CHANGE_MASTER', 'BOM', 'DOCUMENTS']); expect(run.state).toBe('queued'); expect(log).toEqual([]); // nothing executed before submit() returns const final = await runner.waitFor(run.runId); expect(final?.state).toBe('completed'); expect(final?.finishedAt).toBeDefined(); expect(log).toEqual(['PRODUCT', 'CHANGE_MASTER', 'BOM', 'DOCUMENTS']); expect(final?.objects.map((o) => ({ t: o.objectType, s: o.state, p: o.processed }))).toEqual([ { t: 'PRODUCT', s: 'completed', p: 1 }, { t: 'CHANGE_MASTER', s: 'completed', p: 1 }, { t: 'BOM', s: 'completed', p: 1 }, { t: 'DOCUMENTS', s: 'completed', p: 1 }, ]); }); it('respects a custom object order', async () => { const log: SyncObjectType[] = []; const runner = new SyncRunner(new InMemoryRunStore(), makeExecutors(log)); const run = runner.submit('delta', ['BOM', 'PRODUCT']); await runner.waitFor(run.runId); expect(log).toEqual(['BOM', 'PRODUCT']); }); it('stops the chain at the first failure and marks the run failed', async () => { const log: SyncObjectType[] = []; const runner = new SyncRunner( new InMemoryRunStore(), makeExecutors(log, { PRODUCT: new Error('boom') }), ); const run = runner.submit('full', ['PRODUCT', 'CHANGE_MASTER', 'BOM', 'DOCUMENTS']); const final = await runner.waitFor(run.runId); expect(final?.state).toBe('failed'); // B5: an unknown error class maps to the fixed unclassified phrase, never its message. expect(final?.error).toBe(`PRODUCT: ${UNCLASSIFIED_ERROR_PHRASE}`); expect(final?.objects[0]?.error).toBe(UNCLASSIFIED_ERROR_PHRASE); expect(JSON.stringify(final)).not.toContain('boom'); expect(log).toEqual(['PRODUCT']); // later objects never executed expect(final?.objects.map((o) => o.state)).toEqual(['failed', 'pending', 'pending', 'pending']); }); it('B4: `processed` is absent while pending and present once an object has run', async () => { const log: SyncObjectType[] = []; const runner = new SyncRunner( new InMemoryRunStore(), makeExecutors(log, { CHANGE_MASTER: new Error('boom') }), ); const queued = runner.submit('full', ['PRODUCT', 'CHANGE_MASTER', 'BOM']); expect(queued.objects.every((o) => !('processed' in o))).toBe(true); const final = await runner.waitFor(queued.runId); expect(final?.objects.map((o) => [o.state, 'processed' in o])).toEqual([ ['completed', true], ['failed', true], // started, so the counter exists (0 here) ['pending', false], ]); expect(final?.objects[1]?.processed).toBe(0); }); it('B5: an ArenaApiError from an executor becomes a fixed status phrase; its body and message never reach the snapshot', async () => { const log: SyncObjectType[] = []; const upstream = new ArenaApiError('Arena GET /items failed with HTTP 503 MARKER-MESSAGE', 503, { errors: [{ message: 'MARKER-BODY worker@example.test' }], }); const runner = new SyncRunner(new InMemoryRunStore(), makeExecutors(log, { BOM: upstream })); const run = runner.submit('full', ['PRODUCT', 'BOM', 'DOCUMENTS']); const final = await runner.waitFor(run.runId); expect(final?.state).toBe('failed'); expect(final?.objects[1]?.error).toBe('Upstream Arena PLM request failed (HTTP 503).'); expect(final?.error).toBe('BOM: Upstream Arena PLM request failed (HTTP 503).'); const wire = JSON.stringify(final); expect(wire).not.toContain('MARKER-MESSAGE'); expect(wire).not.toContain('MARKER-BODY'); expect(wire).not.toContain('worker@example.test'); }); it('drain() resolves only after every in-flight run has settled (success and failure alike)', async () => { const log: SyncObjectType[] = []; let releaseBom!: () => void; const gate = new Promise((resolve) => { releaseBom = resolve; }); class GatedExecutor implements SyncExecutor { readonly objectType: SyncObjectType = 'BOM'; async execute(): Promise { await gate; log.push('BOM'); return { processed: 1 }; } } const executors = makeExecutors(log, { PRODUCT: new Error('boom') }); executors.set('BOM', new GatedExecutor()); const runner = new SyncRunner(new InMemoryRunStore(), executors); const failing = runner.submit('full', ['PRODUCT']); const gated = runner.submit('full', ['BOM']); expect(runner.inFlightCount).toBe(2); let drained = false; const drain = runner.drain().then(() => { drained = true; }); await new Promise((r) => setImmediate(r)); expect(drained).toBe(false); // the gated BOM run is still executing expect(runner.inFlightCount).toBe(1); // the failing run has already settled releaseBom(); await drain; expect(drained).toBe(true); expect(runner.inFlightCount).toBe(0); expect((await runner.waitFor(failing.runId))?.state).toBe('failed'); expect((await runner.waitFor(gated.runId))?.state).toBe('completed'); }); it('drain() resolves immediately when nothing is in flight', async () => { const runner = new SyncRunner(new InMemoryRunStore(), makeExecutors([])); await expect(runner.drain()).resolves.toBeUndefined(); }); it('fails the run when no executor is registered for an object type', async () => { const runner = new SyncRunner(new InMemoryRunStore(), new Map()); const run = runner.submit('full', ['PRODUCT']); const final = await runner.waitFor(run.runId); expect(final?.state).toBe('failed'); expect(final?.error).toBe('PRODUCT: no executor registered'); }); });