From c6c0a9a2e4c54d354b1ac16744e830cbeeabaac3 Mon Sep 17 00:00:00 2001
From: Ezeh <chijioke@clivealliance.com>
Date: Sun, 30 Aug 2026 00:45:49 +0100
Subject: [PATCH 1/2] feat(workers): add scheduled background job runner

---
 .env.example                                  |  20 +-
 Dockerfile                                    |   3 +-
 docker-compose.yml                            |  24 ++
 docker/entrypoint-dev-scheduler.sh            |  11 +
 docker/entrypoint-scheduler.sh                |  11 +
 docs/DATA_LIFECYCLE.md                        |   5 +-
 docs/DEVELOPMENT_STACK.md                     |  39 +-
 package.json                                  |   7 +-
 .../migration.sql                             |  15 +
 prisma/schema.prisma                          |  14 +
 scripts/scheduler-verification.sh             | 126 ++++++
 src/audit/classification.ts                   |  12 +
 src/config/env.ts                             |   2 +-
 src/config/scheduler.ts                       |  72 ++++
 src/controllers/account.controller.ts         |  10 -
 src/lib/transactions/job-lease.service.ts     |  81 +++-
 src/lib/transactions/types.ts                 |  12 +
 src/server.ts                                 |  32 +-
 src/services/account-lifecycle.service.ts     |   5 -
 src/services/data-export.service.ts           |   4 -
 src/services/email.service.ts                 |   4 -
 src/services/notification.service.ts          |   7 +-
 src/services/stellar-funding.service.ts       |  11 +-
 src/services/webhook.service.ts               |   3 -
 src/workers/queue-metrics.ts                  | 113 ++++++
 src/workers/queue-registry.ts                 | 114 ++++++
 src/workers/scheduled-job-runner.ts           | 242 ++++++++++++
 src/workers/scheduler.worker.ts               |  62 +++
 tests/email.service.test.ts                   |   3 +-
 tests/lib/job-lease-queue.test.ts             | 106 +++++
 tests/notification.service.test.ts            |  31 +-
 tests/services/webhook.service.spec.ts        |   3 +-
 tests/workers/scheduled-job-runner.test.ts    | 374 ++++++++++++++++++
 33 files changed, 1487 insertions(+), 91 deletions(-)
 create mode 100755 docker/entrypoint-dev-scheduler.sh
 create mode 100755 docker/entrypoint-scheduler.sh
 create mode 100644 prisma/migrations/20260829120000_scheduled_job_runner/migration.sql
 create mode 100755 scripts/scheduler-verification.sh
 create mode 100644 src/config/scheduler.ts
 create mode 100644 src/workers/queue-metrics.ts
 create mode 100644 src/workers/queue-registry.ts
 create mode 100644 src/workers/scheduled-job-runner.ts
 create mode 100644 src/workers/scheduler.worker.ts
 create mode 100644 tests/lib/job-lease-queue.test.ts
 create mode 100644 tests/workers/scheduled-job-runner.test.ts

diff --git a/.env.example b/.env.example
index d6b8c521..1929c080 100644
--- a/.env.example
+++ b/.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,7 +62,6 @@ 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
@@ -63,9 +69,9 @@ LIFECYCLE_SWEEP_INTERVAL_MS=0
 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
diff --git a/Dockerfile b/Dockerfile
index db4dc82f..9541e982 100644
--- a/Dockerfile
+++ b/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 && \
diff --git a/docker-compose.yml b/docker-compose.yml
index 8921f699..3bc21830 100644
--- a/docker-compose.yml
+++ b/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:
diff --git a/docker/entrypoint-dev-scheduler.sh b/docker/entrypoint-dev-scheduler.sh
new file mode 100755
index 00000000..f88f4ea1
--- /dev/null
+++ b/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
diff --git a/docker/entrypoint-scheduler.sh b/docker/entrypoint-scheduler.sh
new file mode 100755
index 00000000..6b7de1e1
--- /dev/null
+++ b/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
diff --git a/docs/DATA_LIFECYCLE.md b/docs/DATA_LIFECYCLE.md
index 92982178..6e136725 100644
--- a/docs/DATA_LIFECYCLE.md
+++ b/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.
 
 ---
 
diff --git a/docs/DEVELOPMENT_STACK.md b/docs/DEVELOPMENT_STACK.md
index 2f293475..9fb734f3 100644
--- a/docs/DEVELOPMENT_STACK.md
+++ b/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
 
diff --git a/package.json b/package.json
index c30a7c2b..21690f45 100644
--- a/package.json
+++ b/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",
diff --git a/prisma/migrations/20260829120000_scheduled_job_runner/migration.sql b/prisma/migrations/20260829120000_scheduled_job_runner/migration.sql
new file mode 100644
index 00000000..aa07da9e
--- /dev/null
+++ b/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");
diff --git a/prisma/schema.prisma b/prisma/schema.prisma
index 2477b20b..7d3490b8 100644
--- a/prisma/schema.prisma
+++ b/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")
+}
diff --git a/scripts/scheduler-verification.sh b/scripts/scheduler-verification.sh
new file mode 100755
index 00000000..822c557d
--- /dev/null
+++ b/scripts/scheduler-verification.sh
@@ -0,0 +1,126 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+COMPOSE_FILE="${COMPOSE_FILE:-docker-compose.yml}"
+POSTGRES_USER="${POSTGRES_USER:-learnault}"
+POSTGRES_DB="${POSTGRES_DB:-learnault_dev}"
+INTERVAL_MS="${SCHEDULER_INTERVAL_MS:-15000}"
+BATCH_SIZE="${BATCH_SIZE:-40}"
+
+dc() { docker compose -f "$COMPOSE_FILE" "$@"; }
+psql_q() { dc exec -T db psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -At -c "$1" | tr -d '\r'; }
+
+export LOG_LEVEL=debug
+
+wait_seconds=$(( (INTERVAL_MS / 1000) * 3 + 5 ))
+
+echo "==> Bringing up the stack (LOG_LEVEL=debug)"
+dc up -d --build
+
+echo "==> Waiting for the database"
+until dc exec -T db pg_isready -U "$POSTGRES_USER" -d "$POSTGRES_DB" >/dev/null 2>&1; do
+  sleep 2
+done
+
+seed_user() {
+  psql_q "
+    INSERT INTO users (id, email, username, password, role, \"isVerified\", status, \"createdAt\", \"updatedAt\")
+    VALUES (gen_random_uuid(), 'scheduler-evidence@example.com', 'scheduler-evidence', 'x', 'LEARNER', true, 'ACTIVE', now(), now())
+    ON CONFLICT (email) DO UPDATE SET \"updatedAt\" = now()
+    RETURNING id;
+  "
+}
+
+echo ""
+echo "############################################################"
+echo "# 1. Idle instance: a due retry drains with no HTTP traffic #"
+echo "############################################################"
+
+USER_ID="$(seed_user)"
+
+psql_q "DELETE FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
+
+psql_q "
+  INSERT INTO email_deliveries
+    (id, \"userId\", \"to\", subject, body, type, status, error, \"attemptCount\", \"maxAttempts\", \"nextAttemptAt\", \"createdAt\", \"updatedAt\")
+  VALUES
+    (gen_random_uuid(), '$USER_ID', 'idle@example.com', 'idle-instance retry', '<p>evidence</p>',
+     'SCHEDULER_EVIDENCE', 'pending', 'previous attempt failed', 1, 5, now() - interval '1 minute', now(), now());
+"
+
+echo "--> Before (no HTTP traffic will be sent to the API):"
+psql_q "SELECT status, \"attemptCount\", \"nextAttemptAt\" FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
+
+echo "--> Waiting ${wait_seconds}s (SCHEDULER_INTERVAL_MS=${INTERVAL_MS})"
+sleep "$wait_seconds"
+
+echo "--> After:"
+psql_q "SELECT status, \"attemptCount\", \"sentAt\" FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
+
+echo "--> Scheduler log lines for the email queue:"
+dc logs --no-log-prefix scheduler | grep -E '"queue": ?"email"' | tail -5 || true
+
+IDLE_STATUS="$(psql_q "SELECT status FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE' LIMIT 1;")"
+if [ "$IDLE_STATUS" != "sent" ]; then
+  echo "!! Expected the due row to be drained, got status='$IDLE_STATUS'" >&2
+  dc logs scheduler | tail -40
+  exit 1
+fi
+echo "✅ Due row drained on schedule with no inbound HTTP traffic"
+
+echo ""
+echo "###############################################################"
+echo "# 2. Two replicas: due rows are processed exactly once         #"
+echo "###############################################################"
+
+echo "--> Scaling the scheduler to 2 replicas"
+dc up -d --scale scheduler=2
+sleep 5
+
+psql_q "DELETE FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
+psql_q "
+  INSERT INTO email_deliveries
+    (id, \"userId\", \"to\", subject, body, type, status, \"attemptCount\", \"maxAttempts\", \"nextAttemptAt\", \"createdAt\", \"updatedAt\")
+  SELECT gen_random_uuid(), '$USER_ID', 'replica-' || g || '@example.com', 'replica batch ' || g, '<p>evidence</p>',
+         'SCHEDULER_EVIDENCE', 'pending', 0, 5, now() - interval '1 minute', now(), now()
+  FROM generate_series(1, $BATCH_SIZE) AS g;
+"
+
+echo "--> Queued $BATCH_SIZE due rows across 2 replicas; waiting ${wait_seconds}s"
+sleep "$wait_seconds"
+
+echo "--> Attempt counts (a row drained twice would show attemptCount > 1):"
+psql_q "
+  SELECT \"attemptCount\", count(*)
+  FROM email_deliveries
+  WHERE type = 'SCHEDULER_EVIDENCE'
+  GROUP BY \"attemptCount\"
+  ORDER BY \"attemptCount\";
+"
+
+echo "--> Skipped ticks (the replica that lost the lease race):"
+SKIPPED="$(dc logs --no-log-prefix scheduler | grep -c 'queue tick skipped' || true)"
+echo "    $SKIPPED skipped tick(s) logged"
+
+DUPLICATES="$(psql_q "SELECT count(*) FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE' AND \"attemptCount\" > 1;")"
+PROCESSED="$(psql_q "SELECT count(*) FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE' AND status = 'sent';")"
+
+echo "--> processed=$PROCESSED duplicates=$DUPLICATES"
+
+if [ "$DUPLICATES" != "0" ]; then
+  echo "!! $DUPLICATES row(s) were processed more than once" >&2
+  exit 1
+fi
+if [ "$PROCESSED" != "$BATCH_SIZE" ]; then
+  echo "!! Expected $BATCH_SIZE processed rows, got $PROCESSED" >&2
+  exit 1
+fi
+echo "✅ Two replicas processed $BATCH_SIZE rows with no duplicates"
+
+echo ""
+echo "==> Cleaning up evidence rows"
+psql_q "DELETE FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
+dc up -d --scale scheduler=1
+
+echo ""
+echo "✅ Scheduler verification passed"
diff --git a/src/audit/classification.ts b/src/audit/classification.ts
index 628c37d3..18e70d3a 100644
--- a/src/audit/classification.ts
+++ b/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(
diff --git a/src/config/env.ts b/src/config/env.ts
index b789b9ac..19605caa 100644
--- a/src/config/env.ts
+++ b/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
diff --git a/src/config/scheduler.ts b/src/config/scheduler.ts
new file mode 100644
index 00000000..fac95d97
--- /dev/null
+++ b/src/config/scheduler.ts
@@ -0,0 +1,72 @@
+import { config } from 'dotenv'
+import os from 'os'
+
+config()
+
+const DEFAULT_INTERVAL_MS = 15_000
+const DEFAULT_LEASE_MS = 60_000
+const DEFAULT_SHUTDOWN_TIMEOUT_MS = 30_000
+
+function toInt(value: string | undefined, fallback: number): number {
+  const parsed = parseInt(value ?? '', 10)
+
+  return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback
+}
+
+function toBool(value: string | undefined, fallback = false): boolean {
+  if (value === undefined || value === '') return fallback
+
+  return value === 'true' || value === '1'
+}
+
+function toList(value: string | undefined): string[] {
+  return (value ?? '')
+    .split(',')
+    .map(entry => entry.trim())
+    .filter(entry => entry.length > 0)
+}
+
+function envKeyFor(queueName: string): string {
+  return `SCHEDULER_${queueName.replace(/[^a-zA-Z0-9]+/g, '_').toUpperCase()}_INTERVAL_MS`
+}
+
+export const schedulerConfig = {
+  intervalMs: toInt(process.env.SCHEDULER_INTERVAL_MS, DEFAULT_INTERVAL_MS),
+  leaseMs: toInt(process.env.SCHEDULER_LEASE_MS, DEFAULT_LEASE_MS),
+  shutdownTimeoutMs: toInt(
+    process.env.SCHEDULER_SHUTDOWN_TIMEOUT_MS,
+    DEFAULT_SHUTDOWN_TIMEOUT_MS
+  ),
+  inProcess: toBool(process.env.SCHEDULER_IN_PROCESS, false),
+  only: toList(process.env.SCHEDULER_QUEUES),
+  disabled: toList(process.env.SCHEDULER_DISABLED_QUEUES),
+  ownerId: process.env.SCHEDULER_OWNER_ID || `${os.hostname()}:${process.pid}`,
+
+  isEnabled(queueName: string): boolean {
+    if (this.disabled.includes(queueName)) return false
+
+    return this.only.length === 0 || this.only.includes(queueName)
+  },
+
+  intervalFor(queueName: string): number {
+    const override = process.env[envKeyFor(queueName)]
+    if (override !== undefined) {
+      return toInt(override, this.intervalMs)
+    }
+
+    if (queueName === 'account-lifecycle') {
+      const legacy = parseInt(process.env.LIFECYCLE_SWEEP_INTERVAL_MS ?? '', 10)
+      if (Number.isFinite(legacy) && legacy > 0) {
+        return legacy
+      }
+    }
+
+    return this.intervalMs
+  },
+
+  leaseFor(queueName: string): number {
+    return Math.max(this.leaseMs, this.intervalFor(queueName) * 2)
+  },
+}
+
+export type SchedulerConfig = typeof schedulerConfig
diff --git a/src/controllers/account.controller.ts b/src/controllers/account.controller.ts
index 05b97584..4c9d4234 100644
--- a/src/controllers/account.controller.ts
+++ b/src/controllers/account.controller.ts
@@ -82,8 +82,6 @@ export class AccountController {
         try {
             const userId = req.user!.id
 
-            this.sweepInBackground()
-
             const result = await dataExportService.requestExport(userId)
 
             if (result.kind === 'duplicate') {
@@ -466,8 +464,6 @@ export class AccountController {
      */
     async getDeletionStatus(req: Request, res: Response): Promise<void> {
         try {
-            this.sweepInBackground()
-
             const request = await accountLifecycleService.getLatestDeletionRequest(req.user!.id)
 
             if (!request) {
@@ -619,12 +615,6 @@ export class AccountController {
         }
     }
 
-    private sweepInBackground(): void {
-        accountLifecycleService.sweep().catch(err =>
-            logger.error('Lifecycle sweep error:', err)
-        )
-    }
-
     private generateToken(userId: string, role: string): string {
         return issueAccessToken({ id: userId, role })
     }
diff --git a/src/lib/transactions/job-lease.service.ts b/src/lib/transactions/job-lease.service.ts
index 2d0976f3..3f3fae6a 100644
--- a/src/lib/transactions/job-lease.service.ts
+++ b/src/lib/transactions/job-lease.service.ts
@@ -10,9 +10,18 @@
  */
 
 import { PrismaClient } from '@prisma/client'
-import { LeaseJobOptions, LeaseJobResult, JobResult, JobAttempt } from './types.js'
+import {
+  LeaseJobOptions,
+  LeaseJobResult,
+  JobResult,
+  JobAttempt,
+  AcquireQueueLeaseOptions,
+  QueueLeaseResult,
+} from './types.js'
 import { randomUUID } from 'crypto'
 
+const DEFAULT_QUEUE_LEASE_MS = 60000
+
 export class JobLeaseService {
   constructor(private prisma: PrismaClient) {}
 
@@ -307,6 +316,76 @@ export class JobLeaseService {
     return result.count
   }
 
+  async acquireQueueLease(
+    options: AcquireQueueLeaseOptions
+  ): Promise<QueueLeaseResult | null> {
+    const leaseMs = options.leaseMs ?? DEFAULT_QUEUE_LEASE_MS
+    const leaseToken = randomUUID()
+    const owner = options.owner ?? null
+
+    const rows = await this.prisma.$queryRaw<Array<{ leasedUntil: Date }>>`
+      INSERT INTO "queue_leases"
+        ("id", "queueName", "leaseToken", "leasedUntil", "owner", "createdAt", "updatedAt")
+      VALUES (
+        ${randomUUID()},
+        ${options.queueName},
+        ${leaseToken},
+        now() + (${String(leaseMs)}::text || ' milliseconds')::interval,
+        ${owner},
+        now(),
+        now()
+      )
+      ON CONFLICT ("queueName") DO UPDATE
+        SET "leaseToken" = EXCLUDED."leaseToken",
+            "leasedUntil" = EXCLUDED."leasedUntil",
+            "owner" = EXCLUDED."owner",
+            "updatedAt" = now()
+        WHERE "queue_leases"."leasedUntil" IS NULL
+           OR "queue_leases"."leasedUntil" <= now()
+      RETURNING "leasedUntil"
+    `
+
+    if (rows.length === 0) {
+      return null
+    }
+
+    return {
+      queueName: options.queueName,
+      leaseToken,
+      leasedUntil: rows[0].leasedUntil,
+    }
+  }
+
+  async renewQueueLease(
+    queueName: string,
+    leaseToken: string,
+    leaseMs: number = DEFAULT_QUEUE_LEASE_MS
+  ): Promise<boolean> {
+    const updated = await this.prisma.$executeRaw`
+      UPDATE "queue_leases"
+      SET "leasedUntil" = now() + (${String(leaseMs)}::text || ' milliseconds')::interval,
+          "updatedAt" = now()
+      WHERE "queueName" = ${queueName}
+        AND "leaseToken" = ${leaseToken}
+    `
+
+    return updated > 0
+  }
+
+  async releaseQueueLease(queueName: string, leaseToken: string): Promise<boolean> {
+    const updated = await this.prisma.$executeRaw`
+      UPDATE "queue_leases"
+      SET "leaseToken" = NULL,
+          "leasedUntil" = NULL,
+          "lastTickAt" = now(),
+          "updatedAt" = now()
+      WHERE "queueName" = ${queueName}
+        AND "leaseToken" = ${leaseToken}
+    `
+
+    return updated > 0
+  }
+
   /**
    * Get dead-letter jobs for manual inspection and recovery
    *
diff --git a/src/lib/transactions/types.ts b/src/lib/transactions/types.ts
index 25ec87e9..c71c9bde 100644
--- a/src/lib/transactions/types.ts
+++ b/src/lib/transactions/types.ts
@@ -160,3 +160,15 @@ export interface LeaseJobResult {
   attempt: number
   payload: unknown
 }
+
+export interface AcquireQueueLeaseOptions {
+  queueName: string
+  leaseMs?: number
+  owner?: string
+}
+
+export interface QueueLeaseResult {
+  queueName: string
+  leaseToken: string
+  leasedUntil: Date
+}
diff --git a/src/server.ts b/src/server.ts
index 087c2cc9..0c582154 100644
--- a/src/server.ts
+++ b/src/server.ts
@@ -1,9 +1,9 @@
 import { Server } from 'http'
 import app from './app'
-import { env } from './config/env'
-import { accountLifecycleService } from './services/account-lifecycle.service'
+import { schedulerConfig } from './config/scheduler'
 import logger from './utils/logger'
 import prisma from './config/database'
+import { createScheduledJobRunner, ScheduledJobRunner } from './workers/scheduled-job-runner'
 
 const PORT = process.env.PORT || 5000
 const SHUTDOWN_TIMEOUT_MS = parseInt(process.env.SHUTDOWN_TIMEOUT_MS || '30000', 10)
@@ -13,18 +13,14 @@ const server: Server = app.listen(PORT, () => {
 })
 
 let isShuttingDown = false
-let lifecycleSweepInterval: NodeJS.Timeout | null = null
-
-// Periodic lifecycle sweep (export generation, deletion finalization, artifact
-// purge). Disabled when LIFECYCLE_SWEEP_INTERVAL_MS is 0 — the sweep still runs
-// lazily from account endpoints, and a dedicated worker can call sweep() directly.
-if (env.LIFECYCLE_SWEEP_INTERVAL_MS > 0) {
-  lifecycleSweepInterval = setInterval(() => {
-    accountLifecycleService.sweep().catch(err =>
-      logger.error('Scheduled lifecycle sweep error:', err)
-    )
-  }, env.LIFECYCLE_SWEEP_INTERVAL_MS)
-  lifecycleSweepInterval.unref()
+let scheduler: ScheduledJobRunner | null = null
+
+if (schedulerConfig.inProcess) {
+  scheduler = createScheduledJobRunner({ prisma })
+  scheduler.start()
+  logger.info(
+    `In-process scheduler enabled for queues: ${scheduler.registeredQueues.join(', ') || 'none'}`
+  )
 }
 
 /**
@@ -63,10 +59,10 @@ async function gracefulShutdown(signal: string): Promise<void> {
     })
 
     // 2. Stop background jobs
-    if (lifecycleSweepInterval) {
-      logger.info('Stopping lifecycle sweep interval...')
-      clearInterval(lifecycleSweepInterval)
-      lifecycleSweepInterval = null
+    if (scheduler) {
+      logger.info('Stopping in-process scheduler...')
+      await scheduler.stop()
+      scheduler = null
     }
 
     // 3. Close database connections
diff --git a/src/services/account-lifecycle.service.ts b/src/services/account-lifecycle.service.ts
index 720a616f..7f12498a 100644
--- a/src/services/account-lifecycle.service.ts
+++ b/src/services/account-lifecycle.service.ts
@@ -284,11 +284,6 @@ export class AccountLifecycleService {
     }
   }
 
-  /**
-   * Runs all due lifecycle work. Invoked lazily from account endpoints,
-   * optionally on an interval from server.ts, and callable from a future
-   * dedicated worker (docker/entrypoint-worker.sh).
-   */
   async sweep(): Promise<void> {
     const results = await Promise.allSettled([
       this.processDue(),
diff --git a/src/services/data-export.service.ts b/src/services/data-export.service.ts
index e0b5ad5e..eaabe5bd 100644
--- a/src/services/data-export.service.ts
+++ b/src/services/data-export.service.ts
@@ -59,10 +59,6 @@ export class DataExportService {
 
     await auditService.record({ userId, action: AuditAction.EXPORT_REQUESTED, metadata: { requestId: request.id } })
 
-    this.processQueue().catch(err =>
-      logger.error('[DataExportService] Queue processing error:', err)
-    )
-
     return { kind: 'created', request }
   }
 
diff --git a/src/services/email.service.ts b/src/services/email.service.ts
index c8f6f5e6..5973e0fa 100644
--- a/src/services/email.service.ts
+++ b/src/services/email.service.ts
@@ -39,10 +39,6 @@ export class EmailService {
       },
     })
 
-    this.processQueue().catch(err =>
-      logger.error('[EmailService] Queue processing error:', err)
-    )
-
     return delivery as unknown as EmailDeliveryRecord
   }
 
diff --git a/src/services/notification.service.ts b/src/services/notification.service.ts
index a706bceb..cdc6cfc0 100644
--- a/src/services/notification.service.ts
+++ b/src/services/notification.service.ts
@@ -1,5 +1,5 @@
 import prisma from '../config/database'
-import * as admin from 'firebase-admin'
+import admin from 'firebase-admin'
 
 // Local type definition to avoid @prisma/client import at test time
 interface NotificationLog {
@@ -94,11 +94,6 @@ export class NotificationService {
       data: { userId, type, title, body, status: 'pending', nextAttemptAt: new Date() }
     })
 
-    // Process asynchronously – same pattern as webhook service
-    this.processQueue().catch(err =>
-      console.error('[NotificationService] Queue processing error:', err)
-    )
-
     return log as unknown as NotificationLog
   }
 
diff --git a/src/services/stellar-funding.service.ts b/src/services/stellar-funding.service.ts
index 78e166fd..95f46ae1 100644
--- a/src/services/stellar-funding.service.ts
+++ b/src/services/stellar-funding.service.ts
@@ -1,7 +1,6 @@
 import prisma from '../config/database'
 import { stellarConfig } from '../config/stellar'
 import { StellarService, StellarServiceError } from './stellar.service'
-import logger from '../utils/logger'
 
 interface StellarFundingRecord {
   id: string
@@ -33,11 +32,7 @@ export class StellarFundingService {
     })
 
     if (existing) {
-      this.processQueue().catch((err) =>
-        logger.error('[StellarFundingService] Queue processing error:', err)
-      )
-      
-return existing as unknown as StellarFundingRecord
+      return existing as unknown as StellarFundingRecord
     }
 
     const funding = await prisma.stellarFunding.create({
@@ -49,10 +44,6 @@ return existing as unknown as StellarFundingRecord
       },
     })
 
-    this.processQueue().catch((err) =>
-      logger.error('[StellarFundingService] Queue processing error:', err)
-    )
-
     return funding as unknown as StellarFundingRecord
   }
 
diff --git a/src/services/webhook.service.ts b/src/services/webhook.service.ts
index fdfadf8f..f0c69b5b 100644
--- a/src/services/webhook.service.ts
+++ b/src/services/webhook.service.ts
@@ -56,9 +56,6 @@ export class WebhookService {
                 })
             })
         )
-
-        // Process asynchronously
-        this.processQueue().catch(err => console.error('[Webhook] Queue processing error:', err))
     }
 
     /**
diff --git a/src/workers/queue-metrics.ts b/src/workers/queue-metrics.ts
new file mode 100644
index 00000000..1f775f86
--- /dev/null
+++ b/src/workers/queue-metrics.ts
@@ -0,0 +1,113 @@
+import logger from '../utils/logger'
+
+export type TickOutcome = 'ran' | 'skipped' | 'failed'
+
+export interface QueueDepthSnapshot {
+  depth: number
+  due: number
+  oldestDueAt: Date | null
+}
+
+export interface QueueTickSample {
+  queue: string
+  outcome: TickOutcome
+  durationMs: number
+  depth: number
+  due: number
+  lagMs: number
+  error?: string
+}
+
+export interface QueueMetricsSnapshot {
+  queue: string
+  attempts: number
+  failures: number
+  skipped: number
+  depth: number
+  due: number
+  lagMs: number
+  lastDurationMs: number
+  lastRunAt: string | null
+  lastError: string | null
+}
+
+const emptyMetrics = (queue: string): QueueMetricsSnapshot => ({
+  queue,
+  attempts: 0,
+  failures: 0,
+  skipped: 0,
+  depth: 0,
+  due: 0,
+  lagMs: 0,
+  lastDurationMs: 0,
+  lastRunAt: null,
+  lastError: null,
+})
+
+export class QueueMetricsRegistry {
+  private readonly queues = new Map<string, QueueMetricsSnapshot>()
+
+  record(sample: QueueTickSample): QueueMetricsSnapshot {
+    const current = this.queues.get(sample.queue) ?? emptyMetrics(sample.queue)
+
+    const next: QueueMetricsSnapshot = {
+      ...current,
+      depth: sample.depth,
+      due: sample.due,
+      lagMs: sample.lagMs,
+      lastDurationMs: sample.durationMs,
+    }
+
+    if (sample.outcome === 'skipped') {
+      next.skipped = current.skipped + 1
+    } else {
+      next.attempts = current.attempts + 1
+      next.lastRunAt = new Date().toISOString()
+    }
+
+    if (sample.outcome === 'failed') {
+      next.failures = current.failures + 1
+      next.lastError = sample.error ?? 'unknown error'
+    } else if (sample.outcome === 'ran') {
+      next.lastError = null
+    }
+
+    this.queues.set(sample.queue, next)
+    this.emit(sample, next)
+
+    return next
+  }
+
+  snapshot(): QueueMetricsSnapshot[] {
+    return [...this.queues.values()].sort((a, b) => a.queue.localeCompare(b.queue))
+  }
+
+  reset(): void {
+    this.queues.clear()
+  }
+
+  private emit(sample: QueueTickSample, totals: QueueMetricsSnapshot): void {
+    const meta = {
+      queue: sample.queue,
+      outcome: sample.outcome,
+      depth: totals.depth,
+      due: totals.due,
+      lagMs: totals.lagMs,
+      durationMs: totals.lastDurationMs,
+      attempts: totals.attempts,
+      failures: totals.failures,
+      skipped: totals.skipped,
+      ...(sample.error ? { error: sample.error } : {}),
+    }
+
+    if (sample.outcome === 'failed') {
+      logger.error('[scheduler] queue tick failed', meta)
+    } else if (sample.outcome === 'skipped') {
+      logger.debug('[scheduler] queue tick skipped (lease held elsewhere)', meta)
+    } else {
+      logger.info('[scheduler] queue tick', meta)
+    }
+  }
+}
+
+export const queueMetrics = new QueueMetricsRegistry()
diff --git a/src/workers/queue-registry.ts b/src/workers/queue-registry.ts
new file mode 100644
index 00000000..9e2f31fd
--- /dev/null
+++ b/src/workers/queue-registry.ts
@@ -0,0 +1,114 @@
+import type { PrismaClient } from '@prisma/client'
+import defaultPrisma from '../config/database'
+import { accountLifecycleService } from '../services/account-lifecycle.service'
+import { dataExportService } from '../services/data-export.service'
+import { emailService } from '../services/email.service'
+import { NotificationService } from '../services/notification.service'
+import { stellarFundingService } from '../services/stellar-funding.service'
+import { WebhookService } from '../services/webhook.service'
+import { DeletionStatus, ExportStatus } from '../types/account.types'
+import type { QueueDepthSnapshot } from './queue-metrics'
+
+export interface ScheduledQueue {
+  name: string
+  drain(): Promise<void>
+  inspect(): Promise<QueueDepthSnapshot>
+}
+
+interface DelegateDepthOptions {
+  pending: Record<string, unknown>
+  dueField?: string
+}
+
+interface CountableDelegate {
+  count(args: { where: Record<string, unknown> }): Promise<number>
+  findFirst(args: {
+    where: Record<string, unknown>
+    orderBy: Record<string, unknown>
+    select: Record<string, unknown>
+  }): Promise<Record<string, Date | null> | null>
+}
+
+function delegateDepth(
+  delegate: CountableDelegate,
+  options: DelegateDepthOptions
+): () => Promise<QueueDepthSnapshot> {
+  const dueField = options.dueField ?? 'nextAttemptAt'
+
+  return async () => {
+    const now = new Date()
+    const dueWhere = {
+      ...options.pending,
+      OR: [{ [dueField]: null }, { [dueField]: { lte: now } }],
+    }
+
+    const [depth, due, oldest] = await Promise.all([
+      delegate.count({ where: options.pending }),
+      delegate.count({ where: dueWhere }),
+      delegate.findFirst({
+        where: dueWhere,
+        orderBy: { [dueField]: 'asc' },
+        select: { [dueField]: true },
+      }),
+    ])
+
+    return { depth, due, oldestDueAt: (oldest?.[dueField] as Date | null) ?? null }
+  }
+}
+
+export interface QueueRegistryDeps {
+  prisma?: PrismaClient
+  notificationService?: Pick<NotificationService, 'processQueue'>
+  webhookService?: Pick<WebhookService, 'processQueue'>
+}
+
+export function createDefaultQueues(deps: QueueRegistryDeps = {}): ScheduledQueue[] {
+  const prisma = deps.prisma ?? defaultPrisma
+  const notificationService = deps.notificationService ?? new NotificationService()
+  const webhookService = deps.webhookService ?? new WebhookService()
+
+  const db = prisma as unknown as Record<string, CountableDelegate>
+
+  return [
+    {
+      name: 'email',
+      drain: () => emailService.processQueue(),
+      inspect: delegateDepth(db.emailDelivery, { pending: { status: 'pending' } }),
+    },
+    {
+      name: 'notification',
+      drain: () => notificationService.processQueue(),
+      inspect: delegateDepth(db.notificationLog, { pending: { status: 'pending' } }),
+    },
+    {
+      name: 'webhook',
+      drain: () => webhookService.processQueue(),
+      inspect: delegateDepth(db.webhookDelivery, { pending: { status: 'pending' } }),
+    },
+    {
+      name: 'stellar-funding',
+      drain: () => stellarFundingService.processQueue(),
+      inspect: delegateDepth(db.stellarFunding, {
+        pending: { status: { in: ['pending', 'submitted'] } },
+      }),
+    },
+    {
+      name: 'data-export',
+      drain: async () => {
+        await dataExportService.processQueue()
+        await dataExportService.purgeExpired()
+      },
+      inspect: delegateDepth(db.dataExportRequest, {
+        pending: { status: ExportStatus.PENDING },
+      }),
+    },
+    {
+      name: 'account-lifecycle',
+      drain: () => accountLifecycleService.processDue(),
+      inspect: delegateDepth(db.accountDeletionRequest, {
+        pending: { status: DeletionStatus.PENDING },
+        dueField: 'scheduledFor',
+      }),
+    },
+  ]
+}
diff --git a/src/workers/scheduled-job-runner.ts b/src/workers/scheduled-job-runner.ts
new file mode 100644
index 00000000..9d1946b9
--- /dev/null
+++ b/src/workers/scheduled-job-runner.ts
@@ -0,0 +1,242 @@
+import type { PrismaClient } from '@prisma/client'
+import defaultPrisma from '../config/database'
+import { schedulerConfig, type SchedulerConfig } from '../config/scheduler'
+import { createJobLeaseService, JobLeaseService } from '../lib/transactions/job-lease.service'
+import logger from '../utils/logger'
+import { queueMetrics, QueueMetricsRegistry, type QueueDepthSnapshot, type TickOutcome } from './queue-metrics'
+import { createDefaultQueues, type ScheduledQueue } from './queue-registry'
+
+export type QueueLeaseApi = Pick<
+  JobLeaseService,
+  'acquireQueueLease' | 'renewQueueLease' | 'releaseQueueLease'
+>
+
+export interface ScheduledJobRunnerOptions {
+  queues: ScheduledQueue[]
+  leaseService: QueueLeaseApi
+  config?: SchedulerConfig
+  metrics?: QueueMetricsRegistry
+  log?: Pick<typeof logger, 'info' | 'warn' | 'error' | 'debug'>
+}
+
+const EMPTY_DEPTH: QueueDepthSnapshot = { depth: 0, due: 0, oldestDueAt: null }
+
+export class ScheduledJobRunner {
+  private readonly queues: ScheduledQueue[]
+  private readonly leaseService: QueueLeaseApi
+  private readonly config: SchedulerConfig
+  private readonly metrics: QueueMetricsRegistry
+  private readonly log: NonNullable<ScheduledJobRunnerOptions['log']>
+
+  private readonly timers = new Map<string, NodeJS.Timeout>()
+  private readonly inFlight = new Map<string, Promise<TickOutcome>>()
+  private started = false
+  private stopping = false
+
+  constructor(options: ScheduledJobRunnerOptions) {
+    this.config = options.config ?? schedulerConfig
+    this.queues = options.queues.filter(queue => this.config.isEnabled(queue.name))
+    this.leaseService = options.leaseService
+    this.metrics = options.metrics ?? queueMetrics
+    this.log = options.log ?? logger
+  }
+
+  get registeredQueues(): string[] {
+    return this.queues.map(queue => queue.name)
+  }
+
+  start(): void {
+    if (this.started) return
+    this.started = true
+    this.stopping = false
+
+    if (this.queues.length === 0) {
+      this.log.warn('[scheduler] no queues enabled; runner idle')
+
+      return
+    }
+
+    for (const queue of this.queues) {
+      this.log.info(
+        `[scheduler] registered queue "${queue.name}" (every ${this.config.intervalFor(queue.name)}ms)`
+      )
+      this.schedule(queue, 0)
+    }
+  }
+
+  async runOnce(): Promise<Record<string, TickOutcome>> {
+    const outcomes: Record<string, TickOutcome> = {}
+
+    for (const queue of this.queues) {
+      outcomes[queue.name] = await this.runQueue(queue)
+    }
+
+    return outcomes
+  }
+
+  async stop(): Promise<void> {
+    if (!this.started || this.stopping) return
+    this.stopping = true
+
+    for (const timer of this.timers.values()) {
+      clearTimeout(timer)
+    }
+    this.timers.clear()
+
+    const pending = [...this.inFlight.values()]
+    if (pending.length > 0) {
+      this.log.info(`[scheduler] draining ${pending.length} in-flight tick(s)`)
+      const drained = await this.withDeadline(
+        Promise.allSettled(pending),
+        this.config.shutdownTimeoutMs
+      )
+
+      if (!drained) {
+        this.log.warn(
+          `[scheduler] shutdown deadline (${this.config.shutdownTimeoutMs}ms) reached; ` +
+            'remaining leases expire on their own'
+        )
+      }
+    }
+
+    this.started = false
+    this.log.info('[scheduler] stopped')
+  }
+
+  private schedule(queue: ScheduledQueue, delayMs: number): void {
+    const timer = setTimeout(() => {
+      void this.tick(queue)
+    }, delayMs)
+
+    this.timers.set(queue.name, timer)
+  }
+
+  private async tick(queue: ScheduledQueue): Promise<void> {
+    this.timers.delete(queue.name)
+    if (this.stopping) return
+
+    const pending = this.runQueue(queue)
+    this.inFlight.set(queue.name, pending)
+
+    try {
+      await pending
+    } finally {
+      this.inFlight.delete(queue.name)
+    }
+
+    if (!this.stopping) {
+      this.schedule(queue, this.config.intervalFor(queue.name))
+    }
+  }
+
+  private async runQueue(queue: ScheduledQueue): Promise<TickOutcome> {
+    const leaseMs = this.config.leaseFor(queue.name)
+    const before = await this.inspect(queue)
+    const lagMs = before.oldestDueAt
+      ? Math.max(0, Date.now() - before.oldestDueAt.getTime())
+      : 0
+
+    let lease
+    try {
+      lease = await this.leaseService.acquireQueueLease({
+        queueName: queue.name,
+        leaseMs,
+        owner: this.config.ownerId,
+      })
+    } catch (error) {
+      return this.record(queue, 'failed', 0, before, lagMs, error)
+    }
+
+    if (!lease) {
+      return this.record(queue, 'skipped', 0, before, lagMs)
+    }
+
+    const heartbeat = setInterval(() => {
+      void this.leaseService
+        .renewQueueLease(queue.name, lease.leaseToken, leaseMs)
+        .catch(error =>
+          this.log.warn(`[scheduler] failed to renew lease for "${queue.name}"`, error)
+        )
+    }, Math.max(1_000, Math.floor(leaseMs / 2)))
+
+    const startedAt = Date.now()
+
+    try {
+      await queue.drain()
+
+      return this.record(queue, 'ran', Date.now() - startedAt, before, lagMs)
+    } catch (error) {
+      return this.record(queue, 'failed', Date.now() - startedAt, before, lagMs, error)
+    } finally {
+      clearInterval(heartbeat)
+      await this.leaseService
+        .releaseQueueLease(queue.name, lease.leaseToken)
+        .catch(error =>
+          this.log.warn(`[scheduler] failed to release lease for "${queue.name}"`, error)
+        )
+    }
+  }
+
+  private async inspect(queue: ScheduledQueue): Promise<QueueDepthSnapshot> {
+    try {
+      return await queue.inspect()
+    } catch (error) {
+      this.log.warn(`[scheduler] depth probe failed for "${queue.name}"`, error)
+
+      return EMPTY_DEPTH
+    }
+  }
+
+  private record(
+    queue: ScheduledQueue,
+    outcome: TickOutcome,
+    durationMs: number,
+    depth: QueueDepthSnapshot,
+    lagMs: number,
+    error?: unknown
+  ): TickOutcome {
+    this.metrics.record({
+      queue: queue.name,
+      outcome,
+      durationMs,
+      depth: depth.depth,
+      due: depth.due,
+      lagMs,
+      error: error === undefined ? undefined : toMessage(error),
+    })
+
+    return outcome
+  }
+
+  private async withDeadline(work: Promise<unknown>, timeoutMs: number): Promise<boolean> {
+    let timer: NodeJS.Timeout | undefined
+
+    const deadline = new Promise<false>(resolve => {
+      timer = setTimeout(() => resolve(false), timeoutMs)
+    })
+
+    try {
+      return await Promise.race([work.then(() => true), deadline])
+    } finally {
+      if (timer) clearTimeout(timer)
+    }
+  }
+}
+
+function toMessage(error: unknown): string {
+  return error instanceof Error ? error.message : String(error)
+}
+
+export function createScheduledJobRunner(
+  overrides: Partial<ScheduledJobRunnerOptions> & { prisma?: PrismaClient } = {}
+): ScheduledJobRunner {
+  const prisma = overrides.prisma ?? defaultPrisma
+
+  return new ScheduledJobRunner({
+    queues: overrides.queues ?? createDefaultQueues({ prisma }),
+    leaseService: overrides.leaseService ?? createJobLeaseService(prisma),
+    config: overrides.config,
+    metrics: overrides.metrics,
+    log: overrides.log,
+  })
+}
diff --git a/src/workers/scheduler.worker.ts b/src/workers/scheduler.worker.ts
new file mode 100644
index 00000000..47660c5d
--- /dev/null
+++ b/src/workers/scheduler.worker.ts
@@ -0,0 +1,62 @@
+import 'dotenv/config'
+import prisma from '../config/database'
+import { schedulerConfig } from '../config/scheduler'
+import logger from '../utils/logger'
+import { createScheduledJobRunner } from './scheduled-job-runner'
+
+const runner = createScheduledJobRunner({ prisma })
+
+let isShuttingDown = false
+
+async function gracefulShutdown(signal: string): Promise<void> {
+  if (isShuttingDown) {
+    logger.warn('[scheduler] shutdown already in progress, ignoring additional signal')
+
+    return
+  }
+
+  isShuttingDown = true
+  logger.info(`[scheduler] received ${signal}, starting graceful shutdown...`)
+
+  const forceExit = setTimeout(() => {
+    logger.error('[scheduler] shutdown deadline exceeded, forcing exit')
+    process.exit(1)
+  }, schedulerConfig.shutdownTimeoutMs + 5_000)
+
+  try {
+    await runner.stop()
+    await prisma.$disconnect()
+    clearTimeout(forceExit)
+    logger.info('[scheduler] graceful shutdown completed')
+    process.exit(0)
+  } catch (error) {
+    logger.error('[scheduler] error during graceful shutdown:', error)
+    clearTimeout(forceExit)
+    process.exit(1)
+  }
+}
+
+process.on('SIGTERM', () => void gracefulShutdown('SIGTERM'))
+process.on('SIGINT', () => void gracefulShutdown('SIGINT'))
+
+process.on('uncaughtException', (error: Error) => {
+  logger.error('[scheduler] uncaught exception:', error)
+  void gracefulShutdown('uncaughtException')
+})
+
+process.on('unhandledRejection', (reason: unknown) => {
+  logger.error('[scheduler] unhandled rejection:', reason)
+  void gracefulShutdown('unhandledRejection')
+})
+
+logger.info(
+  `[scheduler] starting runner ${schedulerConfig.ownerId} ` +
+    `(base interval ${schedulerConfig.intervalMs}ms, lease ${schedulerConfig.leaseMs}ms)`
+)
+
+runner.start()
+
+if (runner.registeredQueues.length === 0) {
+  logger.error('[scheduler] no queues registered; check SCHEDULER_QUEUES / SCHEDULER_DISABLED_QUEUES')
+  process.exit(1)
+}
diff --git a/tests/email.service.test.ts b/tests/email.service.test.ts
index 712a2148..e4a27b04 100644
--- a/tests/email.service.test.ts
+++ b/tests/email.service.test.ts
@@ -29,7 +29,7 @@ describe('EmailService', () => {
     })
 
     describe('queueEmail', () => {
-        it('should create an email delivery record and trigger queue processing', async () => {
+        it('should create a pending email delivery record without draining the queue', async () => {
             const mockDelivery = {
                 id: 'del1',
                 userId: 'user1',
@@ -63,6 +63,7 @@ describe('EmailService', () => {
                 })
             )
             expect(result).toEqual(mockDelivery)
+            expect(prisma.emailDelivery.findMany).not.toHaveBeenCalled()
         })
     })
 
diff --git a/tests/lib/job-lease-queue.test.ts b/tests/lib/job-lease-queue.test.ts
new file mode 100644
index 00000000..60f90207
--- /dev/null
+++ b/tests/lib/job-lease-queue.test.ts
@@ -0,0 +1,106 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import { JobLeaseService } from '../../src/lib/transactions/job-lease.service'
+
+function sqlOf(call: unknown[]): string {
+  const [strings] = call as [TemplateStringsArray]
+
+  return strings.join('?')
+}
+
+describe('JobLeaseService queue leases', () => {
+  let queryRaw: ReturnType<typeof vi.fn>
+  let executeRaw: ReturnType<typeof vi.fn>
+  let service: JobLeaseService
+
+  beforeEach(() => {
+    queryRaw = vi.fn()
+    executeRaw = vi.fn()
+    service = new JobLeaseService({ $queryRaw: queryRaw, $executeRaw: executeRaw } as any)
+  })
+
+  describe('acquireQueueLease', () => {
+    it('returns a token when the conditional upsert claims the queue', async () => {
+      const leasedUntil = new Date(Date.now() + 60_000)
+      queryRaw.mockResolvedValue([{ leasedUntil }])
+
+      const lease = await service.acquireQueueLease({
+        queueName: 'email',
+        leaseMs: 60_000,
+        owner: 'scheduler@host:1',
+      })
+
+      expect(lease).not.toBeNull()
+      expect(lease!.queueName).toBe('email')
+      expect(lease!.leasedUntil).toBe(leasedUntil)
+      expect(lease!.leaseToken).toMatch(/^[0-9a-f-]{36}$/)
+    })
+
+    it('returns null when another holder still owns the lease', async () => {
+      queryRaw.mockResolvedValue([])
+
+      const lease = await service.acquireQueueLease({ queueName: 'email' })
+
+      expect(lease).toBeNull()
+    })
+
+    it('gates the upsert on the database clock, not the caller clock', async () => {
+      queryRaw.mockResolvedValue([{ leasedUntil: new Date() }])
+
+      await service.acquireQueueLease({ queueName: 'webhook', leaseMs: 30_000 })
+
+      const sql = sqlOf(queryRaw.mock.calls[0])
+      expect(sql).toContain('ON CONFLICT ("queueName") DO UPDATE')
+      expect(sql).toContain('"queue_leases"."leasedUntil" IS NULL')
+      expect(sql).toContain('"queue_leases"."leasedUntil" <= now()')
+
+      const values = queryRaw.mock.calls[0].slice(1)
+      expect(values).toContain('webhook')
+      expect(values).toContain('30000')
+    })
+
+    it('issues a distinct token per acquisition', async () => {
+      queryRaw.mockResolvedValue([{ leasedUntil: new Date() }])
+
+      const first = await service.acquireQueueLease({ queueName: 'email' })
+      const second = await service.acquireQueueLease({ queueName: 'email' })
+
+      expect(first!.leaseToken).not.toBe(second!.leaseToken)
+    })
+  })
+
+  describe('renewQueueLease', () => {
+    it('reports success only when the row still carries this token', async () => {
+      executeRaw.mockResolvedValueOnce(1)
+      await expect(service.renewQueueLease('email', 'token-a', 5_000)).resolves.toBe(true)
+
+      executeRaw.mockResolvedValueOnce(0)
+      await expect(service.renewQueueLease('email', 'stale-token')).resolves.toBe(false)
+
+      const values = executeRaw.mock.calls[0].slice(1)
+      expect(values).toContain('email')
+      expect(values).toContain('token-a')
+    })
+  })
+
+  describe('releaseQueueLease', () => {
+    it('clears the lease scoped to the holding token', async () => {
+      executeRaw.mockResolvedValue(1)
+
+      await expect(service.releaseQueueLease('data-export', 'token-a')).resolves.toBe(true)
+
+      const sql = sqlOf(executeRaw.mock.calls[0])
+      expect(sql).toContain('"leaseToken" = NULL')
+      expect(sql).toContain('"lastTickAt" = now()')
+      expect(sql).toContain('AND "leaseToken" =')
+
+      const values = executeRaw.mock.calls[0].slice(1)
+      expect(values).toEqual(['data-export', 'token-a'])
+    })
+
+    it('does not release a lease a successor now holds', async () => {
+      executeRaw.mockResolvedValue(0)
+
+      await expect(service.releaseQueueLease('data-export', 'stale')).resolves.toBe(false)
+    })
+  })
+})
diff --git a/tests/notification.service.test.ts b/tests/notification.service.test.ts
index fb03b14f..a11f2da3 100644
--- a/tests/notification.service.test.ts
+++ b/tests/notification.service.test.ts
@@ -2,20 +2,25 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
 import { NotificationService } from '../src/services/notification.service'
 
 // Use vi.hoisted to mock dependencies before they are imported by the service
-const { mockSendEachForMulticast } = vi.hoisted(() => ({
-  mockSendEachForMulticast: vi.fn().mockResolvedValue({ failureCount: 0, responses: [] })
-}))
+const { mockSendEachForMulticast, mockAdmin } = vi.hoisted(() => {
+  const mockSendEachForMulticast = vi.fn().mockResolvedValue({ failureCount: 0, responses: [] })
+
+  return {
+    mockSendEachForMulticast,
+    mockAdmin: {
+      apps: [{ name: 'mock-app' }],
+      initializeApp: vi.fn(),
+      credential: {
+        cert: vi.fn().mockReturnValue({})
+      },
+      messaging: vi.fn().mockReturnValue({
+        sendEachForMulticast: mockSendEachForMulticast
+      })
+    }
+  }
+})
 
-vi.mock('firebase-admin', () => ({
-  apps: [{ name: 'mock-app' }],
-  initializeApp: vi.fn(),
-  credential: {
-    cert: vi.fn().mockReturnValue({})
-  },
-  messaging: vi.fn().mockReturnValue({
-    sendEachForMulticast: mockSendEachForMulticast
-  })
-}))
+vi.mock('firebase-admin', () => ({ ...mockAdmin, default: mockAdmin }))
 
 // Use vi.hoisted to ensure these are available for vi.mock
 const { mockPrisma } = vi.hoisted(() => ({
diff --git a/tests/services/webhook.service.spec.ts b/tests/services/webhook.service.spec.ts
index 25f98fd3..fd1fc78b 100644
--- a/tests/services/webhook.service.spec.ts
+++ b/tests/services/webhook.service.spec.ts
@@ -72,11 +72,12 @@ describe('WebhookService', () => {
                 { id: 'ep1', url: 'https://ep1.com', secret: 's1', events: 'module.completed', isActive: true },
             ])
             mockPrismaInstance.webhookDelivery.create.mockResolvedValue({ id: 'd1' })
-            mockPrismaInstance.webhookDelivery.findMany.mockResolvedValue([]) // for processQueue
+            mockPrismaInstance.webhookDelivery.findMany.mockResolvedValue([])
 
             await service.queueEvent('module.completed', { foo: 'bar' })
 
             expect(mockPrismaInstance.webhookDelivery.create).toHaveBeenCalledOnce()
+            expect(mockPrismaInstance.webhookDelivery.findMany).not.toHaveBeenCalled()
             const createCall = mockPrismaInstance.webhookDelivery.create.mock.calls[0][0]
             expect(createCall.data.eventType).toBe('module.completed')
             expect(JSON.parse(createCall.data.payload).data).toEqual({ foo: 'bar' })
diff --git a/tests/workers/scheduled-job-runner.test.ts b/tests/workers/scheduled-job-runner.test.ts
new file mode 100644
index 00000000..2a2e6d0a
--- /dev/null
+++ b/tests/workers/scheduled-job-runner.test.ts
@@ -0,0 +1,374 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import type { SchedulerConfig } from '../../src/config/scheduler'
+import { QueueMetricsRegistry } from '../../src/workers/queue-metrics'
+import {
+  ScheduledJobRunner,
+  type QueueLeaseApi,
+} from '../../src/workers/scheduled-job-runner'
+import type { ScheduledQueue } from '../../src/workers/queue-registry'
+
+vi.mock('../../src/utils/logger', () => ({
+  default: {
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    debug: vi.fn(),
+  },
+}))
+
+const silentLog = {
+  info: vi.fn(),
+  warn: vi.fn(),
+  error: vi.fn(),
+  debug: vi.fn(),
+}
+
+function testConfig(overrides: Partial<SchedulerConfig> = {}): SchedulerConfig {
+  return {
+    intervalMs: 10,
+    leaseMs: 2_000,
+    shutdownTimeoutMs: 1_000,
+    inProcess: false,
+    only: [],
+    disabled: [],
+    ownerId: 'test-runner',
+    isEnabled(queueName: string) {
+      return (
+        !this.disabled.includes(queueName) &&
+        (this.only.length === 0 || this.only.includes(queueName))
+      )
+    },
+    intervalFor() {
+      return this.intervalMs
+    },
+    leaseFor() {
+      return this.leaseMs
+    },
+    ...overrides,
+  } as SchedulerConfig
+}
+
+class FakeLeaseStore implements QueueLeaseApi {
+  private readonly rows = new Map<string, { token: string; until: number }>()
+  private seq = 0
+
+  acquireQueueLease = vi.fn(
+    async (options: { queueName: string; leaseMs?: number; owner?: string }) => {
+      const now = Date.now()
+      const leaseMs = options.leaseMs ?? 60_000
+      const current = this.rows.get(options.queueName)
+
+      if (current && current.until > now) {
+        return null
+      }
+
+      const leaseToken = `${options.owner ?? 'anon'}-${++this.seq}`
+      this.rows.set(options.queueName, { token: leaseToken, until: now + leaseMs })
+
+      return {
+        queueName: options.queueName,
+        leaseToken,
+        leasedUntil: new Date(now + leaseMs),
+      }
+    }
+  )
+
+  renewQueueLease = vi.fn(async (queueName: string, leaseToken: string, leaseMs = 60_000) => {
+    const current = this.rows.get(queueName)
+    if (!current || current.token !== leaseToken) return false
+    current.until = Date.now() + leaseMs
+
+    return true
+  })
+
+  releaseQueueLease = vi.fn(async (queueName: string, leaseToken: string) => {
+    const current = this.rows.get(queueName)
+    if (!current || current.token !== leaseToken) return false
+    this.rows.delete(queueName)
+
+    return true
+  })
+
+  hold(queueName: string, forMs: number): void {
+    this.rows.set(queueName, { token: 'foreign-holder', until: Date.now() + forMs })
+  }
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise<void> {
+  const deadline = Date.now() + timeoutMs
+  while (!predicate()) {
+    if (Date.now() > deadline) {
+      throw new Error('waitFor timed out')
+    }
+    await new Promise(resolve => setTimeout(resolve, 5))
+  }
+}
+
+interface FakeRow {
+  id: string
+  nextAttemptAt: Date
+}
+
+function fakeQueue(
+  name: string,
+  rows: FakeRow[],
+  processed: string[],
+  drainImpl?: () => Promise<void>
+): ScheduledQueue {
+  return {
+    name,
+    drain:
+      drainImpl ??
+      (async () => {
+        const now = Date.now()
+        const due = rows.filter(row => row.nextAttemptAt.getTime() <= now)
+        for (const row of due) {
+          rows.splice(rows.indexOf(row), 1)
+          processed.push(row.id)
+        }
+      }),
+    inspect: async () => {
+      const now = Date.now()
+      const due = rows.filter(row => row.nextAttemptAt.getTime() <= now)
+      const oldest = [...due].sort(
+        (a, b) => a.nextAttemptAt.getTime() - b.nextAttemptAt.getTime()
+      )[0]
+
+      return {
+        depth: rows.length,
+        due: due.length,
+        oldestDueAt: oldest?.nextAttemptAt ?? null,
+      }
+    },
+  }
+}
+
+describe('ScheduledJobRunner', () => {
+  let leases: FakeLeaseStore
+  let metrics: QueueMetricsRegistry
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    leases = new FakeLeaseStore()
+    metrics = new QueueMetricsRegistry()
+  })
+
+  it('drains due rows on a timer with no inbound traffic', async () => {
+    const processed: string[] = []
+    const rows: FakeRow[] = [{ id: 'row-1', nextAttemptAt: new Date(Date.now() - 1) }]
+
+    const runner = new ScheduledJobRunner({
+      queues: [fakeQueue('email', rows, processed)],
+      leaseService: leases,
+      config: testConfig(),
+      metrics,
+      log: silentLog,
+    })
+
+    runner.start()
+    await waitFor(() => processed.length === 1)
+    await runner.stop()
+
+    expect(processed).toEqual(['row-1'])
+  })
+
+  it('picks up a row whose nextAttemptAt falls due within one interval', async () => {
+    const processed: string[] = []
+    const rows: FakeRow[] = [{ id: 'retry-1', nextAttemptAt: new Date(Date.now() + 40) }]
+
+    const runner = new ScheduledJobRunner({
+      queues: [fakeQueue('email', rows, processed)],
+      leaseService: leases,
+      config: testConfig({ intervalMs: 10 }),
+      metrics,
+      log: silentLog,
+    })
+
+    runner.start()
+
+    await new Promise(resolve => setTimeout(resolve, 15))
+    expect(processed).toEqual([])
+
+    await waitFor(() => processed.length === 1)
+    await runner.stop()
+
+    expect(processed).toEqual(['retry-1'])
+  })
+
+  it('skips a tick when another holder owns the queue lease', async () => {
+    const processed: string[] = []
+    const rows: FakeRow[] = [{ id: 'row-1', nextAttemptAt: new Date(Date.now() - 1) }]
+    leases.hold('email', 60_000)
+
+    const runner = new ScheduledJobRunner({
+      queues: [fakeQueue('email', rows, processed)],
+      leaseService: leases,
+      config: testConfig(),
+      metrics,
+      log: silentLog,
+    })
+
+    const outcomes = await runner.runOnce()
+
+    expect(outcomes.email).toBe('skipped')
+    expect(processed).toEqual([])
+    expect(metrics.snapshot()[0].skipped).toBe(1)
+    expect(metrics.snapshot()[0].attempts).toBe(0)
+  })
+
+  it('never lets two replicas process the same row twice', async () => {
+    const processed: string[] = []
+    const rows: FakeRow[] = Array.from({ length: 25 }, (_, index) => ({
+      id: `row-${index}`,
+      nextAttemptAt: new Date(Date.now() - 1),
+    }))
+
+    const slowDrain = async () => {
+      const now = Date.now()
+      const due = rows.filter(row => row.nextAttemptAt.getTime() <= now)
+      for (const row of due) {
+        const index = rows.indexOf(row)
+        if (index === -1) continue
+        rows.splice(index, 1)
+        await new Promise(resolve => setTimeout(resolve, 1))
+        processed.push(row.id)
+      }
+    }
+
+    const makeRunner = (owner: string) =>
+      new ScheduledJobRunner({
+        queues: [fakeQueue('email', rows, processed, slowDrain)],
+        leaseService: leases,
+        config: testConfig({ ownerId: owner, intervalMs: 5 }),
+        metrics: new QueueMetricsRegistry(),
+        log: silentLog,
+      })
+
+    const replicaA = makeRunner('replica-a')
+    const replicaB = makeRunner('replica-b')
+
+    replicaA.start()
+    replicaB.start()
+
+    await waitFor(() => processed.length === 25, 5_000)
+    await Promise.all([replicaA.stop(), replicaB.stop()])
+
+    expect(new Set(processed).size).toBe(25)
+    expect(processed).toHaveLength(25)
+  })
+
+  it('releases the lease and records a failure when a drain throws', async () => {
+    const failing: ScheduledQueue = {
+      name: 'webhook',
+      drain: async () => {
+        throw new Error('provider down')
+      },
+      inspect: async () => ({ depth: 3, due: 3, oldestDueAt: new Date(Date.now() - 5_000) }),
+    }
+
+    const runner = new ScheduledJobRunner({
+      queues: [failing],
+      leaseService: leases,
+      config: testConfig(),
+      metrics,
+      log: silentLog,
+    })
+
+    const outcomes = await runner.runOnce()
+
+    expect(outcomes.webhook).toBe('failed')
+    expect(leases.releaseQueueLease).toHaveBeenCalledOnce()
+
+    const [snapshot] = metrics.snapshot()
+    expect(snapshot.failures).toBe(1)
+    expect(snapshot.lastError).toBe('provider down')
+
+    expect(await runner.runOnce()).toEqual({ webhook: 'failed' })
+    expect(leases.acquireQueueLease).toHaveBeenCalledTimes(2)
+  })
+
+  it('emits depth, due, attempt, failure and lag metrics per queue', async () => {
+    const oldestDueAt = new Date(Date.now() - 30_000)
+    const queue: ScheduledQueue = {
+      name: 'notification',
+      drain: async () => undefined,
+      inspect: async () => ({ depth: 7, due: 4, oldestDueAt }),
+    }
+
+    const runner = new ScheduledJobRunner({
+      queues: [queue],
+      leaseService: leases,
+      config: testConfig(),
+      metrics,
+      log: silentLog,
+    })
+
+    await runner.runOnce()
+
+    const [snapshot] = metrics.snapshot()
+    expect(snapshot.queue).toBe('notification')
+    expect(snapshot.depth).toBe(7)
+    expect(snapshot.due).toBe(4)
+    expect(snapshot.attempts).toBe(1)
+    expect(snapshot.failures).toBe(0)
+    expect(snapshot.lagMs).toBeGreaterThanOrEqual(30_000)
+    expect(snapshot.lastRunAt).not.toBeNull()
+  })
+
+  it('drains the in-flight tick and releases its lease on shutdown', async () => {
+    let started = false
+
+    const queue: ScheduledQueue = {
+      name: 'data-export',
+      drain: async () => {
+        started = true
+        await new Promise(resolve => setTimeout(resolve, 60))
+      },
+      inspect: async () => ({ depth: 1, due: 1, oldestDueAt: null }),
+    }
+
+    const runner = new ScheduledJobRunner({
+      queues: [queue],
+      leaseService: leases,
+      config: testConfig(),
+      metrics,
+      log: silentLog,
+    })
+
+    runner.start()
+    await waitFor(() => started)
+
+    await runner.stop()
+
+    expect(leases.releaseQueueLease).toHaveBeenCalledOnce()
+    expect(await leases.acquireQueueLease({ queueName: 'data-export' })).not.toBeNull()
+
+    const ticksAtStop = metrics.snapshot()[0].attempts
+    await new Promise(resolve => setTimeout(resolve, 40))
+    expect(metrics.snapshot()[0].attempts).toBe(ticksAtStop)
+  })
+
+  it('honours the disabled and allow lists when registering queues', () => {
+    const queues = [
+      fakeQueue('email', [], []),
+      fakeQueue('webhook', [], []),
+      fakeQueue('notification', [], []),
+    ]
+
+    const disabled = new ScheduledJobRunner({
+      queues,
+      leaseService: leases,
+      config: testConfig({ disabled: ['webhook'] }),
+      log: silentLog,
+    })
+    expect(disabled.registeredQueues).toEqual(['email', 'notification'])
+
+    const only = new ScheduledJobRunner({
+      queues,
+      leaseService: leases,
+      config: testConfig({ only: ['notification'] }),
+      log: silentLog,
+    })
+    expect(only.registeredQueues).toEqual(['notification'])
+  })
+})

From 8dd4e815d1071b6c6a6c78da44e2e76dc72fc695 Mon Sep 17 00:00:00 2001
From: Ezeh <chijioke@clivealliance.com>
Date: Sun, 30 Aug 2026 00:53:43 +0100
Subject: [PATCH 2/2] chore: added a bg job verification script

---
 scripts/scheduler-verification.sh | 187 +++++++++++++-----------------
 1 file changed, 81 insertions(+), 106 deletions(-)

diff --git a/scripts/scheduler-verification.sh b/scripts/scheduler-verification.sh
index 822c557d..df436b8b 100755
--- a/scripts/scheduler-verification.sh
+++ b/scripts/scheduler-verification.sh
@@ -1,126 +1,101 @@
 #!/usr/bin/env bash
-set -euo pipefail
+set -uo pipefail
 
-COMPOSE_FILE="${COMPOSE_FILE:-docker-compose.yml}"
-POSTGRES_USER="${POSTGRES_USER:-learnault}"
-POSTGRES_DB="${POSTGRES_DB:-learnault_dev}"
-INTERVAL_MS="${SCHEDULER_INTERVAL_MS:-15000}"
-BATCH_SIZE="${BATCH_SIZE:-40}"
+command -v docker >/dev/null 2>&1 || export PATH="$PATH:/c/Program Files/Docker/Docker/resources/bin"
 
-dc() { docker compose -f "$COMPOSE_FILE" "$@"; }
-psql_q() { dc exec -T db psql -U "$POSTGRES_USER" -d "$POSTGRES_DB" -At -c "$1" | tr -d '\r'; }
+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
 
-wait_seconds=$(( (INTERVAL_MS / 1000) * 3 + 5 ))
+q() { docker compose exec -T db psql -U "$PGUSER_" -d "$PGDB_" -qAt -c "$1" | tr -d '\r'; }
 
-echo "==> Bringing up the stack (LOG_LEVEL=debug)"
-dc up -d --build
+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
 
-echo "==> Waiting for the database"
-until dc exec -T db pg_isready -U "$POSTGRES_USER" -d "$POSTGRES_DB" >/dev/null 2>&1; do
-  sleep 2
-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
 
-seed_user() {
-  psql_q "
-    INSERT INTO users (id, email, username, password, role, \"isVerified\", status, \"createdAt\", \"updatedAt\")
-    VALUES (gen_random_uuid(), 'scheduler-evidence@example.com', 'scheduler-evidence', 'x', 'LEARNER', true, 'ACTIVE', now(), now())
-    ON CONFLICT (email) DO UPDATE SET \"updatedAt\" = now()
-    RETURNING id;
-  "
-}
+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 "# 1. Idle instance: a due retry drains with no HTTP traffic #"
-echo "############################################################"
-
-USER_ID="$(seed_user)"
+echo "============================================================"
+echo " SCENARIO 1 — idle instance: failed delivery retried on time"
+echo "============================================================"
 
-psql_q "DELETE FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
+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
 
-psql_q "
-  INSERT INTO email_deliveries
-    (id, \"userId\", \"to\", subject, body, type, status, error, \"attemptCount\", \"maxAttempts\", \"nextAttemptAt\", \"createdAt\", \"updatedAt\")
-  VALUES
-    (gen_random_uuid(), '$USER_ID', 'idle@example.com', 'idle-instance retry', '<p>evidence</p>',
-     'SCHEDULER_EVIDENCE', 'pending', 'previous attempt failed', 1, 5, now() - interval '1 minute', now(), now());
-"
+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';")"
 
-echo "--> Before (no HTTP traffic will be sent to the API):"
-psql_q "SELECT status, \"attemptCount\", \"nextAttemptAt\" FROM email_deliveries WHERE type = 'SCHEDULER_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
 
-echo "--> Waiting ${wait_seconds}s (SCHEDULER_INTERVAL_MS=${INTERVAL_MS})"
-sleep "$wait_seconds"
+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
 
-echo "--> After:"
-psql_q "SELECT status, \"attemptCount\", \"sentAt\" FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
+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;")"
 
-echo "--> Scheduler log lines for the email queue:"
-dc logs --no-log-prefix scheduler | grep -E '"queue": ?"email"' | tail -5 || true
-
-IDLE_STATUS="$(psql_q "SELECT status FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE' LIMIT 1;")"
-if [ "$IDLE_STATUS" != "sent" ]; then
-  echo "!! Expected the due row to be drained, got status='$IDLE_STATUS'" >&2
-  dc logs scheduler | tail -40
-  exit 1
-fi
-echo "✅ Due row drained on schedule with no inbound HTTP traffic"
+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 "# 2. Two replicas: due rows are processed exactly once         #"
-echo "###############################################################"
-
-echo "--> Scaling the scheduler to 2 replicas"
-dc up -d --scale scheduler=2
-sleep 5
-
-psql_q "DELETE FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
-psql_q "
-  INSERT INTO email_deliveries
-    (id, \"userId\", \"to\", subject, body, type, status, \"attemptCount\", \"maxAttempts\", \"nextAttemptAt\", \"createdAt\", \"updatedAt\")
-  SELECT gen_random_uuid(), '$USER_ID', 'replica-' || g || '@example.com', 'replica batch ' || g, '<p>evidence</p>',
-         'SCHEDULER_EVIDENCE', 'pending', 0, 5, now() - interval '1 minute', now(), now()
-  FROM generate_series(1, $BATCH_SIZE) AS g;
-"
-
-echo "--> Queued $BATCH_SIZE due rows across 2 replicas; waiting ${wait_seconds}s"
-sleep "$wait_seconds"
-
-echo "--> Attempt counts (a row drained twice would show attemptCount > 1):"
-psql_q "
-  SELECT \"attemptCount\", count(*)
-  FROM email_deliveries
-  WHERE type = 'SCHEDULER_EVIDENCE'
-  GROUP BY \"attemptCount\"
-  ORDER BY \"attemptCount\";
-"
-
-echo "--> Skipped ticks (the replica that lost the lease race):"
-SKIPPED="$(dc logs --no-log-prefix scheduler | grep -c 'queue tick skipped' || true)"
-echo "    $SKIPPED skipped tick(s) logged"
-
-DUPLICATES="$(psql_q "SELECT count(*) FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE' AND \"attemptCount\" > 1;")"
-PROCESSED="$(psql_q "SELECT count(*) FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE' AND status = 'sent';")"
-
-echo "--> processed=$PROCESSED duplicates=$DUPLICATES"
-
-if [ "$DUPLICATES" != "0" ]; then
-  echo "!! $DUPLICATES row(s) were processed more than once" >&2
-  exit 1
-fi
-if [ "$PROCESSED" != "$BATCH_SIZE" ]; then
-  echo "!! Expected $BATCH_SIZE processed rows, got $PROCESSED" >&2
-  exit 1
+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 "✅ Two replicas processed $BATCH_SIZE rows with no duplicates"
-
-echo ""
-echo "==> Cleaning up evidence rows"
-psql_q "DELETE FROM email_deliveries WHERE type = 'SCHEDULER_EVIDENCE';"
-dc up -d --scale scheduler=1
-
-echo ""
-echo "✅ Scheduler verification passed"
+echo "RESULT         : FAIL — processed=$DONE_ duplicates=$DUPE_"
+exit 1