Merge pull request #156 from Ezeh20/feat/Add-Outbox-Event-Relay-and-Handler-Registry

Feat/add outbox event relay and handler registry

Author: Kaylahray

.env.example

@@ -39,6 +39,13 @@ REDIS_URL=redis://localhost:6379
 # Wallet-provisioning worker poll interval (ms)
 WORKER_POLL_INTERVAL_MS=5000
 
+SCHEDULER_INTERVAL_MS=15000
+SCHEDULER_LEASE_MS=60000
+SCHEDULER_SHUTDOWN_TIMEOUT_MS=30000
+SCHEDULER_QUEUES=
+SCHEDULER_DISABLED_QUEUES=
+SCHEDULER_IN_PROCESS=false
+
 # Logging Configuration
 # LOG_LEVEL=info (options: error, warn, info, http, verbose, debug, silly)
 
@@ -55,17 +62,16 @@ STELLAR_FUNDING_MAX_RETRIES=5
 # Account Lifecycle Configuration
 DELETION_COOLING_OFF_DAYS=30
 EXPORT_TTL_DAYS=7
-# Interval for the background lifecycle sweep (export generation, deletion finalization). 0 = disabled (lazy sweep on requests only)
 LIFECYCLE_SWEEP_INTERVAL_MS=0
 
 # Phone OTP Configuration
 # Only "mock" is implemented until a real carrier (Twilio, Termii, etc.) is integrated — see docs/decisions/0001-phone-otp-authentication.md
 SMS_PROVIDER=mock
 RATE_LIMIT_OTP_WINDOW_MS=900000
 RATE_LIMIT_OTP_MAX=5
-
-# Data Lifecycle / Audit Configuration — see docs/DATA_LIFECYCLE.md
-# HMAC key for the source-IP hash on immutable audit events. Rotating it makes
-# older hashes uncorrelatable with newer ones. If unset in production, audit
-# events omit the IP hash rather than storing an unkeyed (reversible) digest.
-# AUDIT_IP_HASH_SECRET=change-me-in-production
+
+# Data Lifecycle / Audit Configuration — see docs/DATA_LIFECYCLE.md
+# HMAC key for the source-IP hash on immutable audit events. Rotating it makes
+# older hashes uncorrelatable with newer ones. If unset in production, audit
+# events omit the IP hash rather than storing an unkeyed (reversible) digest.
+# AUDIT_IP_HASH_SECRET=change-me-in-production

Dockerfile

@@ -85,8 +85,9 @@ COPY --from=build /app/prisma.config.ts ./
 # Entrypoint scripts
 COPY docker/entrypoint-api.sh  ./entrypoint-api.sh
 COPY docker/entrypoint-worker.sh ./entrypoint-worker.sh
+COPY docker/entrypoint-scheduler.sh ./entrypoint-scheduler.sh
 
-RUN chmod +x entrypoint-api.sh entrypoint-worker.sh
+RUN chmod +x entrypoint-api.sh entrypoint-worker.sh entrypoint-scheduler.sh
 
 # Non-root user
 RUN groupadd --gid 1001 appgroup && \

docker-compose.yml

@@ -1,12 +1,11 @@
 # ============================================================================
 # Learnault API — Local Development Stack
 #
-# One command starts a healthy stack (API + wallet worker + PostgreSQL + Redis):
+# One command starts a healthy stack (API + PostgreSQL + Redis):
 #   docker compose up -d --build
 #
 # The API container applies migrations and seeds deterministic fixtures on
-# boot (see docker/entrypoint-dev-api.sh). The worker drains the idempotent
-# wallet-provisioning outbox (see src/workers/wallet-provisioning.worker.ts).
+# boot (see docker/entrypoint-dev-api.sh).
 #
 # Useful commands:
 #   docker compose ps            → service status + health
@@ -95,10 +94,7 @@ services:
       start_period: 20s
     stop_grace_period: 30s
 
-  # --------------------------------------------------------------------------
-  # Worker — drains the wallet-provisioning outbox
-  # --------------------------------------------------------------------------
-  worker:
+  scheduler:
     build:
       context: .
       dockerfile: docker/Dockerfile.dev
@@ -110,12 +106,17 @@ services:
       NODE_ENV: development
       DATABASE_URL: postgresql://${POSTGRES_USER:-learnault}:${POSTGRES_PASSWORD:-learnault}@db:5432/${POSTGRES_DB:-learnault_dev}?schema=public
       RUN_MIGRATIONS: 'true'
-      WORKER_POLL_INTERVAL_MS: ${WORKER_POLL_INTERVAL_MS:-5000}
+      LOG_LEVEL: ${LOG_LEVEL:-info}
+      SCHEDULER_INTERVAL_MS: ${SCHEDULER_INTERVAL_MS:-15000}
+      SCHEDULER_LEASE_MS: ${SCHEDULER_LEASE_MS:-60000}
+      SCHEDULER_SHUTDOWN_TIMEOUT_MS: ${SCHEDULER_SHUTDOWN_TIMEOUT_MS:-30000}
+      SCHEDULER_QUEUES: ${SCHEDULER_QUEUES:-}
+      SCHEDULER_DISABLED_QUEUES: ${SCHEDULER_DISABLED_QUEUES:-}
     volumes:
       - .:/app
       - /app/node_modules
-    command: ['./docker/entrypoint-dev-worker.sh']
-    stop_grace_period: 30s
+    command: ['./docker/entrypoint-dev-scheduler.sh']
+    stop_grace_period: 40s
 
 volumes:
   pgdata:

docker/entrypoint-dev-scheduler.sh

@@ -0,0 +1,11 @@
+#!/bin/sh
+set -e
+
+if [ "${RUN_MIGRATIONS:-true}" = "true" ]; then
+  echo "[entrypoint] Applying database migrations …"
+  npx prisma migrate deploy
+  echo "[entrypoint] Migrations applied."
+fi
+
+echo "[entrypoint] Starting scheduled job runner …"
+exec pnpm scheduler:dev

docker/entrypoint-dev-worker.sh

@@ -0,0 +1,11 @@
+#!/bin/sh
+set -e
+
+if [ "${RUN_MIGRATIONS}" = "true" ]; then
+  echo "[entrypoint] Running database migrations …"
+  npx prisma migrate deploy
+  echo "[entrypoint] Migrations applied."
+fi
+
+echo "[entrypoint] Starting scheduled job runner …"
+exec node dist/workers/scheduler.worker.js

docker/entrypoint-scheduler.sh

@@ -154,12 +154,15 @@ rather than removing it.
 | `OutboxEvent` | MUTABLE | 30d (`createdAt`) | Retain | No |
 | `JobAttempt` | MUTABLE | 30d (`createdAt`) | Cascade | No |
 | `RolledBackRecord` | IMMUTABLE | 30d (`createdAt`) | Retain | No |
+| `QueueLease` | MUTABLE | Indefinite | Retain | No |
 | `WalletProvisioningJob` | MUTABLE | 90d (`updatedAt`) | Cascade | No |
 
 `EmailDelivery` and `NotificationLog` hold rendered message bodies, which is
 personal data — hence the short window and hard deletion on erasure.
 `DeviceToken` is deleted rather than archived: an archived push token would still
-be a live address.
+be a live address. `QueueLease` holds one long-lived row per recurring queue
+drain — a queue name, the current lease token, and the holder id — so there is
+no user data to erase and nothing to age out.
 
 ---
 

docs/DATA_LIFECYCLE.md

@@ -1,6 +1,6 @@
 # Local Development Stack (Docker Compose)
 
-A reproducible local stack for the Learnault API: **API**, **wallet worker**, **PostgreSQL**, and **Redis** — started with one command.
+A reproducible local stack for the Learnault API: **API**, **wallet worker**, **scheduler**, **PostgreSQL**, and **Redis** — started with one command.
 
 ## Prerequisites
 
@@ -23,6 +23,7 @@ docker compose ps
 #    learnault-dev-db      Up ... (healthy)
 #    learnault-dev-redis   Up ... (healthy)
 #    learnault-dev-worker  Up ... (healthy)
+#    learnault-dev-scheduler  Up ...
 ```
 
 The API is available at `http://localhost:5000` (Swagger UI at `http://localhost:5000/api-docs`).
@@ -37,6 +38,37 @@ The `api` service entrypoint (`docker/entrypoint-dev-api.sh`) waits for PostgreS
 
 The `worker` service runs `src/workers/wallet-provisioning.worker.ts`, which polls the idempotent wallet-provisioning outbox and generates Stellar keys through the dev in-memory KMS adapter. In production, swap the KMS adapter for a real one (e.g. AWS KMS) behind the same `KmsSecretStore` interface.
 
+The `scheduler` service runs `src/workers/scheduler.worker.ts`. See below.
+
+## Scheduled job runner
+
+Every recurring queue drain is owned by the `scheduler` service, not by the request that enqueued the work — so a delivery whose `nextAttemptAt` falls due is retried on time even when the API is receiving no traffic, and request latency never includes queue-drain work.
+
+Registered queues: `email`, `notification`, `webhook`, `stellar-funding`, `data-export`, `account-lifecycle`.
+
+Each tick takes a row lease on `queue_leases` via `JobLeaseService.acquireQueueLease()` before draining, so extra replicas are safe:
+
+```bash
+docker compose up -d --scale scheduler=2
+```
+
+A replica that loses the race logs a skipped tick and moves on; a replica that crashes mid-drain has its lease expire, and the next tick reclaims the queue.
+
+| Variable | Default | Purpose |
+| --- | --- | --- |
+| `SCHEDULER_INTERVAL_MS` | `15000` | Base tick interval for every queue |
+| `SCHEDULER_<QUEUE>_INTERVAL_MS` | — | Per-queue override, e.g. `SCHEDULER_WEBHOOK_INTERVAL_MS` |
+| `SCHEDULER_LEASE_MS` | `60000` | Lease held per tick (floored at 2× the interval) |
+| `SCHEDULER_QUEUES` | all | Comma list restricting which queues this replica runs |
+| `SCHEDULER_DISABLED_QUEUES` | — | Comma list of queues to skip |
+| `SCHEDULER_SHUTDOWN_TIMEOUT_MS` | `30000` | How long `SIGTERM` waits for in-flight ticks |
+| `SCHEDULER_IN_PROCESS` | `false` | Opt-in: run the runner inside the API process for single-process deployments |
+| `LIFECYCLE_SWEEP_INTERVAL_MS` | `0` | When `> 0`, overrides the `account-lifecycle` queue interval |
+
+Every tick emits a structured log line carrying per-queue `depth`, `due`, `lagMs` (age of the oldest due row), `durationMs`, and cumulative `attempts` / `failures` / `skipped`.
+
+`pnpm scheduler:verify` runs both evidence scenarios against the stack: a due-but-failed delivery drained with no inbound HTTP traffic, then a batch drained by two replicas with no row processed twice.
+
 ## Health checks & readiness
 
 | Endpoint            | Meaning                                              |
@@ -54,7 +86,7 @@ The API container only reports **healthy** after `/health/live` responds; `depen
 pnpm stack:up        # docker compose up -d --build
 pnpm stack:down      # stop the stack (keeps data volumes)
 pnpm stack:reset     # stop + delete data volumes (project-scoped reset)
-pnpm stack:logs      # follow API + worker logs
+pnpm stack:logs      # follow API + worker + scheduler logs
 pnpm stack:validate  # docker compose config --quiet
 pnpm stack:smoke     # validate + start + probe health endpoints
 ```
@@ -65,9 +97,10 @@ pnpm stack:smoke     # validate + start + probe health endpoints
 docker compose logs -f          # all services
 docker compose logs -f api      # API only
 docker compose logs worker      # worker only
+docker compose logs -f scheduler # scheduled job runner only
 ```
 
-Both services have `stop_grace_period: 30s`, matching the app's graceful-shutdown handler (`SHUTDOWN_TIMEOUT_MS`): `docker compose down` sends `SIGTERM`, the server drains HTTP connections and closes the Prisma pool before exiting.
+`api` and `worker` have `stop_grace_period: 30s` and `scheduler` has `40s`, matching each process's graceful-shutdown handler (`SHUTDOWN_TIMEOUT_MS` / `SCHEDULER_SHUTDOWN_TIMEOUT_MS`): `docker compose down` sends `SIGTERM`, the server drains HTTP connections and closes the Prisma pool before exiting, and the scheduler stops scheduling, waits for in-flight ticks, and releases their queue leases so no queue is left parked.
 
 ## Data persistence & reset
 

docs/DEVELOPMENT_STACK.md

@@ -14,6 +14,7 @@ This document maps the key request flows and domain event propagation patterns a
 6. [Referral Application Flow](#referral-application-flow)
 7. [Withdrawal Flow](#withdrawal-flow)
 8. [Notification Delivery Flow](#notification-delivery-flow)
+9. [Subscribing to Domain Events](#subscribing-to-domain-events)
 
 ---
 
@@ -508,3 +509,121 @@ Target state: Event-driven communication via domain events
 | Credentials | Blockchain Infra | Service Call | On-chain storage |
 | Credentials | Notifications | Domain Event | Credential notification |
 | All Domains | Shared Kernel | Direct Import | Config, errors, middleware, utils |
+
+---
+
+## Subscribing to Domain Events
+
+A new domain event needs a registered handler, not a new worker process. The
+outbox relay leases every pending event and dispatches it by `eventType` to the
+handlers registered for that type.
+
+### 1. Declare the event schema
+
+Add the payload schema in `src/lib/transactions/event-schema.ts`. A handler for
+an event type with no schema is rejected at startup.
+
+```ts
+registry.register({
+  version: 1,
+  eventType: 'ModuleCompleted',
+  validate: async (payload) => {
+    await z.object({
+      completionId: z.string().uuid(),
+      userId: z.string().uuid(),
+    }).parseAsync(payload)
+  },
+})
+```
+
+### 2. Emit the event in the same transaction as the domain write
+
+```ts
+await prisma.$transaction(async (tx) => {
+  const completion = await tx.completion.create({ data: ... })
+
+  await createOutboxService(prisma).createEvent(tx, {
+    aggregateId: completion.id,
+    aggregateType: 'Completion',
+    eventType: 'ModuleCompleted',
+    eventVersion: 1,
+    payload: { completionId: completion.id, userId },
+    source: 'api.module.complete',
+  })
+
+  return completion
+})
+```
+
+If the transaction rolls back the event disappears with it, so an event can
+never describe a write that did not happen.
+
+### 3. Write the handler
+
+```ts
+export class RewardOnModuleCompleted implements OutboxEventHandler {
+  readonly name = 'rewards.on-module-completed'
+  readonly eventType = 'ModuleCompleted'
+  readonly eventVersion = 1
+  readonly maxAttempts = 5
+
+  async handle(ctx: OutboxEventHandlerContext) {
+    const payload = ctx.payload as ModuleCompletedPayload
+    await rewardService.grant(payload.userId, payload.completionId)
+
+    return { idempotencyKey: `${ctx.eventId}:${this.name}` }
+  }
+}
+```
+
+`name` must be unique across the whole registry — it becomes `JobAttempt.jobType`.
+
+**Handlers must be idempotent.** A handler can run more than once for the same
+event: after a crash mid-lease, or after an operator replays a dead-lettered
+event. Make the side effect an upsert, or guard it on a key derived from
+`ctx.eventId`.
+
+Throwing from `handle()` schedules a retry with exponential backoff. Returning
+normally completes the attempt.
+
+### 4. Register it
+
+Add the handler in `src/jobs/handler-registrations.ts`, and add its event type
+to `EMITTED_EVENT_TYPES` if the application emits it:
+
+```ts
+registry.register(new RewardOnModuleCompleted())
+```
+
+`registerOutboxHandlers()` throws at startup on a duplicate handler name, on a
+handler whose event type has no schema, and on an emitted event type with no
+handler — so a missing subscription fails loudly instead of leaving rows PENDING
+forever.
+
+### What the relay guarantees
+
+- One `JobAttempt` per (event, handler). Several handlers may subscribe to the
+  same event type and each is tracked separately.
+- An event becomes `PUBLISHED` only once **every** handler for its type has
+  completed. One failing handler holds the event back without blocking others.
+- A handler that exhausts `maxAttempts` dead-letters its own job and the event,
+  leaving every other event type unaffected.
+- An event with no registered handler is dead-lettered immediately and logged at
+  error level, rather than sitting `PENDING` unnoticed.
+
+### Operating dead letters
+
+```bash
+pnpm outbox:replay list                  # dead-lettered events and last error
+pnpm outbox:replay replay <eventId> ...  # reset to PENDING for another pass
+```
+
+Replay resets the dead-lettered `JobAttempt` rows and returns the event to
+`PENDING`; the relay picks it up on its next tick. Completed handlers are not
+re-run, and idempotent handlers make a repeated run harmless.
+
+### Where it runs
+
+The relay is a queue on the scheduled job runner
+(`src/workers/scheduler.worker.ts`), registered as `outbox-relay`. There is no
+per-domain worker process: adding a domain event means adding a handler.