Add Scheduled Background Job Runner

.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

@@ -117,6 +117,30 @@ services:
     command: ['./docker/entrypoint-dev-worker.sh']
     stop_grace_period: 30s
 
+  scheduler:
+    build:
+      context: .
+      dockerfile: docker/Dockerfile.dev
+    restart: unless-stopped
+    depends_on:
+      db:
+        condition: service_healthy
+    environment:
+      NODE_ENV: development
+      DATABASE_URL: postgresql://${POSTGRES_USER:-learnault}:${POSTGRES_PASSWORD:-learnault}@db:5432/${POSTGRES_DB:-learnault_dev}?schema=public
+      RUN_MIGRATIONS: 'true'
+      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-scheduler.sh']
+    stop_grace_period: 40s
+
 volumes:
   pgdata:
   redisdata:

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-scheduler.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

docs/DATA_LIFECYCLE.md

@@ -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/DEVELOPMENT_STACK.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
 

package.json

@@ -39,12 +39,15 @@
     "db:seed": "npm run seed",
     "db:studio": "prisma studio",
     "worker:dev": "tsx src/workers/wallet-provisioning.worker.ts",
+    "scheduler": "node dist/workers/scheduler.worker.js",
+    "scheduler:dev": "tsx src/workers/scheduler.worker.ts",
     "stack:validate": "docker compose config --quiet",
     "stack:up": "docker compose up -d --build",
     "stack:down": "docker compose down",
     "stack:reset": "docker compose down -v",
-    "stack:logs": "docker compose logs -f api worker",
-    "stack:smoke": "bash scripts/stack-smoke-test.sh"
+    "stack:logs": "docker compose logs -f api worker scheduler",
+    "stack:smoke": "bash scripts/stack-smoke-test.sh",
+    "scheduler:verify": "bash scripts/scheduler-verification.sh"
   },
   "dependencies": {
     "@prisma/adapter-pg": "^7.4.2",

prisma/migrations/20260829120000_scheduled_job_runner/migration.sql

@@ -0,0 +1,15 @@
+CREATE TABLE "queue_leases" (
+    "id" TEXT NOT NULL,
+    "queueName" TEXT NOT NULL,
+    "leaseToken" TEXT,
+    "leasedUntil" TIMESTAMP(3),
+    "owner" TEXT,
+    "lastTickAt" TIMESTAMP(3),
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "queue_leases_pkey" PRIMARY KEY ("id")
+);
+
+CREATE UNIQUE INDEX "queue_leases_queueName_key" ON "queue_leases"("queueName");
+CREATE INDEX "queue_leases_leasedUntil_idx" ON "queue_leases"("leasedUntil");

prisma/schema.prisma

@@ -841,3 +841,17 @@ model RolledBackRecord {
   @@index([createdAt]) // For periodic cleanup
   @@map("rolled_back_records")
 }
+
+model QueueLease {
+  id           String    @id @default(uuid())
+  queueName    String    @unique
+  leaseToken   String?
+  leasedUntil  DateTime?
+  owner        String?
+  lastTickAt   DateTime?
+  createdAt    DateTime  @default(now())
+  updatedAt    DateTime  @updatedAt
+
+  @@index([leasedUntil])
+  @@map("queue_leases")
+}

scripts/scheduler-verification.sh

@@ -0,0 +1,101 @@
+#!/usr/bin/env bash
+set -uo pipefail
+
+command -v docker >/dev/null 2>&1 || export PATH="$PATH:/c/Program Files/Docker/Docker/resources/bin"
+
+PGUSER_="${POSTGRES_USER:-learnault}"
+PGDB_="${POSTGRES_DB:-learnault_dev}"
+PGPORT_="${POSTGRES_PORT:-5433}"
+INTERVAL="${SCHEDULER_INTERVAL_MS:-4000}"
+BATCH="${BATCH_SIZE:-40}"
+LOGDIR="$(mktemp -d)"
+
+export DATABASE_URL="postgresql://${PGUSER_}:learnault@localhost:${PGPORT_}/${PGDB_}?schema=public"
+export NODE_ENV=production
+export LOG_LEVEL=debug
+export SCHEDULER_INTERVAL_MS="$INTERVAL"
+export SCHEDULER_LEASE_MS=10000
+
+q() { docker compose exec -T db psql -U "$PGUSER_" -d "$PGDB_" -qAt -c "$1" | tr -d '\r'; }
+
+echo "==> Starting PostgreSQL"
+docker compose up -d db >/dev/null 2>&1
+until docker compose exec -T db pg_isready -U "$PGUSER_" -d "$PGDB_" >/dev/null 2>&1; do sleep 1; done
+
+if [ -z "$(q "SELECT to_regclass('public.queue_leases');")" ]; then
+  echo "==> Syncing schema"
+  npx prisma db push --accept-data-loss >/dev/null 2>&1
+fi
+
+q "DELETE FROM email_deliveries WHERE type='SCHED_EVIDENCE';" >/dev/null
+q "DELETE FROM users WHERE username='sched-evidence';" >/dev/null
+USR="$(q "INSERT INTO users (id,email,username,password,role,\"isVerified\",status,\"createdAt\",\"updatedAt\")
+          VALUES (gen_random_uuid(),'sched-evidence@example.com','sched-evidence','x','LEARNER',true,'ACTIVE',now(),now())
+          RETURNING id;" | head -1)"
+
+echo ""
+echo "============================================================"
+echo " SCENARIO 1 — idle instance: failed delivery retried on time"
+echo "============================================================"
+
+q "INSERT INTO email_deliveries (id,\"userId\",\"to\",subject,body,type,status,error,\"attemptCount\",\"maxAttempts\",\"nextAttemptAt\",\"createdAt\",\"updatedAt\")
+   VALUES (gen_random_uuid(),'$USR','idle@example.com','retry me','<p>x</p>','SCHED_EVIDENCE','pending','previous attempt failed',1,5,now()-interval '1 minute',now(),now());" >/dev/null
+
+printf 'API on :5000   : '
+if curl -s -m 2 http://localhost:5000/health/live >/dev/null 2>&1; then echo "RUNNING (stop it for a clean result)"; else echo "not running — no HTTP traffic is possible"; fi
+printf 'before         : %s\n' "$(q "SELECT 'status='||status||'  attemptCount='||\"attemptCount\"||'  error='||error FROM email_deliveries WHERE type='SCHED_EVIDENCE';")"
+
+./node_modules/.bin/tsx src/workers/scheduler.worker.ts > "$LOGDIR/a.log" 2>&1 &
+PID_A=$!
+echo "scheduler      : started (interval ${INTERVAL}ms, no API process)"
+sleep 10
+
+printf 'after          : %s\n' "$(q "SELECT 'status='||status||'  attemptCount='||\"attemptCount\" FROM email_deliveries WHERE type='SCHED_EVIDENCE';")"
+echo "email tick     :"
+grep '"queue":"email"' "$LOGDIR/a.log" | head -1
+
+kill -TERM "$PID_A" 2>/dev/null; wait "$PID_A" 2>/dev/null
+echo "after SIGTERM  : $(q "SELECT count(*)||'/6 leases released' FROM queue_leases WHERE \"leaseToken\" IS NULL;")"
+
+S1="$(q "SELECT status FROM email_deliveries WHERE type='SCHED_EVIDENCE' LIMIT 1;")"
+[ "$S1" = "sent" ] && echo "RESULT         : PASS — drained on schedule with zero inbound HTTP" \
+                   || echo "RESULT         : FAIL — status=$S1"
+
+echo ""
+echo "============================================================"
+echo " SCENARIO 2 — two replicas: no duplicate processing"
+echo "============================================================"
+
+q "DELETE FROM email_deliveries WHERE type='SCHED_EVIDENCE';" >/dev/null
+q "INSERT INTO email_deliveries (id,\"userId\",\"to\",subject,body,type,status,\"attemptCount\",\"maxAttempts\",\"nextAttemptAt\",\"createdAt\",\"updatedAt\")
+   SELECT gen_random_uuid(),'$USR','r'||g||'@example.com','batch '||g,'<p>x</p>','SCHED_EVIDENCE','pending',0,5,now()-interval '1 minute',now(),now()
+   FROM generate_series(1,$BATCH) g;" >/dev/null
+
+echo "queued         : $(q "SELECT count(*) FROM email_deliveries WHERE type='SCHED_EVIDENCE';") due rows"
+
+SCHEDULER_OWNER_ID=replica-A ./node_modules/.bin/tsx src/workers/scheduler.worker.ts > "$LOGDIR/1.log" 2>&1 &
+P1=$!
+SCHEDULER_OWNER_ID=replica-B ./node_modules/.bin/tsx src/workers/scheduler.worker.ts > "$LOGDIR/2.log" 2>&1 &
+P2=$!
+echo "replicas       : replica-A and replica-B running concurrently"
+sleep 14
+kill -TERM "$P1" "$P2" 2>/dev/null; wait "$P1" "$P2" 2>/dev/null
+
+echo "attemptCounts  : $(q "SELECT string_agg('attemptCount='||\"attemptCount\"||' -> '||c||' rows',', ') FROM (SELECT \"attemptCount\",count(*) c FROM email_deliveries WHERE type='SCHED_EVIDENCE' GROUP BY 1 ORDER BY 1) t;")"
+A_SKIP=$(grep -c 'tick skipped' "$LOGDIR/1.log" 2>/dev/null || true)
+B_SKIP=$(grep -c 'tick skipped' "$LOGDIR/2.log" 2>/dev/null || true)
+echo "lease races    : replica-A skipped ${A_SKIP:-0}, replica-B skipped ${B_SKIP:-0}"
+
+DONE_=$(q "SELECT count(*) FROM email_deliveries WHERE type='SCHED_EVIDENCE' AND status='sent';")
+DUPE_=$(q "SELECT count(*) FROM email_deliveries WHERE type='SCHED_EVIDENCE' AND \"attemptCount\">1;")
+
+q "DELETE FROM email_deliveries WHERE type='SCHED_EVIDENCE';" >/dev/null
+q "DELETE FROM users WHERE username='sched-evidence';" >/dev/null
+rm -rf "$LOGDIR"
+
+if [ "$DUPE_" = "0" ] && [ "$DONE_" = "$BATCH" ]; then
+  echo "RESULT         : PASS — $DONE_/$BATCH processed exactly once, 0 duplicates"
+  exit 0
+fi
+echo "RESULT         : FAIL — processed=$DONE_ duplicates=$DUPE_"
+exit 1

src/audit/classification.ts

@@ -476,6 +476,18 @@ const RULES: readonly LifecycleRule[] = [
     audited: false,
     notes: 'Tombstone marking an event as unprocessable. Written once, then only read.',
   },
+  {
+    model: 'QueueLease',
+    table: 'queue_leases',
+    recordClass: RecordClass.MUTABLE,
+    category: DataCategory.OPERATIONAL,
+    retentionDays: Retention.INDEFINITE,
+    retentionAnchor: null,
+    onErasure: ErasureAction.RETAIN,
+    audited: false,
+    notes:
+      'One row per recurring queue drain, reused by every scheduler tick. Holds a queue name, lease token, and holder id — no user data, so nothing to erase and nothing to age out.',
+  },
 ]
 
 const BY_MODEL: ReadonlyMap<string, LifecycleRule> = new Map(

src/config/env.ts

@@ -35,7 +35,7 @@ export const env = {
   // Account lifecycle configurations
   DELETION_COOLING_OFF_DAYS: parseInt(process.env.DELETION_COOLING_OFF_DAYS || '30', 10),
   EXPORT_TTL_DAYS: parseInt(process.env.EXPORT_TTL_DAYS || '7', 10),
-  LIFECYCLE_SWEEP_INTERVAL_MS: parseInt(process.env.LIFECYCLE_SWEEP_INTERVAL_MS || '0', 10), // 0 = disabled
+  LIFECYCLE_SWEEP_INTERVAL_MS: parseInt(process.env.LIFECYCLE_SWEEP_INTERVAL_MS || '0', 10),
 
   // Data lifecycle / audit configurations — see docs/DATA_LIFECYCLE.md
   // HMAC key for the source-IP hash on audit events. Unset in production means