Events

const config: EventIngestionConfig = {
  topics: DEFAULT_INGESTION_TOPICS, // { process: "webhook.process", sweepPending: "webhook.sweep-pending" }
  retryLimit: 5,
  retryDelayMs: 30_000,
  retryBackoff: "exponential",
  sweepStaleMinutes: 10,
  sweepCronExpression: "*/5 * * * *",
  gracefulShutdownTimeoutMs: 10_000,
  replayBatchSize: 100,
};

// HTTP side: the verifier is required, the request is the raw webhook delivery.
export async function handleWebhook(
  verifier: WebhookVerifier<Provider>,
  repository: IngestedEventsRepository,
  queue: Queue,
  request: Request
) {
  const ingestion = createEventIngestionService({
    verifier,
    repository,
    queue,
    registry,
    config,
  });
  const result = await ingestion.ingest({ provider: "billing", request });
  // => { eventId, duplicate: false } on first delivery
  // => { eventId, duplicate: true } on a repeat delivery, nothing re-enqueued
  return new Response(JSON.stringify(result), { status: 202 });
}

// Worker side: no verifier; `process` claims, handles and marks the row.
export async function startWorker(repository: IngestedEventsRepository, queue: Queue) {
  const ingestion = createEventIngestionService({ repository, queue, registry, config });
  await queue.work<{ id: string }>(
    config.topics.process,
    (msg) => ingestion.process(msg.data.id), // re-throws after markFailed so the queue retries
    {
      maxAttempts: config.retryLimit,
      initialDelayMs: config.retryDelayMs,
      backoff: config.retryBackoff,
    }
  );
}

export async function replay(ingestion: EventIngestionService, eventId: string) {
  try {
    await ingestion.replayOne(eventId);
  } catch (error) {
    if (error instanceof InvalidReplayStateError) {
      error.currentStatus; // => "pending" | "processing" | "done" | "skipped"
      error.code; // => "INGESTED_EVENT_INVALID_REPLAY_STATE"
    } else if (isAppError(error)) {
      error.code; // => "INGESTED_EVENT_NOT_FOUND"
    }
    throw error;
  }
  // Bulk variant, filtered and paged by `config.replayBatchSize`:
  return ingestion.replayFailed({ provider: "billing", status: ["failed", "exhausted"] });
  // => { enqueued, skipped }
}

type Provider = "billing";

// Idempotent: replaying the same payload converges on the same state.
declare function applyInvoicePaid(payload: unknown, eventId: string): Promise<void>;

const invoicePaid: IngestedEventHandler<Provider> = {
  provider: "billing",
  eventType: "invoice.paid",
  handle: (payload, ctx) => applyInvoicePaid(payload, ctx.eventId),
};

const registry = createIngestedEventHandlerRegistry<Provider>([invoicePaid]);
registry.get("billing", "invoice.paid"); // => invoicePaid
registry.get("billing", "invoice.voided"); // => undefined: the event will be stored as `skipped`
registry.hasHandlers(); // => true

export async function workerLifecycleInTests() {
  const repository = createFakeIngestedEventsRepository();
  const queue = createFakeQueue({ mode: "manual" }); // `send` records, `drain` runs the handlers
  const ingestion = createEventIngestionService({ repository, queue, registry, config });

  const created = await repository.create({
    provider: "billing",
    externalId: "evt_1",
    eventType: "invoice.paid",
    occurredAt: new Date(),
    payload: {},
  });
  // created => status "pending", attempts 0

  await ingestion.process(created?.id ?? "");
  // claim: pending → processing (attempts 1), then markDone: → done
  await ingestion.process(created?.id ?? ""); // claim lost: returns without running the handler

  return repository.statsByStatus(); // => { pending: 0, processing: 0, done: 1, failed: 0, exhausted: 0, skipped: 0 }
}