ApiCore/ApiCore-Backend/test/domain-event-dispatcher.e2e-spec.ts

import { Injectable, Logger } from '@nestjs/common';
import { Test, TestingModule } from '@nestjs/testing';
import { Prisma } from '../generated/prisma';
import * as outboxClaims from '../src/common/outbox/outbox-claims';
import { domainEventContext } from '../src/domain-events/domain-event-context';
import {
  DispatchFilter,
  DomainEventDispatcher,
} from '../src/domain-events/domain-event-dispatcher';
import type { DomainEventEnvelope } from '../src/domain-events/domain-event-types';
import { DomainEventsModule } from '../src/domain-events/domain-events.module';
import { DomainEventsService } from '../src/domain-events/domain-events.service';
import { OnDomainEvent } from '../src/domain-events/on-domain-event.decorator';
import { PrismaModule } from '../src/prisma/prisma.module';
import { PrismaService } from '../src/prisma/prisma.service';
import { drainDomainEvents } from './utils/domain-events';

const ENTITY_TYPE = 'DispatcherE2E';

/** `entered` resolves when a handler reaches the gate; `open()` lets it on. */
class Gate {
  private signalEntered!: () => void;
  private release!: () => void;
  readonly entered = new Promise<void>((resolve) => {
    this.signalEntered = resolve;
  });
  private readonly opened = new Promise<void>((resolve) => {
    this.release = resolve;
  });

  pass(): Promise<void> {
    this.signalEntered();
    return this.opened;
  }

  open(): void {
    this.release();
  }
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

/** True if `promise` has not settled within `ms`. */
function isPending(promise: Promise<unknown>, ms = 300): Promise<boolean> {
  return Promise.race([promise.then(() => false), sleep(ms).then(() => true)]);
}

/** Waits until a run reaches `gate`; fails if the run ends without it. */
async function enteredBeforeDone(
  gate: Gate,
  run: Promise<number>,
): Promise<void> {
  await Promise.race([
    gate.entered,
    run.then((claimed) => {
      throw new Error(`Run ended (${claimed} claimed) before the gate`);
    }),
  ]);
}

/**
 * TaskDue, QueenAged and InspectionOverdue have no real handlers in the
 * plan, so these test handlers are the only ones that react - and only to
 * this file's events, because every dispatch below is filtered by
 * correlation id.
 */
@Injectable()
class DispatcherTestHandlers {
  readonly seen: { event: DomainEventEnvelope; correlationId?: string }[] = [];

  @OnDomainEvent('TaskDue', 'test.counter')
  async count(
    event: DomainEventEnvelope,
    tx: Prisma.TransactionClient,
  ): Promise<void> {
    this.seen.push({
      event,
      correlationId: domainEventContext.getStore()?.correlationId,
    });
    await tx.auditLog.create({
      data: {
        entityType: ENTITY_TYPE,
        entityId: event.aggregateId,
        action: 'CREATE',
      },
    });
  }

  @OnDomainEvent('QueenAged', 'test.failing')
  fail(): Promise<void> {
    return Promise.reject(new Error('boom'));
  }

  /** Aggregates this instance ran 'test.gated' for, in order. */
  readonly gatedRuns: string[] = [];
  private readonly gates = new Map<string, Gate>();

  /** Holds 'test.gated' for one aggregate inside its delivery transaction. */
  hold(aggregateId: string): Gate {
    const gate = new Gate();
    this.gates.set(aggregateId, gate);
    return gate;
  }

  @OnDomainEvent('InspectionOverdue', 'test.gated')
  async gated(
    event: DomainEventEnvelope,
    tx: Prisma.TransactionClient,
  ): Promise<void> {
    this.gatedRuns.push(event.aggregateId);
    await this.gates.get(event.aggregateId)?.pass();
    await tx.auditLog.create({
      data: {
        entityType: ENTITY_TYPE,
        entityId: event.aggregateId,
        action: 'CREATE',
      },
    });
  }
}

async function createDispatcherContext(): Promise<TestingModule> {
  const moduleRef = await Test.createTestingModule({
    imports: [PrismaModule, DomainEventsModule],
    providers: [DispatcherTestHandlers],
  }).compile();
  await moduleRef.init();
  return moduleRef;
}

describe('DomainEventDispatcher (e2e)', () => {
  const suffix = `dispatch-${Date.now()}-${Math.round(Math.random() * 1e6)}`;
  let appA: TestingModule;
  let appB: TestingModule;
  let prisma: PrismaService;

  /** Records n TaskDue events in one transaction; returns their filter. */
  async function recordTaskDue(
    n: number,
    group: string,
  ): Promise<DispatchFilter> {
    const events = appA.get(DomainEventsService);
    await prisma.$transaction((tx) =>
      events.recordAll(
        tx,
        Array.from({ length: n }, (_, i) => ({
          type: 'TaskDue' as const,
          aggregateType: 'Task' as const,
          aggregateId: `${suffix}-${group}-${i}`,
          payload: { dueOn: '2026-10-13' },
        })),
      ),
    );
    return filterFor(group);
  }

  async function filterFor(group: string): Promise<DispatchFilter> {
    const rows = await prisma.domainEvent.findMany({
      where: { aggregateId: { startsWith: `${suffix}-${group}-` } },
      select: { correlationId: true },
      distinct: ['correlationId'],
    });
    return { correlationIds: rows.map((r) => r.correlationId) };
  }

  async function waitForStatus(id: string, status: string): Promise<void> {
    for (let i = 0; i < 250; i++) {
      const event = await prisma.domainEvent.findUnique({
        where: { id },
        select: { status: true },
      });
      if (event?.status === status) {
        return;
      }
      await sleep(20);
    }
    throw new Error(`Domain event ${id} never reached ${status}`);
  }

  function effects(prefix: string): Promise<number> {
    return prisma.auditLog.count({
      where: { entityType: ENTITY_TYPE, entityId: { startsWith: prefix } },
    });
  }

  beforeAll(async () => {
    [appA, appB] = await Promise.all([
      createDispatcherContext(),
      createDispatcherContext(),
    ]);
    prisma = appA.get(PrismaService);
  });

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

  afterAll(async () => {
    await prisma.auditLog.deleteMany({
      where: { entityType: ENTITY_TYPE, entityId: { startsWith: suffix } },
    });
    await prisma.domainEvent.deleteMany({
      where: { aggregateId: { startsWith: suffix } },
    });
    await Promise.all([appA.close(), appB.close()]);
  });

  it('two dispatchers in parallel produce exactly one effect per event and handler', async () => {
    const filter = await recordTaskDue(20, 'parallel');

    const [a, b] = await Promise.all([
      appA.get(DomainEventDispatcher).drain(filter),
      appB.get(DomainEventDispatcher).drain(filter),
    ]);

    expect(a + b).toBe(20);
    expect(await effects(`${suffix}-parallel-`)).toBe(20);
    const events = await prisma.domainEvent.findMany({
      where: { aggregateId: { startsWith: `${suffix}-parallel-` } },
      include: { deliveries: true },
    });
    expect(events).toHaveLength(20);
    for (const event of events) {
      expect(event).toMatchObject({
        status: 'DONE',
        attempts: 0,
        claimedAt: null,
        lastError: null,
      });
      expect(event.processedAt).toBeInstanceOf(Date);
      expect(event.deliveries).toEqual([
        expect.objectContaining({
          handler: 'test.counter',
          status: 'DONE',
          attempts: 1,
        }),
      ]);
    }
  });

  it('hands the handler the stored event and its correlation id', async () => {
    const filter = await recordTaskDue(1, 'envelope');
    const handlers = appA.get(DispatcherTestHandlers);

    await appA.get(DomainEventDispatcher).drain(filter);

    const seen = handlers.seen.find(
      (entry) => entry.event.aggregateId === `${suffix}-envelope-0`,
    );
    expect(seen?.correlationId).toBe(filter.correlationIds?.[0]);
    expect(seen?.event).toMatchObject({
      type: 'TaskDue',
      aggregateType: 'Task',
      payload: { dueOn: '2026-10-13' },
      actorUserId: null,
      correlationId: filter.correlationIds?.[0],
      attempts: 0,
    });
    expect(seen?.event.occurredAt).toBeInstanceOf(Date);
  });

  it('skips an event another worker holds locked instead of waiting for it', async () => {
    const filter = await recordTaskDue(2, 'locked');
    const { id } = await prisma.domainEvent.findFirstOrThrow({
      where: { aggregateId: `${suffix}-locked-0` },
    });
    const lock = new Gate();
    // Another worker in the middle of its claim statement.
    const holder = appB.get(PrismaService).$transaction(
      async (tx) => {
        await tx.$queryRaw`SELECT "id" FROM "DomainEvent" WHERE "id" = ${id} FOR UPDATE`;
        await lock.pass();
      },
      { timeout: 15_000 },
    );
    await lock.entered;
    const dispatcher = appA.get(DomainEventDispatcher);

    try {
      const run = dispatcher.runOnce(filter);
      // Generous window against slow CI; without SKIP LOCKED the claim
      // blocks until the lock opens, so it would still be pending here.
      expect(await isPending(run, 4_000)).toBe(false);
      await expect(run).resolves.toBe(1);
      expect(await effects(`${suffix}-locked-1`)).toBe(1);
      expect(await effects(`${suffix}-locked-0`)).toBe(0);
    } finally {
      lock.open();
      await holder;
    }
    await expect(dispatcher.runOnce(filter)).resolves.toBe(1);
    expect(await effects(`${suffix}-locked-0`)).toBe(1);
  }, 20_000);

  it('a stale release that ends in FAILED also fails the open deliveries', async () => {
    const filter = await recordTaskDue(1, 'stale-failed');
    const { id } = await prisma.domainEvent.findFirstOrThrow({
      where: { aggregateId: `${suffix}-stale-failed-0` },
    });
    // Seven attempts used up, the worker of the eighth died mid-event.
    await prisma.$executeRaw`UPDATE "DomainEvent" SET "status" = 'PROCESSING', "attempts" = 7, "claimedAt" = now() - interval '6 minutes' WHERE "id" = ${id}`;
    await prisma.domainEventDelivery.create({
      data: { eventId: id, handler: 'test.counter', attempts: 7 },
    });

    await expect(
      appA.get(DomainEventDispatcher).releaseStaleClaims(),
    ).resolves.toBeGreaterThanOrEqual(1);

    expect(
      await prisma.domainEvent.findUnique({
        where: { id },
        include: { deliveries: true },
      }),
    ).toMatchObject({
      status: 'FAILED',
      attempts: 8,
      claimedAt: null,
      lastError: 'claim expired (worker restarted?)',
      deliveries: [{ handler: 'test.counter', status: 'FAILED', attempts: 7 }],
    });
    await expect(appA.get(DomainEventDispatcher).drain(filter)).resolves.toBe(
      0,
    );
  });

  it('drains a backlog of 120 events in batches of at most 50', async () => {
    const filter = await recordTaskDue(120, 'backlog');
    const dispatcher = appA.get(DomainEventDispatcher);

    const rounds: number[] = [];
    for (let i = 0; i < 4; i++) {
      rounds.push(await dispatcher.runOnce(filter));
    }

    expect(rounds).toEqual([50, 50, 20, 0]);
    expect(await effects(`${suffix}-backlog-`)).toBe(120);
  });

  it('drainDomainEvents processes only the events of the given root aggregates', async () => {
    await recordTaskDue(2, 'root');
    await recordTaskDue(1, 'other');

    await drainDomainEvents(appA, [`${suffix}-root-0`]);

    expect(await effects(`${suffix}-root-`)).toBe(2);
    expect(await effects(`${suffix}-other-`)).toBe(0);
    await drainDomainEvents(appA, [`${suffix}-other-0`]);
    expect(await effects(`${suffix}-other-`)).toBe(1);
  });

  it('a stale-released event does not repeat a handler that already committed', async () => {
    const filter = await recordTaskDue(1, 'stale');
    const dispatcher = appA.get(DomainEventDispatcher);
    // The worker dies after the handler committed, before settling the event.
    jest
      .spyOn(outboxClaims, 'settleSuccess')
      .mockRejectedValueOnce(new Error('worker killed'));
    const logError = jest
      .spyOn(Logger.prototype, 'error')
      .mockImplementation(() => undefined);

    await dispatcher.runOnce(filter);
    expect(logError).toHaveBeenCalledWith(
      expect.stringContaining('not settled: worker killed'),
    );

    const id = (
      await prisma.domainEvent.findFirstOrThrow({
        where: { aggregateId: `${suffix}-stale-0` },
      })
    ).id;
    expect(
      await prisma.domainEvent.findUnique({ where: { id } }),
    ).toMatchObject({ status: 'PROCESSING' });
    expect(await effects(`${suffix}-stale-`)).toBe(1);

    await prisma.$executeRaw`UPDATE "DomainEvent" SET "claimedAt" = now() - interval '6 minutes' WHERE "id" = ${id}`;
    await expect(
      dispatcher.releaseStaleClaims(),
    ).resolves.toBeGreaterThanOrEqual(1);
    expect(
      await prisma.domainEvent.findUnique({ where: { id } }),
    ).toMatchObject({
      status: 'PENDING',
      attempts: 1,
      claimedAt: null,
      lastError: 'claim expired (worker restarted?)',
    });

    await appB.get(DomainEventDispatcher).drain(filter);

    expect(await effects(`${suffix}-stale-`)).toBe(1);
    expect(
      await prisma.domainEvent.findUnique({
        where: { id },
        include: { deliveries: true },
      }),
    ).toMatchObject({
      status: 'DONE',
      attempts: 1,
      deliveries: [{ handler: 'test.counter', status: 'DONE', attempts: 1 }],
    });
  });

  it('a worker whose claim expired waits for the delivery lock and skips the handler the new owner committed', async () => {
    const events = appA.get(DomainEventsService);
    for (const i of [0, 1]) {
      await prisma.$transaction((tx) =>
        events.record(tx, {
          type: 'InspectionOverdue',
          aggregateType: 'Colony',
          aggregateId: `${suffix}-gated-${i}`,
          payload: { dueOn: '2026-10-13', lastInspectedOn: null },
        }),
      );
    }
    const filter = await filterFor('gated');
    const [e0, e1] = await Promise.all(
      [0, 1].map((i) =>
        prisma.domainEvent.findFirstOrThrow({
          where: { aggregateId: `${suffix}-gated-${i}` },
        }),
      ),
    );
    // Worker A handles e0 before e1.
    await prisma.$executeRaw`UPDATE "DomainEvent" SET "occurredAt" = "occurredAt" - interval '1 second' WHERE "id" = ${e0.id}`;
    // e1's handler failed once before, so its delivery row exists: only
    // the row lock (SELECT ... FOR UPDATE), not the unique insert, can
    // make a second worker wait.
    await prisma.domainEventDelivery.create({
      data: {
        eventId: e1.id,
        handler: 'test.gated',
        attempts: 1,
        lastError: 'earlier failure',
      },
    });
    const handlersA = appA.get(DispatcherTestHandlers);
    const handlersB = appB.get(DispatcherTestHandlers);
    const gateA = handlersA.hold(e0.aggregateId);
    const gateB = handlersB.hold(e1.aggregateId);
    const logWarn = jest
      .spyOn(Logger.prototype, 'warn')
      .mockImplementation(() => undefined);

    const runs: Promise<number>[] = [];
    try {
      // A claims both events and is held inside the handler for e0 ...
      const runA = appA.get(DomainEventDispatcher).runOnce(filter);
      runs.push(runA);
      await enteredBeforeDone(gateA, runA);
      // ... while its claim on e1 expires: B claims e1 and is held inside
      // e1's delivery transaction, the delivery row locked.
      await prisma.$executeRaw`UPDATE "DomainEvent" SET "claimedAt" = now() - interval '6 minutes' WHERE "id" = ${e1.id}`;
      await appB.get(DomainEventDispatcher).releaseStaleClaims();
      const runB = appB.get(DomainEventDispatcher).runOnce(filter);
      runs.push(runB);
      await enteredBeforeDone(gateB, runB);

      // A finishes e0, moves on to e1 and has to wait for B's delivery row.
      gateA.open();
      await waitForStatus(e0.id, 'DONE');
      expect(await isPending(runA)).toBe(true);

      // B commits; A then finds e1's delivery DONE and skips the handler.
      gateB.open();
      await expect(runB).resolves.toBe(1);
      await expect(runA).resolves.toBe(2);
    } finally {
      gateA.open();
      gateB.open();
      await Promise.allSettled(runs);
    }

    expect(handlersA.gatedRuns).toEqual([e0.aggregateId]);
    expect(handlersB.gatedRuns).toEqual([e1.aggregateId]);
    expect(await effects(`${suffix}-gated-`)).toBe(2);
    const delivered = (attempts: number) => ({
      deliveries: [{ handler: 'test.gated', status: 'DONE', attempts }],
    });
    expect(
      await prisma.domainEvent.findMany({
        where: { id: { in: [e0.id, e1.id] } },
        include: { deliveries: true },
        orderBy: { aggregateId: 'asc' },
      }),
    ).toMatchObject([
      { id: e0.id, status: 'DONE', attempts: 0, ...delivered(1) },
      { id: e1.id, status: 'DONE', attempts: 1, ...delivered(2) },
    ]);
    // A's claim on e1 was gone, so B - the owner - settled it.
    expect(logWarn).toHaveBeenCalledWith(
      expect.stringContaining(`${e1.id} (InspectionOverdue): claim lost`),
    );
  });

  it('schedules a failed event about 30 s ahead by the database clock', async () => {
    const events = appA.get(DomainEventsService);
    await prisma.$transaction((tx) =>
      events.record(tx, {
        type: 'QueenAged',
        aggregateType: 'Queen',
        aggregateId: `${suffix}-failing-0`,
        payload: { hatchYear: 2024, colonyId: null },
      }),
    );
    const filter = await filterFor('failing');

    await expect(appA.get(DomainEventDispatcher).drain(filter)).resolves.toBe(
      1,
    );

    const event = await prisma.domainEvent.findFirstOrThrow({
      where: { aggregateId: `${suffix}-failing-0` },
      include: { deliveries: true },
    });
    expect(event).toMatchObject({
      status: 'PENDING',
      attempts: 1,
      claimedAt: null,
      lastError: 'test.failing: boom',
      deliveries: [
        {
          handler: 'test.failing',
          status: 'PENDING',
          attempts: 1,
          lastError: 'boom',
        },
      ],
    });
    const [{ s }] = await prisma.$queryRaw<
      { s: number }[]
    >`SELECT EXTRACT(EPOCH FROM "availableAt" - now())::float AS s FROM "DomainEvent" WHERE id = ${event.id}`;
    expect(s).toBeGreaterThan(25);
    expect(s).toBeLessThanOrEqual(30.5);
  });
});