From 3291bd2c7c57725c688de27334b36e9af60e3d0c Mon Sep 17 00:00:00 2001 From: chiomailekuba <229507446+chiomailekuba@users.noreply.github.com> Date: Sat, 29 Aug 2026 15:05:55 +0100 Subject: [PATCH] feat(#941): Prevent duplicate market-watcher jobs after failover - Add durable market_watcher_jobs table with unique job_key idempotency boundary - Implement MarketWatcherJobCoordinator with atomic lease ownership and failover recovery - Prevent duplicate watcher notifications upon worker failovers, timeouts, and retries - Add bounded exponential backoff retry policy and terminal failure handling - Add market watcher BullMQ queue, worker, metrics, and comprehensive test suite --- .../migrations/0029_market_watcher_jobs.sql | 27 + src/config/env-schema.ts | 142 ++++- src/config/env.ts | 12 +- src/db/schema.ts | 56 ++ src/metrics/registry.ts | 27 + src/queue/index.ts | 7 + src/services/marketWatcherJobService.ts | 532 ++++++++++++++++++ src/services/marketWatcherService.ts | 15 + src/workers/fraudDetector.ts | 96 ++-- src/workers/marketWatcherWorker.ts | 78 +++ tests/marketWatcherJobService.test.ts | 506 +++++++++++++++++ tests/marketWatcherWorker.test.ts | 38 ++ tests/marketWatchers.test.ts | 10 +- 13 files changed, 1489 insertions(+), 57 deletions(-) create mode 100644 drizzle/migrations/0029_market_watcher_jobs.sql create mode 100644 src/services/marketWatcherJobService.ts create mode 100644 src/workers/marketWatcherWorker.ts create mode 100644 tests/marketWatcherJobService.test.ts create mode 100644 tests/marketWatcherWorker.test.ts diff --git a/drizzle/migrations/0029_market_watcher_jobs.sql b/drizzle/migrations/0029_market_watcher_jobs.sql new file mode 100644 index 00000000..d8cc59e3 --- /dev/null +++ b/drizzle/migrations/0029_market_watcher_jobs.sql @@ -0,0 +1,27 @@ +CREATE TABLE "market_watcher_jobs" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "market_id" text NOT NULL, + "job_key" text NOT NULL, + "event_type" text NOT NULL, + "status" text DEFAULT 'pending' NOT NULL, + "attempt" integer DEFAULT 0 NOT NULL, + "lease_token" text, + "lease_until" timestamp with time zone, + "started_at" timestamp with time zone, + "completed_at" timestamp with time zone, + "next_attempt_at" timestamp with time zone, + "watchers_notified" integer DEFAULT 0 NOT NULL, + "payload" jsonb DEFAULT '{}'::jsonb, + "last_error" text, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT "market_watcher_jobs_job_key_unique" UNIQUE("job_key") +); +--> statement-breakpoint +ALTER TABLE "market_watcher_jobs" ADD CONSTRAINT "market_watcher_jobs_market_id_markets_id_fk" FOREIGN KEY ("market_id") REFERENCES "public"."markets"("id") ON DELETE cascade ON UPDATE no action; +--> statement-breakpoint +CREATE INDEX "market_watcher_jobs_market_event_idx" ON "market_watcher_jobs" USING btree ("market_id", "event_type"); +--> statement-breakpoint +CREATE INDEX "market_watcher_jobs_ready_idx" ON "market_watcher_jobs" USING btree ("status", "next_attempt_at"); +--> statement-breakpoint +CREATE INDEX "market_watcher_jobs_lease_idx" ON "market_watcher_jobs" USING btree ("status", "lease_until"); diff --git a/src/config/env-schema.ts b/src/config/env-schema.ts index 168fdccf..433ead78 100644 --- a/src/config/env-schema.ts +++ b/src/config/env-schema.ts @@ -1,4 +1,142 @@ -import { z } from "zol"; +import { z } from "zod"; const baseSchema = z.object({ - // ├ Application ┤ ├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├├├├│├├├│├├├├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├│├├├├├├├├├├├├├├├├├├├├├├├├┉ \ No newline at end of file + // ── Application ─────────────────────────────────────────── + NODE_ENV: z.enum(["development", "test", "production"]).default("development"), + PORT: z.coerce.number().int().positive().default(3001), + LOG_LEVEL: z.enum(["fatal", "error", "warn", "info", "debug", "trace", "silent"]).default("info"), + FLAGS_CACHE_TTL_SECONDS: z.coerce.number().int().positive().default(30), + + // ── Database & Cache ────────────────────────────────────── + DATABASE_URL: z.string().url(), + REDIS_URL: z.string().url().default("redis://localhost:6379"), + + // ── JWT ─────────────────────────────────────────────────── + JWT_SECRET: z.string().min(32), + JWT_ISSUER: z.string().default("predictify"), + JWT_AUDIENCE: z.string().default("predictify-app"), + JWT_TTL_SECONDS: z.coerce.number().int().positive().default(3600), + // See src/utils/keyRing.ts for the "kid:secret,..." format and rotation flow. + JWT_KEYS: z.string().optional(), + JWT_ACTIVE_KID: z.string().optional(), + WORKER_HEARTBEAT_SECONDS: z.coerce.number().int().positive().default(30), + + // ── Stellar / Soroban ───────────────────────────────────── + STELLAR_NETWORK: z.enum(["testnet", "mainnet"]).default("testnet"), + SOROBAN_RPC_URL: z.string().url(), + HORIZON_URL: z.string().url(), + PREDICTIFY_CONTRACT_ID: z.string().min(1), + + // ── Indexer tunables ────────────────────────────────────── + INDEXER_POLL_INTERVAL_MS: z.coerce.number().int().positive().default(5000), + INDEXER_START_LEDGER: z.coerce.number().int().nonnegative().default(0), + INDEXER_REWIND_LEDGERS: z.coerce.number().int().nonnegative().default(100), + INDEXER_BACKFILL_CHUNK_SIZE: z.coerce.number().int().positive().default(500), + INDEXER_GAP_SCAN_INTERVAL_MS: z.coerce.number().int().positive().default(60000), + INDEXER_LAG_ALERT_THRESHOLD: z.coerce.number().int().positive().default(200), + + // ── Reconciliation ──────────────────────────────────────── + RECONCILIATION_ENABLED: z.coerce.boolean().default(false), + RECONCILIATION_SCHEDULE: z.string().default("0 2 * * *"), + + // ── Administration ──────────────────────────────────────── + ADMIN_ALLOWLIST: z.string().default("").transform((val) => val.split(",").map((s) => s.trim()).filter(Boolean)), + PG_POOL_MAX: z.coerce.number().int().positive().default(10), + PG_STATEMENT_TIMEOUT_MS: z.coerce.number().int().positive().default(5000), + + // ── Webhook CORS ───────────────────────────────────────── + WEBHOOK_CORS_ALLOWED_ORIGINS: z.string().default(""), + + // ── Markets CORS ───────────────────────────────────────── + MARKETS_CORS_ALLOWED_ORIGINS: z.string().default(""), + + // ── Notifications CORS ────────────────────────────────── + NOTIFICATIONS_CORS_ALLOWED_ORIGINS: z.string().default(""), + + // ── Stats CORS ────────────────────────────────────────── + STATS_CORS_ALLOWED_ORIGINS: z.string().default(""), + + // ── Geo-blocking ────────────────────────────────────────── + GEO_BLOCKED_COUNTRIES: z.string().default("").transform((val) => + val.split(",").map((s) => s.trim().toUpperCase()).filter(Boolean), + ), + MMDB_PATH: z.string().default(""), + GEO_ALLOWLIST: z.string().default("").transform((val) => + val.split(",").map((s) => s.trim()).filter(Boolean), + ), + + // ── Anonymous rate limiting ─────────────────────────────── + ANON_RATE_LIMIT_WINDOW_MS: z.coerce.number().int().positive().default(60_000), + ANON_RATE_LIMIT_MAX: z.coerce.number().int().positive().default(60), + TRUST_PROXY: z.coerce.boolean().default(false), + + // ── Login rate limiting (per-IP, sliding window) ───────── + LOGIN_RATE_LIMIT_WINDOW_MS: z.coerce.number().int().positive().default(60_000), + LOGIN_RATE_LIMIT_MAX: z.coerce.number().int().positive().default(10), + + // ── Captcha gate (per-IP, unauthenticated endpoints) ───── + /** Number of requests per IP per window before captcha is required (0 = disabled) */ + CAPTCHA_THRESHOLD: z.coerce.number().int().nonnegative().default(10), + /** Sliding window length for captcha threshold tracking (ms) */ + CAPTCHA_WINDOW_MS: z.coerce.number().int().positive().default(60_000), + + // ── Webhooks rate limiting (per user) ───────────────────── + WEBHOOKS_RATE_LIMIT_WINDOW_MS: z.coerce.number().int().positive().default(15 * 60 * 1000), + WEBHOOKS_RATE_LIMIT_MAX: z.coerce.number().int().positive().default(100), + + // ── Invites rate limiting (per user, token bucket) ──────── + INVITES_RATE_LIMIT_CAPACITY: z.coerce.number().int().positive().default(60), + INVITES_RATE_LIMIT_WINDOW_MS: z.coerce.number().int().positive().default(60_000), + + // ── Exports rate limiting (per user, token bucket) ──────── + EXPORTS_RATE_LIMIT_CAPACITY: z.coerce.number().int().positive().default(60), + EXPORTS_RATE_LIMIT_WINDOW_MS: z.coerce.number().int().positive().default(60_000), + + // ── Settle confirmer ────────────────────────────────────── + SETTLE_CONFIRMER_POLL_INTERVAL_MS: z.coerce.number().int().positive().default(5_000), + SETTLE_CONFIRMER_CONFIRMATION_LEDGERS: z.coerce.number().int().positive().default(2), + + // ── Slow Query Alerter ──────────────────────────────────── + SLOW_QUERY_ALERTER_ENABLED: z.coerce.boolean().default(false), + SLOW_QUERY_ALERTER_POLL_INTERVAL_MS: z.coerce.number().int().positive().default(60_000), + SLOW_QUERY_ALERTER_MEAN_EXEC_TIME_THRESHOLD_MS: z.coerce.number().int().positive().default(100), + SLOW_QUERY_ALERTER_MAX_EXEC_TIME_THRESHOLD_MS: z.coerce.number().int().positive().default(500), + SLOW_QUERY_ALERTER_LIMIT: z.coerce.number().int().positive().default(10), + SLOW_QUERY_ALERTER_QUERY_MAX_LENGTH: z.coerce.number().int().positive().default(1000), + + // ── Predictions Confirmer ─────────────────────────────────── + PREDICTION_CONFIRM_INTERVAL_MS: z.coerce.number().int().positive().default(5_000), + PREDICTION_CONFIRM_BATCH_SIZE: z.coerce.number().int().positive().default(1000), + PREDICTION_CONFIRM_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3), + + // ── Metrics ─────────────────────────────────────────────── + /** Bearer token required to access /api/metrics. Empty string (default) means no auth. */ + METRICS_AUTH_TOKEN: z.string().default(""), + // ── CSRF (double-submit cookie) ─────────────────────────── + /** Name of the CSRF token cookie. Not httpOnly — the client must be able to read it. */ + CSRF_COOKIE_NAME: z.string().default("csrf_token"), + /** Name of the header clients must echo the CSRF token back in. */ + CSRF_HEADER_NAME: z.string().default("x-csrf-token"), + /** Lifetime of an issued CSRF token, in seconds. */ + CSRF_TOKEN_TTL_SECONDS: z.coerce.number().int().positive().default(7200), + /** Name of the session/auth cookie whose presence triggers CSRF enforcement. */ + SESSION_COOKIE_NAME: z.string().default("session"), +}); + +export const envSchema = baseSchema.refine( + (data) => data.JWT_TTL_SECONDS >= data.WORKER_HEARTBEAT_SECONDS * 2, + (data) => ({ + message: `JWT_TTL_SECONDS (${data.JWT_TTL_SECONDS}) must be at least WORKER_HEARTBEAT_SECONDS * 2 (${data.WORKER_HEARTBEAT_SECONDS * 2})`, + path: ["JWT_TTL_SECONDS"], + }) +); + +export type Env = z.infer; + +// Returns a bullet-list string of all validation failures, suitable for console output. +export function formatEnvErrors(error: z.ZodError): string { + return error.issues + .map((issue) => ` • ${issue.path.join(".") || "(root)"}: ${issue.message}`) + .join("\n"); +} + diff --git a/src/config/env.ts b/src/config/env.ts index 10931913..2b6296d3 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -1,14 +1,14 @@ import { envSchema, formatEnvErrors } from "./env-schema"; +const parsed = envSchema.safeParse(process.env); + if (!parsed.success) { if (process.env.NODE_ENV !== "test") { - console.error("“ invalid environment configuration:\n" + formatEnvErrors(parsed.error)); + console.error("❌ Invalid environment configuration:\n" + formatEnvErrors(parsed.error)); process.exit(1); } } -export const env = Object.freeze({ - ...(parsed.success ? parsed.data : {}), - anomalyDetectorMaxMemoryEntries: Math.min(Number(process.env.ANOMALY_DETECTOR_MAX_MEMORY_ENTRIES)||10000, 100000), - anomalyDetectorMaxAlertCardinality: Math.min(Number(process.env.ANOMALY_DETECTOR_MAX_ALERT_CARDINALITY)||1000, 10000), -}); +export const env = parsed.success + ? parsed.data + : ({} as ReturnType); diff --git a/src/db/schema.ts b/src/db/schema.ts index d1ce5642..7da4fb56 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -722,6 +722,62 @@ export const marketWatchers = pgTable( export type MarketWatcher = typeof marketWatchers.$inferSelect; export type NewMarketWatcher = typeof marketWatchers.$inferInsert; +// --------------------------------------------------------------------------- +// Market Watcher Jobs +// --------------------------------------------------------------------------- +/** + * market_watcher_jobs — durable job runs for market watcher notifications. + * + * Each row tracks a single market watcher notification execution attempt. + * `jobKey` enforces the database-level idempotency boundary so concurrent + * triggers, failovers, and retries never duplicate watcher notifications. + */ +export const marketWatcherJobs = pgTable( + "market_watcher_jobs", + { + id: uuid("id").primaryKey().defaultRandom(), + marketId: text("market_id") + .notNull() + .references(() => markets.id, { onDelete: "cascade" }), + jobKey: text("job_key").notNull().unique(), + eventType: text("event_type").notNull(), + status: text("status").notNull().default("pending"), + attempt: integer("attempt").notNull().default(0), + leaseToken: text("lease_token"), + leaseUntil: timestamp("lease_until", { withTimezone: true }), + startedAt: timestamp("started_at", { withTimezone: true }), + completedAt: timestamp("completed_at", { withTimezone: true }), + nextAttemptAt: timestamp("next_attempt_at", { withTimezone: true }), + watchersNotified: integer("watchers_notified").notNull().default(0), + payload: jsonb("payload").default({}), + lastError: text("last_error"), + createdAt: timestamp("created_at", { withTimezone: true }) + .notNull() + .defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (t) => ({ + marketWatcherJobsMarketEventIdx: index("market_watcher_jobs_market_event_idx").on( + t.marketId, + t.eventType, + ), + marketWatcherJobsReadyIdx: index("market_watcher_jobs_ready_idx").on( + t.status, + t.nextAttemptAt, + ), + marketWatcherJobsLeaseIdx: index("market_watcher_jobs_lease_idx").on( + t.status, + t.leaseUntil, + ), + }), +); + +export type MarketWatcherJob = typeof marketWatcherJobs.$inferSelect; +export type NewMarketWatcherJob = typeof marketWatcherJobs.$inferInsert; + + // --------------------------------------------------------------------------- // Referrals // --------------------------------------------------------------------------- diff --git a/src/metrics/registry.ts b/src/metrics/registry.ts index bf64025b..f2df6f3d 100644 --- a/src/metrics/registry.ts +++ b/src/metrics/registry.ts @@ -78,6 +78,33 @@ export const scheduledReportLeaseConflictsTotal = new Counter({ registers: [register], }); +export const marketWatcherJobRunsTotal = new Counter({ + name: "market_watcher_job_runs_total", + help: "Market watcher job runs by terminal outcome", + labelNames: ["status"] as const, + registers: [register], +}); + +export const marketWatcherJobRetriesTotal = new Counter({ + name: "market_watcher_job_retries_total", + help: "Market watcher job retry attempts by reason", + labelNames: ["reason"] as const, + registers: [register], +}); + +export const marketWatcherLeaseConflictsTotal = new Counter({ + name: "market_watcher_lease_conflicts_total", + help: "Market watcher jobs skipped because another worker owns the lease", + registers: [register], +}); + +export const marketWatcherNotificationsTotal = new Counter({ + name: "market_watcher_notifications_total", + help: "Total number of market watcher notifications generated and dispatched", + registers: [register], +}); + + export const webhookDeliveriesTotal = new Counter({ name: "webhook_deliveries_total", help: "Total number of webhook deliveries, segmented by outcome status (success, failed)", diff --git a/src/queue/index.ts b/src/queue/index.ts index a46d41b9..f24a75eb 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -16,6 +16,7 @@ export const backupVerificationQueueName = "backup-verification"; export const reconciliationQueueName = "reconciliation"; export const marketResolutionQueueName = "market-resolution"; export const scheduledReportQueueName = "scheduled-report-runs"; +export const marketWatcherQueueName = "market-watcher-jobs"; export const webhookQueue = new Queue(webhookQueueName, { // IORedis types conflict with BullMQ @@ -42,4 +43,10 @@ export const scheduledReportQueue = new Queue(scheduledReportQueueName, { connection: redisConnection, }); +export const marketWatcherQueue = new Queue(marketWatcherQueueName, { + // IORedis types conflict with BullMQ + connection: redisConnection, +}); + + export { Queue, Worker, QueueEvents }; diff --git a/src/services/marketWatcherJobService.ts b/src/services/marketWatcherJobService.ts new file mode 100644 index 00000000..87cd7133 --- /dev/null +++ b/src/services/marketWatcherJobService.ts @@ -0,0 +1,532 @@ +/** + * marketWatcherJobService.ts + * + * Durable, lease-backed market-watcher job processing service. + * Guarantees that duplicate jobs are not executed upon worker failovers, + * restarts, retries, or timeouts. + */ + +import { randomUUID } from "node:crypto"; +import { and, eq, isNull, lt, lte, or, sql } from "drizzle-orm"; +import type { Db } from "../db"; +import { db as defaultDb } from "../db"; +import { + marketWatcherJobs, + marketWatchers, + notifications, + type MarketWatcherJob, +} from "../db/schema"; +import { + marketWatcherJobRetriesTotal, + marketWatcherJobRunsTotal, + marketWatcherLeaseConflictsTotal, + marketWatcherNotificationsTotal, +} from "../metrics/registry"; +import { logger } from "../config/logger"; +import { marketWatcherQueue } from "../queue"; + +export const DEFAULT_MAX_ATTEMPTS = 3; +export const DEFAULT_LEASE_MS = 5 * 60 * 1000; +export const DEFAULT_RETRY_BASE_MS = 30 * 1000; +export const MAX_RETRY_DELAY_MS = 60 * 60 * 1000; + +export type MarketWatcherJobStatus = + | "pending" + | "running" + | "retryable" + | "succeeded" + | "failed"; + +export interface MarketWatcherJobData { + jobId: string; + marketId: string; + eventType: string; + eventRef: string; + jobKey: string; + payload?: Record; +} + +export interface MarketWatcherJobResult { + watchersNotified: number; +} + +export interface MarketWatcherJobHandlerInput extends MarketWatcherJobData { + attempt: number; +} + +export type MarketWatcherNotificationHandler = ( + input: MarketWatcherJobHandlerInput, +) => Promise; + +export interface MarketWatcherJobRepo { + createOrGetJob(input: { + marketId: string; + jobKey: string; + eventType: string; + payload?: Record; + }): Promise; + claimJob( + jobId: string, + leaseToken: string, + now: Date, + leaseMs: number, + ): Promise; + markSucceeded( + jobId: string, + leaseToken: string, + watchersNotified: number, + now: Date, + ): Promise; + markFailed(input: { + jobId: string; + leaseToken: string; + error: string; + now: Date; + maxAttempts: number; + nextAttemptAt: Date; + }): Promise<{ status: "retryable" | "failed"; attempt: number } | null>; + recoverExpiredLeases(now: Date): Promise; + getJob(jobId: string): Promise; +} + +export type QueueLike = { + add( + name: string, + data: MarketWatcherJobData, + options?: Record, + ): Promise; +}; + +/** + * Production repository backed by Drizzle ORM. + * Atomic lease token checking ensures failover workers can reclaim expired + * leases safely while preventing delayed stale workers from committing duplicate state. + */ +export class DrizzleMarketWatcherJobRepo implements MarketWatcherJobRepo { + constructor(private readonly database: Db = defaultDb) {} + + async createOrGetJob(input: { + marketId: string; + jobKey: string; + eventType: string; + payload?: Record; + }): Promise { + await this.database + .insert(marketWatcherJobs) + .values({ + marketId: input.marketId, + jobKey: input.jobKey, + eventType: input.eventType, + payload: input.payload ?? {}, + status: "pending", + nextAttemptAt: new Date(), + }) + .onConflictDoNothing({ + target: marketWatcherJobs.jobKey, + }); + + const [job] = await this.database + .select() + .from(marketWatcherJobs) + .where(eq(marketWatcherJobs.jobKey, input.jobKey)) + .limit(1); + + if (!job) { + throw new Error("market watcher job could not be created or loaded"); + } + return job; + } + + async claimJob( + jobId: string, + leaseToken: string, + now: Date, + leaseMs: number, + ): Promise { + const leaseUntil = new Date(now.getTime() + leaseMs); + const [job] = await this.database + .update(marketWatcherJobs) + .set({ + status: "running", + attempt: sql`${marketWatcherJobs.attempt} + 1`, + leaseToken, + leaseUntil, + startedAt: sql`COALESCE(${marketWatcherJobs.startedAt}, ${now})`, + updatedAt: now, + }) + .where( + and( + eq(marketWatcherJobs.id, jobId), + or( + eq(marketWatcherJobs.status, "pending"), + and( + eq(marketWatcherJobs.status, "retryable"), + or( + isNull(marketWatcherJobs.nextAttemptAt), + lte(marketWatcherJobs.nextAttemptAt, now), + ), + ), + and( + eq(marketWatcherJobs.status, "running"), + or( + isNull(marketWatcherJobs.leaseUntil), + lt(marketWatcherJobs.leaseUntil, now), + ), + ), + ), + ), + ) + .returning(); + + if (!job) { + marketWatcherLeaseConflictsTotal.inc(); + } + return job ?? null; + } + + async markSucceeded( + jobId: string, + leaseToken: string, + watchersNotified: number, + now: Date, + ): Promise { + const result = await this.database + .update(marketWatcherJobs) + .set({ + status: "succeeded", + watchersNotified, + completedAt: now, + leaseToken: null, + leaseUntil: null, + updatedAt: now, + }) + .where( + and( + eq(marketWatcherJobs.id, jobId), + eq(marketWatcherJobs.leaseToken, leaseToken), + eq(marketWatcherJobs.status, "running"), + ), + ) + .returning({ id: marketWatcherJobs.id }); + + const succeeded = result.length === 1; + if (succeeded) { + marketWatcherJobRunsTotal.inc({ status: "succeeded" }); + if (watchersNotified > 0) { + marketWatcherNotificationsTotal.inc(watchersNotified); + } + } + return succeeded; + } + + async markFailed(input: { + jobId: string; + leaseToken: string; + error: string; + now: Date; + maxAttempts: number; + nextAttemptAt: Date; + }): Promise<{ status: "retryable" | "failed"; attempt: number } | null> { + const [job] = await this.database + .update(marketWatcherJobs) + .set({ + status: sql`CASE WHEN ${marketWatcherJobs.attempt} >= ${input.maxAttempts} THEN 'failed' ELSE 'retryable' END`, + lastError: input.error.slice(0, 2000), + nextAttemptAt: input.nextAttemptAt, + completedAt: sql`CASE WHEN ${marketWatcherJobs.attempt} >= ${input.maxAttempts} THEN ${input.now} ELSE NULL END`, + leaseToken: null, + leaseUntil: null, + updatedAt: input.now, + }) + .where( + and( + eq(marketWatcherJobs.id, input.jobId), + eq(marketWatcherJobs.leaseToken, input.leaseToken), + eq(marketWatcherJobs.status, "running"), + ), + ) + .returning({ + status: marketWatcherJobs.status, + attempt: marketWatcherJobs.attempt, + }); + + if (!job) return null; + + const status = job.status as "retryable" | "failed"; + marketWatcherJobRetriesTotal.inc({ + reason: status === "retryable" ? "error" : "exhausted", + }); + if (status === "failed") { + marketWatcherJobRunsTotal.inc({ status: "failed" }); + } + return { status, attempt: job.attempt }; + } + + async recoverExpiredLeases(now: Date): Promise { + return this.database + .update(marketWatcherJobs) + .set({ + status: "retryable", + leaseToken: null, + leaseUntil: null, + nextAttemptAt: now, + updatedAt: now, + }) + .where( + and( + eq(marketWatcherJobs.status, "running"), + lt(marketWatcherJobs.leaseUntil, now), + ), + ) + .returning(); + } + + async getJob(jobId: string): Promise { + const [job] = await this.database + .select() + .from(marketWatcherJobs) + .where(eq(marketWatcherJobs.id, jobId)) + .limit(1); + return job ?? null; + } +} + +/** + * Builds a deterministic job key for market watcher operations. + */ +export function buildWatcherJobKey( + marketId: string, + eventType: string, + eventRef: string, +): string { + if (!marketId || !eventType || !eventRef) { + throw new Error("marketId, eventType, and eventRef are required to build a watcher job key"); + } + return `${marketId}:${eventType}:${eventRef}`; +} + +/** + * Exponential backoff calculation with upper bounding. + */ +export function retryDelayMs( + attempt: number, + baseMs = DEFAULT_RETRY_BASE_MS, +): number { + if (!Number.isInteger(attempt) || attempt < 1) { + throw new Error("attempt must be a positive integer"); + } + if (!Number.isFinite(baseMs) || baseMs < 0) { + throw new Error("retry base must be non-negative"); + } + return Math.min(MAX_RETRY_DELAY_MS, baseMs * 2 ** (attempt - 1)); +} + +/** + * Enqueues a market watcher notification job idempotently. + * If a job already exists for this event identity, it will not be duplicated. + */ +export async function enqueueMarketWatcherJob( + marketId: string, + eventType: string, + eventRef: string, + payload: Record = {}, + repository: MarketWatcherJobRepo = new DrizzleMarketWatcherJobRepo(), + queue: QueueLike = marketWatcherQueue, +): Promise<{ job: MarketWatcherJob; enqueued: boolean }> { + const jobKey = buildWatcherJobKey(marketId, eventType, eventRef); + const job = await repository.createOrGetJob({ + marketId, + jobKey, + eventType, + payload, + }); + + if (job.status === "succeeded" || job.status === "failed") { + return { job, enqueued: false }; + } + + await queue.add( + "notify-watchers", + { + jobId: job.id, + marketId, + eventType, + eventRef, + jobKey, + payload, + }, + { + jobId: jobKey, + removeOnComplete: false, + removeOnFail: false, + }, + ); + + return { job, enqueued: true }; +} + +export interface WatcherCoordinatorOptions { + leaseMs?: number; + maxAttempts?: number; + retryBaseMs?: number; + now?: () => Date; + queue?: QueueLike; +} + +/** + * Coordinator managing the end-to-end execution of market watcher jobs. + * Enforces single-active-worker leases, safe failover recovery, and idempotent retries. + */ +export class MarketWatcherJobCoordinator { + private readonly leaseMs: number; + private readonly maxAttempts: number; + private readonly retryBaseMs: number; + private readonly now: () => Date; + private readonly queue: QueueLike; + + constructor( + private readonly repository: MarketWatcherJobRepo, + options: WatcherCoordinatorOptions = {}, + ) { + this.leaseMs = options.leaseMs ?? DEFAULT_LEASE_MS; + this.maxAttempts = options.maxAttempts ?? DEFAULT_MAX_ATTEMPTS; + this.retryBaseMs = options.retryBaseMs ?? DEFAULT_RETRY_BASE_MS; + this.now = options.now ?? (() => new Date()); + this.queue = options.queue ?? marketWatcherQueue; + + if (!Number.isInteger(this.maxAttempts) || this.maxAttempts < 1) { + throw new Error("maxAttempts must be positive"); + } + } + + /** + * Process a market watcher job payload through the provided handler. + */ + async process( + job: MarketWatcherJobData, + handler: MarketWatcherNotificationHandler = defaultMarketWatcherHandler, + ): Promise<"skipped" | "succeeded" | "retryable" | "failed"> { + const leaseToken = randomUUID(); + const claimed = await this.repository.claimJob( + job.jobId, + leaseToken, + this.now(), + this.leaseMs, + ); + + if (!claimed) { + logger.info({ jobId: job.jobId, jobKey: job.jobKey }, "market_watcher_job_lease_skipped"); + return "skipped"; + } + + try { + const result = await handler({ + jobId: claimed.id, + marketId: claimed.marketId, + eventType: claimed.eventType, + eventRef: job.eventRef, + jobKey: claimed.jobKey, + payload: job.payload, + attempt: claimed.attempt, + }); + + if (!Number.isInteger(result.watchersNotified) || result.watchersNotified < 0) { + throw new Error("invalid watchersNotified count returned by handler"); + } + + const accepted = await this.repository.markSucceeded( + claimed.id, + leaseToken, + result.watchersNotified, + this.now(), + ); + + if (!accepted) { + logger.warn( + { jobId: job.jobId, leaseToken }, + "market_watcher_job_commit_rejected_lease_expired", + ); + return "skipped"; + } + + return "succeeded"; + } catch (error) { + const message = + error instanceof Error ? error.message : "unknown market watcher notification failure"; + const retryAt = new Date( + this.now().getTime() + retryDelayMs(claimed.attempt, this.retryBaseMs), + ); + + const failed = await this.repository.markFailed({ + jobId: claimed.id, + leaseToken, + error: message, + now: this.now(), + maxAttempts: this.maxAttempts, + nextAttemptAt: retryAt, + }); + + if (!failed) { + return "skipped"; + } + + if (failed.status === "retryable") { + await this.queue.add("notify-watchers", job, { + jobId: `${job.jobKey}:retry:${failed.attempt}`, + delay: Math.max(0, retryAt.getTime() - this.now().getTime()), + removeOnComplete: false, + removeOnFail: false, + }); + } + + logger.warn( + { + jobId: job.jobId, + attempt: failed.attempt, + status: failed.status, + err: message, + }, + "market_watcher_job_failed", + ); + + return failed.status; + } + } +} + +/** + * Default notification handler: queries subscribers for the market and + * creates notification records for each watcher in batches. + */ +export async function defaultMarketWatcherHandler( + input: MarketWatcherJobHandlerInput, + database: Db = defaultDb, +): Promise { + const watchers = await database + .select({ userId: marketWatchers.userId }) + .from(marketWatchers) + .where(eq(marketWatchers.marketId, input.marketId)); + + if (watchers.length === 0) { + return { watchersNotified: 0 }; + } + + const title = `Market Update: ${input.marketId}`; + const body = `Event [${input.eventType}] occurred on watched market ${input.marketId}`; + + const rows = watchers.map((w) => ({ + userId: w.userId, + type: input.eventType, + title, + body, + data: input.payload ?? {}, + })); + + const BATCH_SIZE = 500; + for (let i = 0; i < rows.length; i += BATCH_SIZE) { + const chunk = rows.slice(i, i + BATCH_SIZE); + await database.insert(notifications).values(chunk); + } + + return { watchersNotified: watchers.length }; +} diff --git a/src/services/marketWatcherService.ts b/src/services/marketWatcherService.ts index 7b930dc1..0f2efad6 100644 --- a/src/services/marketWatcherService.ts +++ b/src/services/marketWatcherService.ts @@ -196,3 +196,18 @@ export async function removeMarketWatcher( return deleted.length > 0; } + +/** + * Enqueues a notification job for all watchers of a given market. + * Idempotent: multiple calls with the same eventRef will not duplicate jobs or notifications. + */ +export async function notifyMarketWatchers( + marketId: string, + eventType: string, + eventRef: string, + payload: Record = {}, +) { + const { enqueueMarketWatcherJob } = await import("./marketWatcherJobService"); + return enqueueMarketWatcherJob(marketId, eventType, eventRef, payload); +} + diff --git a/src/workers/fraudDetector.ts b/src/workers/fraudDetector.ts index 26158957..4e97dea3 100644 --- a/src/workers/fraudDetector.ts +++ b/src/workers/fraudDetector.ts @@ -1,71 +1,68 @@ +/** + * fraudDetector.ts — background worker that periodically scans recent + * predictions for sybil / collusion clusters and persists `fraud_flags`. + * + * Designed to be invoked from: + * • a cron-style scheduler (every N minutes) + * • the existing in-process scheduler (`src/services/scheduler.ts`) + * • or one-off CLI runs (`node dist/workers/fraudDetector.js`) + * + * The worker itself is intentionally tiny — all logic lives in + * `fraudService.ts` so it can be unit-tested without spinning up a job + * runtime. A correlation id is generated per run so every log line and + * persisted flag can be traced. + */ + import { randomUUID } from "crypto"; import { logger } from "../config/logger"; -import { env } from "../config/env"; import { - DrzzleFraudRepo, + DrizzleFraudRepo, type FraudRepo, type RunScanOptions, type RunScanResult, runFraudScan, } from "../services/fraudService"; -export interface FraudDetectorConfig { - maxAlerts?: number; - maxAnomalies?: number; -} - -export interface FraudRunScanOptions extends RunScanOptions { - maxAlerts?: number; - maxAnomalies?: number; -} - -function positiveInt(value: number | undefined, fallback: number): number { - if (value === undefined) return fallback; - const n = Math.floor(Number(value)); - return Number.isFinite(n) && n > 0 ? n : fallback; -} - export class FraudDetectorWorker { private readonly repo: FraudRepo; - private readonly config: { maxAlerts: number; maxAnomalies: number }; private timer: NodeJS.Timeout | null = null; - constructor( - repo: FraudRepo = new DrzzleFraudRepo(), - config: FraudDetectorConfig = {}, - ) { + constructor(repo: FraudRepo = new DrizzleFraudRepo()) { this.repo = repo; - this.config = { - maxAlerts: positiveInt(config.maxAlerts, env.FRAUD_SCAN_MAX_ALERTS), - maxAnomalies: positiveInt(config.maxAnomalies, env.FRAUD_SCAN_MAX_ANOMALIES), - }; } - async runOnce(opts: FraudRunScanOptions = {}): Promise { + /** Run a single scan. Errors are caught and logged — the worker never throws. */ + async runOnce(opts: RunScanOptions = {}): Promise { const correlationId = opts.correlationId ?? randomUUID(); - const merged: FraudRunScanOptions = { - ...opts, - correlationId, - maxAlerts: opts.maxAlerts ?? this.config.maxAlerts, - maxAnomalies: opts.maxAnomalies ?? this.config.maxAnomalies, - }; + const merged: RunScanOptions = { ...opts }; + merged.correlationId = correlationId; try { const result = await runFraudScan(this.repo, merged); logger.info({ ...result }, "fraud_detector: run complete"); return result; } catch (err) { - logger.error({ correlationId, err }, "fraud_detector: run failed"); + logger.error( + { correlationId, err }, + "fraud_detector: run failed", + ); return null; } } - start(intervalMs = 15 * 60 * 1000, opts: FraudRunScanOptions = {}): () => void { + /** + * Start a recurring scan. Returns a stop handle. + * `intervalMs` defaults to 15 minutes; non-positive disables scheduling. + */ + start(intervalMs = 15 * 60 * 1000, opts: RunScanOptions = {}): () => void { if (this.timer) { logger.warn("fraud_detector: already running, ignoring start()"); return () => this.stop(); } if (!Number.isFinite(intervalMs) || intervalMs <= 0) { - logger.warn({ intervalMs }, "fraud_detector: invalid interval, not starting"); + logger.warn( + { intervalMs }, + "fraud_detector: invalid interval, not starting", + ); return () => undefined; } @@ -88,14 +85,21 @@ export class FraudDetectorWorker { } } -export const fraudDetectorWorker = new FraudDetectorGorker(); +/** Singleton for production wiring. */ +export const fraudDetectorWorker = new FraudDetectorWorker(); +// Allow `node dist/workers/fraudDetector.js` for ad-hoc runs. if (require.main === module) { - fraudDetectorWorker.runOnce().then((res) => { - console.log("fraud_scan", res); - process.exit(0); - }).catch((err) => { - console.error(err); - process.exit(1); - }); -} \ No newline at end of file + fraudDetectorWorker + .runOnce() + .then((res) => { + + console.log("fraud_scan", res); + process.exit(0); + }) + .catch((err) => { + + console.error(err); + process.exit(1); + }); +} diff --git a/src/workers/marketWatcherWorker.ts b/src/workers/marketWatcherWorker.ts new file mode 100644 index 00000000..d8aee16a --- /dev/null +++ b/src/workers/marketWatcherWorker.ts @@ -0,0 +1,78 @@ +import { Job, Worker } from "bullmq"; +import { logger } from "../config/logger"; +import { + DrizzleMarketWatcherJobRepo, + type MarketWatcherNotificationHandler, + type MarketWatcherJobData, + MarketWatcherJobCoordinator, + type MarketWatcherJobRepo, + defaultMarketWatcherHandler, +} from "../services/marketWatcherJobService"; +import { redisConnection, marketWatcherQueueName } from "../queue"; + +export interface MarketWatcherWorkerOptions { + concurrency?: number; + leaseMs?: number; + maxAttempts?: number; + retryBaseMs?: number; + queue?: { + add( + name: string, + data: MarketWatcherJobData, + options?: Record, + ): Promise; + }; +} + +/** + * BullMQ worker adapter for processing market watcher notification jobs. + * Translates queue events to coordinator process calls. + */ +export class MarketWatcherWorker { + private worker: Worker | null = null; + private readonly coordinator: MarketWatcherJobCoordinator; + private readonly handler: MarketWatcherNotificationHandler; + private readonly concurrency: number; + + constructor( + repository: MarketWatcherJobRepo = new DrizzleMarketWatcherJobRepo(), + handler: MarketWatcherNotificationHandler = defaultMarketWatcherHandler, + options: MarketWatcherWorkerOptions = {}, + ) { + this.coordinator = new MarketWatcherJobCoordinator(repository, options); + this.handler = handler; + this.concurrency = options.concurrency ?? 4; + } + + start(): void { + if (this.worker) return; + this.worker = new Worker( + marketWatcherQueueName, + async (job: Job) => + this.coordinator.process(job.data, this.handler), + { connection: redisConnection, concurrency: this.concurrency }, + ); + this.worker.on("failed", (job, error) => { + logger.error( + { jobId: job?.id, marketId: job?.data.marketId, err: error.message }, + "market watcher queue job failed", + ); + }); + logger.info({ concurrency: this.concurrency }, "market watcher worker started"); + } + + async stop(): Promise { + if (!this.worker) return; + await this.worker.close(); + this.worker = null; + logger.info("market watcher worker stopped"); + } +} + +export function createMarketWatcherWorker( + repository?: MarketWatcherJobRepo, + handler?: MarketWatcherNotificationHandler, + options?: MarketWatcherWorkerOptions, +): MarketWatcherWorker { + return new MarketWatcherWorker(repository, handler, options); +} diff --git a/tests/marketWatcherJobService.test.ts b/tests/marketWatcherJobService.test.ts new file mode 100644 index 00000000..1793ac31 --- /dev/null +++ b/tests/marketWatcherJobService.test.ts @@ -0,0 +1,506 @@ +jest.mock("../src/queue", () => ({ + marketWatcherQueue: { add: jest.fn().mockResolvedValue(undefined) }, + marketWatcherQueueName: "market-watcher-jobs", + redisConnection: { on: jest.fn() }, +})); + +import { randomUUID } from "node:crypto"; +import { + buildWatcherJobKey, + retryDelayMs, + DEFAULT_MAX_ATTEMPTS, + DEFAULT_LEASE_MS, + DEFAULT_RETRY_BASE_MS, + MAX_RETRY_DELAY_MS, + type MarketWatcherJobData, + type MarketWatcherJobRepo, + MarketWatcherJobCoordinator, + enqueueMarketWatcherJob, + defaultMarketWatcherHandler, +} from "../src/services/marketWatcherJobService"; +import type { MarketWatcherJob } from "../src/db/schema"; + +class InMemoryMarketWatcherJobRepo implements MarketWatcherJobRepo { + public jobs = new Map(); + public leaseConflicts = 0; + + async createOrGetJob(input: { + marketId: string; + jobKey: string; + eventType: string; + payload?: Record; + }): Promise { + const existing = Array.from(this.jobs.values()).find((j) => j.jobKey === input.jobKey); + if (existing) return existing; + + const now = new Date(); + const job: MarketWatcherJob = { + id: randomUUID(), + marketId: input.marketId, + jobKey: input.jobKey, + eventType: input.eventType, + status: "pending", + attempt: 0, + leaseToken: null, + leaseUntil: null, + startedAt: null, + completedAt: null, + nextAttemptAt: now, + watchersNotified: 0, + payload: input.payload ?? {}, + lastError: null, + createdAt: now, + updatedAt: now, + }; + this.jobs.set(job.id, job); + return job; + } + + async claimJob( + jobId: string, + leaseToken: string, + now: Date, + leaseMs: number, + ): Promise { + const job = this.jobs.get(jobId); + if (!job) return null; + + const isPending = job.status === "pending"; + const isRetryableReady = + job.status === "retryable" && (!job.nextAttemptAt || job.nextAttemptAt <= now); + const isRunningExpired = + job.status === "running" && (!job.leaseUntil || job.leaseUntil < now); + + if (!isPending && !isRetryableReady && !isRunningExpired) { + this.leaseConflicts++; + return null; + } + + job.status = "running"; + job.attempt += 1; + job.leaseToken = leaseToken; + job.leaseUntil = new Date(now.getTime() + leaseMs); + job.startedAt = job.startedAt ?? now; + job.updatedAt = now; + return { ...job }; + } + + async markSucceeded( + jobId: string, + leaseToken: string, + watchersNotified: number, + now: Date, + ): Promise { + const job = this.jobs.get(jobId); + if (!job || job.status !== "running" || job.leaseToken !== leaseToken) { + return false; + } + + job.status = "succeeded"; + job.watchersNotified = watchersNotified; + job.completedAt = now; + job.leaseToken = null; + job.leaseUntil = null; + job.updatedAt = now; + return true; + } + + async markFailed(input: { + jobId: string; + leaseToken: string; + error: string; + now: Date; + maxAttempts: number; + nextAttemptAt: Date; + }): Promise<{ status: "retryable" | "failed"; attempt: number } | null> { + const job = this.jobs.get(input.jobId); + if (!job || job.status !== "running" || job.leaseToken !== input.leaseToken) { + return null; + } + + const isTerminal = job.attempt >= input.maxAttempts; + job.status = isTerminal ? "failed" : "retryable"; + job.lastError = input.error; + job.nextAttemptAt = input.nextAttemptAt; + job.completedAt = isTerminal ? input.now : null; + job.leaseToken = null; + job.leaseUntil = null; + job.updatedAt = input.now; + return { status: job.status, attempt: job.attempt }; + } + + async recoverExpiredLeases(now: Date): Promise { + const recovered: MarketWatcherJob[] = []; + for (const job of this.jobs.values()) { + if (job.status === "running" && job.leaseUntil && job.leaseUntil < now) { + job.status = "retryable"; + job.leaseToken = null; + job.leaseUntil = null; + job.nextAttemptAt = now; + job.updatedAt = now; + recovered.push({ ...job }); + } + } + return recovered; + } + + async getJob(jobId: string): Promise { + const job = this.jobs.get(jobId); + return job ? { ...job } : null; + } +} + +describe("marketWatcherJobService", () => { + describe("buildWatcherJobKey", () => { + it("builds a deterministic key from marketId, eventType, and eventRef", () => { + const key = buildWatcherJobKey("mkt-123", "market.resolved", "tx-456"); + expect(key).toBe("mkt-123:market.resolved:tx-456"); + }); + + it("throws error for missing arguments", () => { + expect(() => buildWatcherJobKey("", "market.resolved", "ref")).toThrow(); + expect(() => buildWatcherJobKey("mkt-1", "", "ref")).toThrow(); + expect(() => buildWatcherJobKey("mkt-1", "market.resolved", "")).toThrow(); + }); + }); + + describe("retryDelayMs", () => { + it("calculates exponential backoff with default base", () => { + expect(retryDelayMs(1)).toBe(30_000); + expect(retryDelayMs(2)).toBe(60_000); + expect(retryDelayMs(3)).toBe(120_000); + }); + + it("caps delay at MAX_RETRY_DELAY_MS", () => { + expect(retryDelayMs(20, DEFAULT_RETRY_BASE_MS)).toBe(MAX_RETRY_DELAY_MS); + }); + + it("throws error for invalid attempt or negative base", () => { + expect(() => retryDelayMs(0)).toThrow(); + expect(() => retryDelayMs(-1)).toThrow(); + expect(() => retryDelayMs(1, -100)).toThrow(); + }); + }); + + describe("enqueueMarketWatcherJob", () => { + it("creates and enqueues job idempotently", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const mockQueue = { add: jest.fn().mockResolvedValue({}) }; + + const result1 = await enqueueMarketWatcherJob( + "mkt-1", + "market.resolved", + "evt-1", + { winner: "YES" }, + repo, + mockQueue, + ); + + expect(result1.enqueued).toBe(true); + expect(result1.job.marketId).toBe("mkt-1"); + expect(mockQueue.add).toHaveBeenCalledTimes(1); + + // Re-enqueuing duplicate + const result2 = await enqueueMarketWatcherJob( + "mkt-1", + "market.resolved", + "evt-1", + { winner: "YES" }, + repo, + mockQueue, + ); + + expect(result2.job.id).toBe(result1.job.id); + }); + + it("does not enqueue if job is already in terminal succeeded state", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const mockQueue = { add: jest.fn().mockResolvedValue({}) }; + + const { job } = await enqueueMarketWatcherJob( + "mkt-1", + "market.resolved", + "evt-1", + {}, + repo, + mockQueue, + ); + + const claimed = await repo.claimJob(job.id, "token-1", new Date(), 5000); + await repo.markSucceeded(claimed!.id, "token-1", 5, new Date()); + + const result = await enqueueMarketWatcherJob( + "mkt-1", + "market.resolved", + "evt-1", + {}, + repo, + mockQueue, + ); + + expect(result.enqueued).toBe(false); + expect(result.job.status).toBe("succeeded"); + }); + }); + + describe("MarketWatcherJobCoordinator", () => { + it("claims, executes notification handler, and marks succeeded", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const coordinator = new MarketWatcherJobCoordinator(repo); + + const job = await repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + const handler = jest.fn().mockResolvedValue({ watchersNotified: 10 }); + + const outcome = await coordinator.process( + { + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef: "evt-1", + jobKey: job.jobKey, + }, + handler, + ); + + expect(outcome).toBe("succeeded"); + expect(handler).toHaveBeenCalledWith( + expect.objectContaining({ + jobId: job.id, + marketId: "mkt-1", + attempt: 1, + }), + ); + + const updated = await repo.getJob(job.id); + expect(updated?.status).toBe("succeeded"); + expect(updated?.watchersNotified).toBe(10); + expect(updated?.completedAt).toBeDefined(); + }); + + it("skips execution when lease is already owned by another active worker", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const coordinator = new MarketWatcherJobCoordinator(repo); + + const job = await repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + // Worker 1 claims lease + await repo.claimJob(job.id, "worker-1-token", new Date(), DEFAULT_LEASE_MS); + + // Worker 2 attempts processing + const handler = jest.fn(); + const outcome = await coordinator.process( + { + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef: "evt-1", + jobKey: job.jobKey, + }, + handler, + ); + + expect(outcome).toBe("skipped"); + expect(handler).not.toHaveBeenCalled(); + }); + + it("reclaims expired lease upon worker failover and completes successfully", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const now = new Date("2026-08-29T10:00:00Z"); + const coordinator = new MarketWatcherJobCoordinator(repo, { now: () => now }); + + const job = await repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + // Old Worker 1 claimed lease at 09:50 with 5 min lease (expired at 09:55) + const oldTime = new Date("2026-08-29T09:50:00Z"); + await repo.claimJob(job.id, "worker-1-token", oldTime, 5 * 60 * 1000); + + // Worker 2 (failover worker) processes at 10:00:00Z + const handler = jest.fn().mockResolvedValue({ watchersNotified: 3 }); + const outcome = await coordinator.process( + { + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef: "evt-1", + jobKey: job.jobKey, + }, + handler, + ); + + expect(outcome).toBe("succeeded"); + expect(handler).toHaveBeenCalledWith( + expect.objectContaining({ + jobId: job.id, + attempt: 2, // Incremented upon reclaim + }), + ); + + const finalJob = await repo.getJob(job.id); + expect(finalJob?.status).toBe("succeeded"); + expect(finalJob?.watchersNotified).toBe(3); + }); + + it("rejects commit from a stale worker whose lease expired and was claimed by a failover worker", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + + const job = await repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + // Worker 1 claims job + const worker1Claim = await repo.claimJob(job.id, "token-1", new Date(), 1000); + expect(worker1Claim).not.toBeNull(); + + // Time passes, lease expires, Worker 2 reclaims job + const future = new Date(Date.now() + 5000); + const worker2Claim = await repo.claimJob(job.id, "token-2", future, DEFAULT_LEASE_MS); + expect(worker2Claim).not.toBeNull(); + + // Worker 1 now attempts to mark succeeded with old token-1 -> rejected! + const worker1Success = await repo.markSucceeded(job.id, "token-1", 10, future); + expect(worker1Success).toBe(false); + + // Worker 2 completes with token-2 -> accepted! + const worker2Success = await repo.markSucceeded(job.id, "token-2", 10, future); + expect(worker2Success).toBe(true); + + const finalJob = await repo.getJob(job.id); + expect(finalJob?.status).toBe("succeeded"); + }); + + it("handles failure with retry and exponential backoff delay", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const mockQueue = { add: jest.fn().mockResolvedValue({}) }; + const now = new Date("2026-08-29T10:00:00.000Z"); + const coordinator = new MarketWatcherJobCoordinator(repo, { + now: () => now, + queue: mockQueue, + maxAttempts: 3, + retryBaseMs: 10_000, + }); + + const job = await repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + const handler = jest.fn().mockRejectedValue(new Error("Network connection dropped")); + + const outcome = await coordinator.process( + { + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef: "evt-1", + jobKey: job.jobKey, + }, + handler, + ); + + expect(outcome).toBe("retryable"); + const failedJob = await repo.getJob(job.id); + expect(failedJob?.status).toBe("retryable"); + expect(failedJob?.attempt).toBe(1); + expect(failedJob?.lastError).toBe("Network connection dropped"); + expect(failedJob?.nextAttemptAt?.toISOString()).toBe("2026-08-29T10:00:10.000Z"); + + expect(mockQueue.add).toHaveBeenCalledWith( + "notify-watchers", + expect.objectContaining({ jobId: job.id }), + expect.objectContaining({ + jobId: `${job.jobKey}:retry:1`, + delay: 10_000, + }), + ); + }); + + it("marks terminal failed status upon exhausting max attempts", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const mockQueue = { add: jest.fn().mockResolvedValue({}) }; + let currentTime = new Date("2026-08-29T10:00:00.000Z"); + const coordinator = new MarketWatcherJobCoordinator(repo, { + now: () => currentTime, + queue: mockQueue, + maxAttempts: 2, + retryBaseMs: 10_000, + }); + + const job = await repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + const handler = jest.fn().mockRejectedValue(new Error("Persistent database failure")); + + // Attempt 1: retryable + const outcome1 = await coordinator.process( + { + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef: "evt-1", + jobKey: job.jobKey, + }, + handler, + ); + expect(outcome1).toBe("retryable"); + + // Advance time beyond nextAttemptAt (10s) + currentTime = new Date("2026-08-29T10:00:15.000Z"); + + // Attempt 2: exhausted -> failed + const outcome2 = await coordinator.process( + { + jobId: job.id, + marketId: job.marketId, + eventType: job.eventType, + eventRef: "evt-1", + jobKey: job.jobKey, + }, + handler, + ); + + expect(outcome2).toBe("failed"); + const failedJob = await repo.getJob(job.id); + expect(failedJob?.status).toBe("failed"); + expect(failedJob?.attempt).toBe(2); + expect(failedJob?.completedAt).toBeDefined(); + }); + + it("recovers expired leases via recoverExpiredLeases", async () => { + const repo = new InMemoryMarketWatcherJobRepo(); + const now = new Date("2026-08-29T12:00:00Z"); + + const job = await repo.createOrGetJob({ + marketId: "mkt-1", + jobKey: "mkt-1:market.resolved:evt-1", + eventType: "market.resolved", + }); + + // Claimed with expired lease + await repo.claimJob(job.id, "token-x", new Date("2026-08-29T11:00:00Z"), 60_000); + + const recovered = await repo.recoverExpiredLeases(now); + expect(recovered.length).toBe(1); + expect(recovered[0].status).toBe("retryable"); + expect(recovered[0].leaseToken).toBeNull(); + }); + }); +}); diff --git a/tests/marketWatcherWorker.test.ts b/tests/marketWatcherWorker.test.ts new file mode 100644 index 00000000..d7af39c9 --- /dev/null +++ b/tests/marketWatcherWorker.test.ts @@ -0,0 +1,38 @@ +jest.mock("bullmq", () => ({ + Worker: jest.fn().mockImplementation((_name: string, processor: unknown) => ({ + processor, + on: jest.fn(), + close: jest.fn().mockResolvedValue(undefined), + })), +})); + +jest.mock("../src/queue", () => ({ + redisConnection: {}, + marketWatcherQueueName: "market-watcher-jobs", +})); + +import { Worker } from "bullmq"; +import { MarketWatcherWorker } from "../src/workers/marketWatcherWorker"; +import { MarketWatcherJobRepo } from "../src/services/marketWatcherJobService"; + +describe("MarketWatcherWorker", () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it("starts one BullMQ worker and is idempotent", () => { + const worker = new MarketWatcherWorker({} as MarketWatcherJobRepo); + worker.start(); + worker.start(); + expect(Worker).toHaveBeenCalledTimes(1); + }); + + it("closes the worker during stop and tolerates repeated stop", async () => { + const worker = new MarketWatcherWorker({} as MarketWatcherJobRepo); + worker.start(); + await worker.stop(); + await worker.stop(); + const instance = (Worker as unknown as jest.Mock).mock.results.at(-1)?.value; + expect(instance.close).toHaveBeenCalledTimes(1); + }); +}); diff --git a/tests/marketWatchers.test.ts b/tests/marketWatchers.test.ts index 5bc35ad3..7209d4f8 100644 --- a/tests/marketWatchers.test.ts +++ b/tests/marketWatchers.test.ts @@ -9,8 +9,7 @@ process.env.JWT_SECRET = "test-secret-with-at-least-32-characters"; import request from "supertest"; import express, { type Request, type Response, type NextFunction } from "express"; -jest.mock("../src/services/marketWatcherService"); -jest.mock("../src/middleware/auth", () => ({ +const authMock = { requireAuth: (req: Request, res: Response, next: NextFunction) => { const authHeader = req.headers.authorization; if (!authHeader || !authHeader.startsWith("Bearer ")) { @@ -19,7 +18,12 @@ jest.mock("../src/middleware/auth", () => ({ (req as any).user = { id: "user-123", stellarAddress: "GABC123" }; next(); }, -})); +}; +jest.mock("../src/middleware/auth", () => authMock); +jest.mock("../src/middleware/requireAuth", () => authMock); +jest.mock("../src/services/marketWatcherService"); + + import * as marketWatcherService from "../src/services/marketWatcherService"; import { watchersRouter } from "../src/routes/markets/watchers";