ApiCore/ApiCore-Backend/test/worker-runtime.e2e-spec.ts

import { randomUUID } from 'node:crypto';
import * as http from 'node:http';
import { AddressInfo } from 'node:net';
import { Injectable } from '@nestjs/common';
import { Test, TestingModule } from '@nestjs/testing';
import { Prisma } from '../generated/prisma';
import { CalendarSyncService } from '../src/calendar-sync/calendar-sync.service';
import { notifyWorker, WAKE_CHANNELS } from '../src/common/worker-wakeup';
import { domainEventContext } from '../src/domain-events/domain-event-context';
import {
  DispatchFilter,
  DomainEventDispatcher,
} from '../src/domain-events/domain-event-dispatcher';
import { DomainEventRegistry } from '../src/domain-events/domain-event-registry';
import type { DomainEventEnvelope } from '../src/domain-events/domain-event-types';
import { DomainEventsService } from '../src/domain-events/domain-events.service';
import { OnDomainEvent } from '../src/domain-events/on-domain-event.decorator';
import { PrismaService } from '../src/prisma/prisma.service';
import { WEATHER_HISTORY_SOURCE } from '../src/weather/history/weather-history-source';
import { WebhookDeliveryWorker } from '../src/webhooks/webhook-delivery.worker';
import { WorkerModule } from '../src/worker/worker.module';
import { DOMAIN_EVENT_POLL_MS } from '../src/worker/worker-runtime.service';

const ENTITY_TYPE = 'WorkerRuntimeE2E';
/** Every event of this file carries it; the worker below sees no other. */
const CORRELATION_ID = randomUUID();
/** Well below the 5 s poll: only a NOTIFY wake-up can be this fast. */
const WAKE_UP_WITHIN_MS = 1_500;

@Injectable()
class CountingHandler {
  @OnDomainEvent('TaskDue', 'test.worker-counter')
  async count(
    event: DomainEventEnvelope,
    tx: Prisma.TransactionClient,
  ): Promise<void> {
    await tx.auditLog.create({
      data: {
        entityType: ENTITY_TYPE,
        entityId: event.aggregateId,
        action: 'CREATE',
      },
    });
  }
}

/**
 * The real dispatcher, limited to this file's correlation: e2e files run in
 * parallel against one database, and a worker draining all of it would
 * take over (and release the stale claims of) their events.
 */
@Injectable()
class IsolatedDispatcher extends DomainEventDispatcher {
  override drain(
    _filter?: DispatchFilter,
    maxRounds?: number,
    signal?: AbortSignal,
  ): Promise<number> {
    return super.drain({ correlationIds: [CORRELATION_ID] }, maxRounds, signal);
  }

  override releaseStaleClaims(): Promise<number> {
    return Promise.resolve(0);
  }
}

/** Assigning undefined would store the string "undefined". */
function restoreEnv(name: string, value: string | undefined): void {
  if (value === undefined) {
    delete process.env[name];
  } else {
    process.env[name] = value;
  }
}

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

async function freePort(): Promise<number> {
  const server = http.createServer();
  await new Promise<void>((resolve) => server.listen(0, resolve));
  const { port } = server.address() as AddressInfo;
  await new Promise((resolve) => server.close(resolve));
  return port;
}

function getJson(
  port: number,
  path: string,
): Promise<{ status: number; body: unknown }> {
  return new Promise((resolve, reject) => {
    http
      .get({ port, path }, (res) => {
        let raw = '';
        res.on('data', (chunk: Buffer) => (raw += chunk.toString()));
        res.on('end', () =>
          resolve({
            status: res.statusCode ?? 0,
            body: raw ? (JSON.parse(raw) as unknown) : undefined,
          }),
        );
      })
      .on('error', reject);
  });
}

/** Polls `read` until it returns `expected`, for at most `ms`. */
async function expectWithin<T>(
  ms: number,
  read: () => Promise<T> | T,
  expected: T,
): Promise<void> {
  const deadline = Date.now() + ms;
  let actual = await read();
  while (actual !== expected && Date.now() < deadline) {
    await sleep(25);
    actual = await read();
  }
  expect(actual).toBe(expected);
}

/**
 * The worker process as dist/worker.js starts it (WorkerModule, LISTEN,
 * loops, health endpoint) against the test database.
 */
describe('Worker runtime (e2e)', () => {
  const suffix = `worker-${Date.now()}-${Math.round(Math.random() * 1e6)}`;
  const originalHealthPort = process.env.WORKER_HEALTH_PORT;
  let worker: TestingModule;
  let prisma: PrismaService;
  let processOutbox: jest.SpyInstance;
  let deliverWebhooks: jest.SpyInstance;
  let healthPort: number;

  async function recordTaskDueInTx(): Promise<void> {
    const events = worker.get(DomainEventsService);
    await domainEventContext.run({ correlationId: CORRELATION_ID }, () =>
      prisma.$transaction((tx) =>
        events.record(tx, {
          type: 'TaskDue',
          aggregateType: 'Task',
          aggregateId: `${suffix}-0`,
          payload: { dueOn: '2026-10-13' },
        }),
      ),
    );
  }

  const countAuditRows = () =>
    prisma.auditLog.count({
      where: { entityType: ENTITY_TYPE, entityId: { startsWith: suffix } },
    });

  beforeAll(async () => {
    healthPort = await freePort();
    process.env.WORKER_HEALTH_PORT = String(healthPort);
    worker = await Test.createTestingModule({
      imports: [WorkerModule],
      providers: [CountingHandler],
    })
      .overrideProvider(DomainEventDispatcher)
      .useClass(IsolatedDispatcher)
      // The phenology-weather loop ticks right after bootstrap: no real
      // network call in tests (weather-history.e2e-spec.ts covers the sync).
      .overrideProvider(WEATHER_HISTORY_SOURCE)
      .useValue({ name: 'test', dailyMeans: () => Promise.resolve([]) })
      .compile();
    prisma = worker.get(PrismaService);
    // The outbox itself is covered by calendar-sync.e2e-spec.ts; here only
    // whether the loop runs and wakes up counts.
    processOutbox = jest
      .spyOn(worker.get(CalendarSyncService), 'processOutbox')
      .mockResolvedValue(undefined);
    // Likewise the webhook loop (webhooks.e2e-spec.ts covers delivery): an
    // unfiltered drain here would send the deliveries of parallel e2e files.
    const webhookDeliveries = worker.get(WebhookDeliveryWorker);
    deliverWebhooks = jest
      .spyOn(webhookDeliveries, 'drain')
      .mockResolvedValue(0);
    jest.spyOn(webhookDeliveries, 'releaseStaleClaims').mockResolvedValue(0);
    await worker.init();
    // The first ticks run right after bootstrap; wait for the dispatcher's.
    const dispatcher = worker.get(DomainEventDispatcher);
    await expectWithin(5_000, () => dispatcher.lastTickAt !== null, true);
  }, 15_000);

  afterAll(async () => {
    await prisma.auditLog.deleteMany({
      where: { entityType: ENTITY_TYPE, entityId: { startsWith: suffix } },
    });
    await prisma.domainEvent.deleteMany({
      where: { aggregateId: { startsWith: suffix } },
    });
    await worker.close();
    restoreEnv('WORKER_HEALTH_PORT', originalHealthPort);
  });

  it('wakes up on NOTIFY and processes a new event well before the 5 s poll', async () => {
    expect(WAKE_UP_WITHIN_MS).toBeLessThan(DOMAIN_EVENT_POLL_MS);

    await recordTaskDueInTx();

    await expectWithin(WAKE_UP_WITHIN_MS, countAuditRows, 1);
    const event = await prisma.domainEvent.findFirstOrThrow({
      where: { aggregateId: `${suffix}-0` },
      include: { deliveries: true },
    });
    // TaskDue also has the real reminder handler (the task does not exist: a no-op).
    expect(event.status).toBe('DONE');
    expect(
      event.deliveries.map(({ handler, status }) => ({ handler, status })),
    ).toEqual(
      expect.arrayContaining([
        { handler: 'test.worker-counter', status: 'DONE' },
        { handler: 'notifications.task-due', status: 'DONE' },
      ]),
    );
  });

  it('registers the compliance handlers in the worker context too', () => {
    expect(worker.get(DomainEventRegistry).handlerIds()).toEqual(
      expect.arrayContaining([
        'compliance.herd-change',
        'compliance.tsk-reference-day',
      ]),
    );
  });

  it('wakes the calendar outbox loop on a calendar_sync notification', async () => {
    expect(processOutbox).toHaveBeenCalled();
    const before = processOutbox.mock.calls.length;

    await notifyWorker(prisma, WAKE_CHANNELS.calendarSync);

    await expectWithin(
      WAKE_UP_WITHIN_MS,
      () => processOutbox.mock.calls.length > before,
      true,
    );
  });

  it('wakes the webhook delivery loop on a webhook_delivery notification', async () => {
    expect(deliverWebhooks).toHaveBeenCalled();
    const before = deliverWebhooks.mock.calls.length;

    await notifyWorker(prisma, WAKE_CHANNELS.webhookDeliveries);

    await expectWithin(
      WAKE_UP_WITHIN_MS,
      () => deliverWebhooks.mock.calls.length > before,
      true,
    );
  });

  it('answers GET /health with 200 {"status":"ok"}', async () => {
    await expect(getJson(healthPort, '/health')).resolves.toEqual({
      status: 200,
      body: { status: 'ok' },
    });
  });
});