ApiCore/ApiCore-Backend/src/domain-events/domain-event-dispatcher.spec.ts

import { Logger } from '@nestjs/common';
import { Prisma } from '../../generated/prisma';
import { PrismaService } from '../prisma/prisma.service';
import { domainEventContext } from './domain-event-context';
import {
  DOMAIN_EVENT_BATCH_SIZE,
  DomainEventDispatcher,
  DRAIN_MAX_ROUNDS,
  HANDLER_TX_TIMEOUT_MS,
} from './domain-event-dispatcher';
import type {
  DomainEventRegistry,
  RegisteredDomainEventHandler,
} from './domain-event-registry';

interface ClaimedRow {
  id: string;
  type: string;
  aggregateType: string;
  aggregateId: string;
  payload: unknown;
  actorUserId: string | null;
  correlationId: string;
  occurredAt: Date;
  attempts: number;
  claimedAt: Date;
}

type DeliveryStatus = 'PENDING' | 'DONE' | 'FAILED';

interface UpsertArgs {
  where: unknown;
  create: { lastError: string };
  update: { lastError: string };
}

const CLAIMED_AT = new Date('2026-10-13T08:00:01.234Z');

function row(overrides: Partial<ClaimedRow> = {}): ClaimedRow {
  return {
    id: 'e1',
    type: 'TaskDue',
    aggregateType: 'Task',
    aggregateId: 't1',
    payload: { dueOn: '2026-10-13' },
    actorUserId: null,
    correlationId: 'corr-1',
    occurredAt: new Date('2026-10-13T08:00:00.000Z'),
    attempts: 0,
    claimedAt: CLAIMED_AT,
    ...overrides,
  };
}

function handler(
  id: string,
  invoke: RegisteredDomainEventHandler['invoke'] = () => Promise.resolve(),
): RegisteredDomainEventHandler & { invoke: jest.Mock } {
  return { id, invoke: jest.fn(invoke) };
}

function text(query: Prisma.Sql): string {
  return query.sql.replace(/\s+/g, ' ').trim();
}

/**
 * Stands in for Postgres: a queue of claim results, the delivery rows the
 * lock query finds, and every statement the dispatcher sent.
 */
class FakeDb {
  readonly claims: ClaimedRow[][] = [];
  /** Rows the stale release returns, one list per call. */
  readonly releases: { id: string; status: 'PENDING' | 'FAILED' }[][] = [];
  readonly deliveries = new Map<string, DeliveryStatus>();

  readonly tx = {
    domainEventDelivery: {
      createMany: jest.fn().mockResolvedValue({ count: 1 }),
    },
    $queryRaw: jest.fn((query: Prisma.Sql) => {
      const handlerId = query.values[1] as string;
      return Promise.resolve([
        {
          id: `d-${handlerId}`,
          status: this.deliveries.get(handlerId) ?? 'PENDING',
        },
      ]);
    }),
    $executeRaw: jest.fn().mockResolvedValue(1),
  };

  readonly prisma = {
    $queryRaw: jest.fn((query: Prisma.Sql) =>
      Promise.resolve(
        (query.sql.includes('SKIP LOCKED')
          ? this.claims.shift()
          : this.releases.shift()) ?? [],
      ),
    ),
    $executeRaw: jest.fn().mockResolvedValue(1),
    // Called as (fn, { timeout }); the options only matter to Prisma.
    $transaction: jest.fn((fn: (tx: unknown) => Promise<unknown>) =>
      fn(this.tx),
    ),
    domainEventDelivery: {
      upsert: jest.fn().mockResolvedValue({}),
      updateMany: jest.fn().mockResolvedValue({ count: 1 }),
    },
  };

  claimQuery(call = 0): Prisma.Sql {
    return (this.prisma.$queryRaw.mock.calls as unknown as [Prisma.Sql][])[
      call
    ][0];
  }

  /** Statements on the root client: settle and stale release. */
  rootStatements(): Prisma.Sql[] {
    return (this.prisma.$executeRaw.mock.calls as [Prisma.Sql][]).map(
      ([query]) => query,
    );
  }

  /** Statements inside handler transactions after the lock: delivery DONE. */
  deliveryDoneStatements(): Prisma.Sql[] {
    return (this.tx.$executeRaw.mock.calls as [Prisma.Sql][]).map(
      ([query]) => query,
    );
  }

  upsertArgs(call = 0): UpsertArgs {
    return (
      this.prisma.domainEventDelivery.upsert.mock.calls as [UpsertArgs][]
    )[call][0];
  }
}

describe('DomainEventDispatcher', () => {
  let db: FakeDb;
  let handlers: RegisteredDomainEventHandler[];
  let dispatcher: DomainEventDispatcher;
  let logWarn: jest.SpyInstance;

  beforeEach(() => {
    db = new FakeDb();
    handlers = [];
    const registry = {
      handlersFor: jest.fn(() => handlers),
    } as unknown as DomainEventRegistry;
    dispatcher = new DomainEventDispatcher(
      db.prisma as unknown as PrismaService,
      registry,
    );
    logWarn = jest
      .spyOn(Logger.prototype, 'warn')
      .mockImplementation(() => undefined);
    jest.spyOn(Logger.prototype, 'error').mockImplementation(() => undefined);
  });

  afterEach(() => jest.restoreAllMocks());

  it('pins batch size, handler transaction timeout and drain rounds', () => {
    expect(DOMAIN_EVENT_BATCH_SIZE).toBe(50);
    expect(HANDLER_TX_TIMEOUT_MS).toBe(10_000);
    expect(DRAIN_MAX_ROUNDS).toBe(20);
  });

  it('claims at most 50 pending events, oldest first, skipping locked rows, by the database clock', async () => {
    await dispatcher.runOnce();

    const claim = db.claimQuery();
    expect(text(claim)).toContain(
      `UPDATE "DomainEvent" SET "status" = 'PROCESSING', "claimedAt" = now()`,
    );
    expect(text(claim)).toContain(
      `WHERE "status" = 'PENDING' AND "availableAt" <= now()`,
    );
    expect(text(claim)).toContain(
      'ORDER BY "occurredAt" LIMIT ? FOR UPDATE SKIP LOCKED',
    );
    expect(claim.values).toEqual([50]);
  });

  it('restricts the claim to the given correlation ids', async () => {
    await dispatcher.runOnce({ correlationIds: ['c1', 'c2'] });

    const claim = db.claimQuery();
    expect(text(claim)).toContain(
      `"availableAt" <= now() AND "correlationId" = ANY(?) ORDER BY`,
    );
    expect(claim.values).toEqual([['c1', 'c2'], 50]);
  });

  it('marks an event DONE when every handler succeeded', async () => {
    handlers = [handler('a.first'), handler('b.second')];
    db.claims.push([row()]);

    await expect(dispatcher.runOnce()).resolves.toBe(1);

    expect(handlers[0].invoke).toHaveBeenCalledTimes(1);
    expect(handlers[1].invoke).toHaveBeenCalledTimes(1);
    expect(db.deliveryDoneStatements().map((query) => query.values)).toEqual([
      ['d-a.first'],
      ['d-b.second'],
    ]);
    const [settle] = db.rootStatements();
    expect(text(settle)).toContain('"processedAt" = now()');
    expect(text(settle)).toContain(
      `WHERE "id" = ? AND "status" = 'PROCESSING' AND "claimedAt" = ?`,
    );
    expect(settle.values).toEqual(['DONE', 'e1', CLAIMED_AT]);
  });

  it('marks an event without handlers DONE', async () => {
    db.claims.push([row({ type: 'VarroaMeasured' })]);

    await dispatcher.runOnce();

    expect(db.prisma.$transaction).not.toHaveBeenCalled();
    expect(db.rootStatements()[0].values).toEqual(['DONE', 'e1', CLAIMED_AT]);
  });

  it('marks the delivery DONE with attempts + 1 in the handler transaction, by the database clock', async () => {
    handlers = [handler('a.first')];
    db.claims.push([row()]);

    await dispatcher.runOnce();

    const [done] = db.deliveryDoneStatements();
    expect(text(done)).toBe(
      `UPDATE "DomainEventDelivery" SET "status" = 'DONE', "attempts" = "attempts" + 1, "completedAt" = now(), "updatedAt" = now() WHERE "id" = ?`,
    );
  });

  it('retries only the failed handler and schedules the event 30 s ahead after the first failure', async () => {
    const ok = handler('a.ok');
    const flaky = handler('b.flaky');
    flaky.invoke.mockRejectedValueOnce(new Error('boom'));
    handlers = [ok, flaky];
    db.claims.push([row({ attempts: 0 })]);

    await dispatcher.runOnce();

    expect(db.deliveryDoneStatements().map((query) => query.values)).toEqual([
      ['d-a.ok'],
    ]);
    expect(db.prisma.domainEventDelivery.upsert).toHaveBeenCalledWith({
      where: { eventId_handler: { eventId: 'e1', handler: 'b.flaky' } },
      create: {
        eventId: 'e1',
        handler: 'b.flaky',
        attempts: 1,
        lastError: 'boom',
      },
      update: { attempts: { increment: 1 }, lastError: 'boom' },
    });
    const [settle] = db.rootStatements();
    expect(text(settle)).toContain(
      `"availableAt" = now() + (? * interval '1 millisecond')`,
    );
    expect(text(settle)).toContain('"attempts" = "attempts" + 1');
    expect(settle.values).toEqual([
      'PENDING',
      'b.flaky: boom',
      30_000,
      'e1',
      CLAIMED_AT,
    ]);
    expect(db.prisma.domainEventDelivery.updateMany).not.toHaveBeenCalled();

    // The next run: a.ok committed its delivery, so only b.flaky runs again.
    db.deliveries.set('a.ok', 'DONE');
    db.claims.push([row({ attempts: 1 })]);
    await dispatcher.runOnce();

    expect(ok.invoke).toHaveBeenCalledTimes(1);
    expect(flaky.invoke).toHaveBeenCalledTimes(2);
    expect(db.rootStatements()[1].values).toEqual(['DONE', 'e1', CLAIMED_AT]);
  });

  it('marks the event and its open deliveries FAILED after the 8th failed attempt', async () => {
    handlers = [handler('a.broken', () => Promise.reject(new Error('boom')))];
    db.claims.push([row({ attempts: 7 })]);

    await dispatcher.runOnce();

    const [settle] = db.rootStatements();
    expect(settle.values).toEqual([
      'FAILED',
      'a.broken: boom',
      'e1',
      CLAIMED_AT,
    ]);
    expect(text(settle)).not.toContain('interval');
    expect(db.prisma.domainEventDelivery.updateMany).toHaveBeenCalledWith({
      where: { eventId: { in: ['e1'] }, status: 'PENDING' },
      data: { status: 'FAILED' },
    });
  });

  it('keeps the first failure as lastError when several handlers fail', async () => {
    handlers = [
      handler('a.first', () => Promise.reject(new Error('one'))),
      handler('b.second', () => Promise.reject(new Error('two'))),
    ];
    db.claims.push([row()]);

    await dispatcher.runOnce();

    expect(db.prisma.domainEventDelivery.upsert).toHaveBeenCalledTimes(2);
    expect(db.rootStatements()[0].values).toContain('a.first: one');
  });

  it('skips a handler whose delivery is already DONE', async () => {
    handlers = [handler('a.done'), handler('b.open')];
    db.deliveries.set('a.done', 'DONE');
    db.claims.push([row()]);

    await dispatcher.runOnce();

    expect(handlers[0].invoke).not.toHaveBeenCalled();
    expect(handlers[1].invoke).toHaveBeenCalledTimes(1);
    expect(db.deliveryDoneStatements().map((query) => query.values)).toEqual([
      ['d-b.open'],
    ]);
    expect(db.rootStatements()[0].values).toEqual(['DONE', 'e1', CLAIMED_AT]);
  });

  it('runs the handler again when its delivery FAILED before (manual retry)', async () => {
    handlers = [handler('a.failed-before')];
    db.deliveries.set('a.failed-before', 'FAILED');
    db.claims.push([row()]);

    await dispatcher.runOnce();

    expect(handlers[0].invoke).toHaveBeenCalledTimes(1);
  });

  it('runs the handler inside the delivery transaction with the event correlation id', async () => {
    const seen: unknown[] = [];
    handlers = [
      handler('a.first', (event, tx) => {
        seen.push(event, tx, domainEventContext.getStore());
        return Promise.resolve();
      }),
    ];
    db.claims.push([row({ correlationId: 'corr-42' })]);

    await dispatcher.runOnce();

    expect(db.prisma.$transaction).toHaveBeenCalledWith(expect.any(Function), {
      timeout: HANDLER_TX_TIMEOUT_MS,
    });
    expect(seen[1]).toBe(db.tx);
    expect(seen[2]).toEqual({ correlationId: 'corr-42' });
    expect(seen[0]).toEqual({
      id: 'e1',
      type: 'TaskDue',
      aggregateType: 'Task',
      aggregateId: 't1',
      payload: { dueOn: '2026-10-13' },
      actorUserId: null,
      correlationId: 'corr-42',
      occurredAt: new Date('2026-10-13T08:00:00.000Z'),
      attempts: 0,
    });
    expect(domainEventContext.getStore()).toBeUndefined();

    // Delivery row first (idempotent), then the row lock, then the handler.
    expect(db.tx.domainEventDelivery.createMany).toHaveBeenCalledWith({
      data: [{ eventId: 'e1', handler: 'a.first' }],
      skipDuplicates: true,
    });
    const [lock] = db.tx.$queryRaw.mock.calls[0];
    expect(text(lock)).toBe(
      'SELECT "id", "status" FROM "DomainEventDelivery" WHERE "eventId" = ? AND "handler" = ? FOR UPDATE',
    );
    expect(lock.values).toEqual(['e1', 'a.first']);
    expect(
      db.tx.domainEventDelivery.createMany.mock.invocationCallOrder[0],
    ).toBeLessThan(db.tx.$queryRaw.mock.invocationCallOrder[0]);
    expect(db.tx.$queryRaw.mock.invocationCallOrder[0]).toBeLessThan(
      (handlers[0].invoke as jest.Mock).mock.invocationCallOrder[0],
    );
  });

  it('advances lastTickAt after every dispatched event, so a long batch never looks stale', async () => {
    const start = new Date('2026-10-13T08:00:00.000Z');
    jest.useFakeTimers({
      now: start,
      doNotFake: [
        'hrtime',
        'nextTick',
        'performance',
        'queueMicrotask',
        'setImmediate',
        'clearImmediate',
        'setInterval',
        'clearInterval',
        'setTimeout',
        'clearTimeout',
      ],
    });
    const seen: (Date | null)[] = [];
    // Every event takes 100 s, e.g. waiting on locks.
    handlers = [
      handler('a.slow', () => {
        seen.push(dispatcher.lastTickAt);
        jest.setSystemTime(Date.now() + 100_000);
        return Promise.resolve();
      }),
    ];
    db.claims.push([row({ id: 'e1' }), row({ id: 'e2' }), row({ id: 'e3' })]);

    try {
      await dispatcher.runOnce();
    } finally {
      jest.useRealTimers();
    }

    expect(seen.map((tick) => tick?.toISOString())).toEqual([
      '2026-10-13T08:00:00.000Z',
      '2026-10-13T08:01:40.000Z',
      '2026-10-13T08:03:20.000Z',
    ]);
    expect(dispatcher.lastTickAt?.toISOString()).toBe(
      '2026-10-13T08:05:00.000Z',
    );
  });

  it('updates lastTickAt even when nothing was claimed', async () => {
    expect(dispatcher.lastTickAt).toBeNull();

    await expect(dispatcher.runOnce()).resolves.toBe(0);

    expect(dispatcher.lastTickAt).toBeInstanceOf(Date);
  });

  it('stores lastError without the Prisma message', async () => {
    handlers = [
      handler('a.db', () =>
        Promise.reject(
          new Prisma.PrismaClientKnownRequestError(
            "Can't reach database server at secret-host:5432",
            { code: 'P1001', clientVersion: '7.9.1' },
          ),
        ),
      ),
    ];
    db.claims.push([row()]);

    await dispatcher.runOnce();

    const upsert = db.upsertArgs();
    expect(upsert.create.lastError).toBe('PrismaClientKnownRequestError P1001');
    expect(upsert.update.lastError).toBe('PrismaClientKnownRequestError P1001');
    const settleValues = db.rootStatements()[0].values;
    expect(settleValues).toContain('a.db: PrismaClientKnownRequestError P1001');
    expect(JSON.stringify(settleValues)).not.toContain('secret-host');
  });

  it('keeps dispatching the batch when one event cannot be settled', async () => {
    db.claims.push([row({ id: 'e1' }), row({ id: 'e2' })]);
    db.prisma.$executeRaw.mockRejectedValueOnce(new Error('connection lost'));

    await expect(dispatcher.runOnce()).resolves.toBe(2);

    expect(db.rootStatements().map((query) => query.values[1])).toEqual([
      'e1',
      'e2',
    ]);
  });

  it('only logs when the claim was lost before settling (stale release)', async () => {
    db.claims.push([row()]);
    db.prisma.$executeRaw.mockResolvedValueOnce(0);

    await expect(dispatcher.runOnce()).resolves.toBe(1);

    expect(logWarn).toHaveBeenCalledWith(expect.stringContaining('e1'));
  });

  describe('drain', () => {
    it('repeats rounds until a round claims nothing and returns the total', async () => {
      db.claims.push(
        [row({ id: 'e1' }), row({ id: 'e2' })],
        [row({ id: 'e3' })],
      );

      await expect(dispatcher.drain({ correlationIds: ['c1'] })).resolves.toBe(
        3,
      );

      expect(db.prisma.$queryRaw).toHaveBeenCalledTimes(3);
      expect(db.claimQuery(2).values).toEqual([['c1'], 50]);
    });

    it('stops after maxRounds even if work remains', async () => {
      for (let i = 0; i < 5; i++) {
        db.claims.push([row({ id: `e${i}` })]);
      }

      await expect(dispatcher.drain(undefined, 2)).resolves.toBe(2);

      expect(db.prisma.$queryRaw).toHaveBeenCalledTimes(2);
    });

    it('makes at most 20 rounds by default', async () => {
      for (let i = 0; i < 25; i++) {
        db.claims.push([row({ id: `e${i}` })]);
      }

      await expect(dispatcher.drain()).resolves.toBe(DRAIN_MAX_ROUNDS);
    });

    // Worker shutdown: the claimed batch is finished (its events would
    // otherwise sit PROCESSING until the stale-claim release), but no
    // further batch is claimed.
    it('finishes the claimed batch but claims no further one once the signal aborts', async () => {
      const stop = new AbortController();
      handlers = [
        handler('a.stopper', () => {
          stop.abort();
          return Promise.resolve();
        }),
      ];
      db.claims.push(
        [row({ id: 'e1' }), row({ id: 'e2' })],
        [row({ id: 'e3' })],
      );

      await expect(
        dispatcher.drain(undefined, DRAIN_MAX_ROUNDS, stop.signal),
      ).resolves.toBe(2);

      expect(handlers[0].invoke).toHaveBeenCalledTimes(2);
      expect(db.prisma.$queryRaw).toHaveBeenCalledTimes(1);
    });

    it('claims nothing once the signal has aborted', async () => {
      const stop = new AbortController();
      stop.abort();
      db.claims.push([row()]);

      await expect(
        dispatcher.drain(undefined, DRAIN_MAX_ROUNDS, stop.signal),
      ).resolves.toBe(0);

      expect(db.prisma.$queryRaw).not.toHaveBeenCalled();
    });
  });

  it('releases stale claims with attempts + 1, FAILED from the 8th attempt on', async () => {
    db.releases.push([
      { id: 'e1', status: 'PENDING' },
      { id: 'e2', status: 'PENDING' },
      { id: 'e3', status: 'PENDING' },
    ]);

    await expect(dispatcher.releaseStaleClaims()).resolves.toBe(3);

    const [release] = db.prisma.$queryRaw.mock.calls[0];
    expect(text(release)).toBe(
      `UPDATE "DomainEvent" SET "status" = CAST(CASE WHEN "attempts" + 1 >= ? THEN 'FAILED' ELSE 'PENDING' END AS "DomainEventStatus"), "attempts" = "attempts" + 1, "claimedAt" = NULL, "availableAt" = now(), "lastError" = ? WHERE "status" = 'PROCESSING' AND "claimedAt" < now() - (? * interval '1 millisecond') RETURNING "id", "status"`,
    );
    expect(release.values).toEqual([
      8,
      'claim expired (worker restarted?)',
      300_000,
    ]);
    expect(db.prisma.domainEventDelivery.updateMany).not.toHaveBeenCalled();
  });

  it('marks the open deliveries FAILED when a stale release ends an event in FAILED', async () => {
    db.releases.push([
      { id: 'e1', status: 'FAILED' },
      { id: 'e2', status: 'PENDING' },
      { id: 'e3', status: 'FAILED' },
    ]);

    await expect(dispatcher.releaseStaleClaims()).resolves.toBe(3);

    expect(db.prisma.domainEventDelivery.updateMany).toHaveBeenCalledTimes(1);
    expect(db.prisma.domainEventDelivery.updateMany).toHaveBeenCalledWith({
      where: { eventId: { in: ['e1', 'e3'] }, status: 'PENDING' },
      data: { status: 'FAILED' },
    });
  });
});