arena-plm-fiori/worker/test/runner.test.ts

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<SyncExecutionResult> {
    this.log.push(this.objectType);
    if (this.failWith) throw this.failWith;
    return { processed: 1 };
  }
}

function makeExecutors(log: SyncObjectType[], failing?: Partial<Record<SyncObjectType, Error>>) {
  const types: SyncObjectType[] = ['PRODUCT', 'CHANGE_MASTER', 'BOM', 'DOCUMENTS'];
  return new Map<SyncObjectType, SyncExecutor>(
    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<void>((resolve) => {
      releaseBom = resolve;
    });
    class GatedExecutor implements SyncExecutor {
      readonly objectType: SyncObjectType = 'BOM';
      async execute(): Promise<SyncExecutionResult> {
        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');
  });
});