diff --git a/drizzle/migrations/0027_scheduled_report_runs.sql b/drizzle/migrations/0027_scheduled_report_runs.sql new file mode 100644 index 00000000..04ad5826 --- /dev/null +++ b/drizzle/migrations/0027_scheduled_report_runs.sql @@ -0,0 +1,26 @@ +CREATE TABLE "scheduled_report_runs" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "scheduled_report_id" uuid NOT NULL, + "schedule_key" text NOT NULL, + "run_for" timestamp with time zone 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, + "output_ref" text, + "last_error" text, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + "updated_at" timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT "scheduled_report_runs_identity_unique" UNIQUE("scheduled_report_id", "run_for") +); +--> statement-breakpoint +ALTER TABLE "scheduled_report_runs" ADD CONSTRAINT "scheduled_report_runs_scheduled_report_id_fk" FOREIGN KEY ("scheduled_report_id") REFERENCES "public"."scheduled_reports"("id") ON DELETE cascade ON UPDATE no action; +--> statement-breakpoint +CREATE INDEX "scheduled_report_runs_identity_idx" ON "scheduled_report_runs" USING btree ("scheduled_report_id", "run_for"); +--> statement-breakpoint +CREATE INDEX "scheduled_report_runs_ready_idx" ON "scheduled_report_runs" USING btree ("status", "next_attempt_at"); +--> statement-breakpoint +CREATE INDEX "scheduled_report_runs_lease_idx" ON "scheduled_report_runs" USING btree ("status", "lease_until"); diff --git a/package-lock.json b/package-lock.json index 58901627..731e1904 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17,6 +17,7 @@ "drizzle-orm": "^0.45.2", "express": "^4.21.0", "express-rate-limit": "^7.4.0", + "express-session": "^1.19.0", "helmet": "^7.2.0", "ioredis": "^5.11.1", "jsonwebtoken": "^9.0.2", @@ -102,6 +103,7 @@ "integrity": "sha512-RgHBCvtjbOK2gXSNBNIkNoEc9qoVEtau3hj8gEqKQuL3HZAibKarWFEI3Lfm6EYKkLalOh8eSrj9b+ch9H/VBA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@babel/code-frame": "^7.29.7", "@babel/generator": "^7.29.7", @@ -2486,6 +2488,7 @@ "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.1.tgz", "integrity": "sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==", "license": "Apache-2.0", + "peer": true, "engines": { "node": ">=8.0.0" } @@ -2888,6 +2891,7 @@ "integrity": "sha512-sKYVuV7Sv9fbPIt/442koC7+IIwK5olP1KWeD88e/idgoJqDm3JV/YUiPwkoKK92ylff2MGxSz1CSjsXelx0YA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@types/body-parser": "*", "@types/express-serve-static-core": "^5.0.0", @@ -3007,6 +3011,7 @@ "integrity": "sha512-EANqOCF9QFyra+4pfxUcX9STKJpCLjMbObVzljIJomAWSnuSIEAvyzEU53GaajbXJEgdh0iEcPL+DGvpUd4k1Q==", "devOptional": true, "license": "MIT", + "peer": true, "dependencies": { "undici-types": "~6.21.0" } @@ -3017,6 +3022,7 @@ "integrity": "sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==", "devOptional": true, "license": "MIT", + "peer": true, "dependencies": { "@types/node": "*", "pg-protocol": "*", @@ -3220,6 +3226,7 @@ "integrity": "sha512-CZ4nMxWwgu1HEEFNkeaCptra9QCtkmKdgf3sWh1rl1trIhmxLilgTV4cwcbQ4wemnT4sWQN8CaKOmdYx+g2gMA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@typescript-eslint/scope-manager": "8.65.0", "@typescript-eslint/types": "8.65.0", @@ -3489,6 +3496,7 @@ "integrity": "sha512-xRQbDb9BnwDafYNn6Vwl839DYVjqXYb1XVGtWAZ1kcDc6iwAL4hg3B1dZlRiuENFeO2H53gFG3in621AdERVAg==", "dev": true, "license": "MIT", + "peer": true, "bin": { "acorn": "bin/acorn" }, @@ -4056,6 +4064,7 @@ "integrity": "sha512-iQxPClE07hETVpbRoX7JXX3v/ZQViCxe/SYCxylRLzdEx1xJAufPptfiOqR8tqiCtmbtMDANKWszzjLu1PMAZQ==", "devOptional": true, "license": "Apache-2.0", + "peer": true, "dependencies": { "bare-path": "^3.0.0" } @@ -4282,6 +4291,7 @@ } ], "license": "MIT", + "peer": true, "dependencies": { "baseline-browser-mapping": "^2.10.44", "caniuse-lite": "^1.0.30001806", @@ -5513,6 +5523,7 @@ "integrity": "sha512-DgZS62aPLXKlnxILS/AYCoRvHaZeXceIzlXPkkGGzJWSow1aEk0lbTlxUSlyjC8jcaKxAdOnTDz+o1JFSBsyjw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@eslint-community/eslint-utils": "^4.8.0", "@eslint-community/regexpp": "^4.12.1", @@ -5778,6 +5789,7 @@ "resolved": "https://registry.npmjs.org/express/-/express-4.22.2.tgz", "integrity": "sha512-IuL+Elrou2ZvCFHs18/CIzy2Nzvo25nZ1/D2eIZlz7c+QUayAcYoiM2BthCjs+EBHVpjYjcuLDAiCWgeIX3X1Q==", "license": "MIT", + "peer": true, "dependencies": { "accepts": "~1.3.8", "array-flatten": "1.1.1", @@ -5834,6 +5846,44 @@ "express": ">= 4.11" } }, + "node_modules/express-session": { + "version": "1.19.0", + "resolved": "https://registry.npmjs.org/express-session/-/express-session-1.19.0.tgz", + "integrity": "sha512-0csaMkGq+vaiZTmSMMGkfdCOabYv192VbytFypcvI0MANrp+4i/7yEkJ0sbAEhycQjntaKGzYfjfXQyVb7BHMA==", + "license": "MIT", + "dependencies": { + "cookie": "~0.7.2", + "cookie-signature": "~1.0.7", + "debug": "~2.6.9", + "depd": "~2.0.0", + "on-headers": "~1.1.0", + "parseurl": "~1.3.3", + "safe-buffer": "~5.2.1", + "uid-safe": "~2.1.5" + }, + "engines": { + "node": ">= 0.8.0" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/express" + } + }, + "node_modules/express-session/node_modules/debug": { + "version": "2.6.9", + "resolved": "https://registry.npmjs.org/debug/-/debug-2.6.9.tgz", + "integrity": "sha512-bC7ElrdJaJnPbAP+1EotYvqZsb3ecl5wi6Bfi6BJTUcNowp6cvspg0jXznRTKDjm/E7AdgFBVeAPVMNcKGsHMA==", + "license": "MIT", + "dependencies": { + "ms": "2.0.0" + } + }, + "node_modules/express-session/node_modules/ms": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.0.0.tgz", + "integrity": "sha512-Tpp60P6IUJDTuOq/5Z8cdskzJujfwqfOTkrwIwj7IRISpnkJnT6SyJ4PCPnGMoFjC9ddhal5KVIYtAt97ix05A==", + "license": "MIT" + }, "node_modules/express/node_modules/debug": { "version": "2.6.9", "resolved": "https://registry.npmjs.org/debug/-/debug-2.6.9.tgz", @@ -6833,6 +6883,7 @@ "integrity": "sha512-NIy3oAFp9shda19hy4HK0HRTWKtPJmGdnvywu01nOqNC2vZg+Z+fvJDxpMQA88eb2I9EcafcdjYgsDthnYTvGw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@jest/core": "^29.7.0", "@jest/types": "^29.6.3", @@ -7427,6 +7478,7 @@ "integrity": "sha512-AC/7JofJvZGrrneWNaEnJeOLUx+JlGt7tNa0wZiRPT4MY1wmfKjt2+6O2p2uz2+skll8OZZmJMNqeke7kKbNgQ==", "dev": true, "license": "MIT", + "peer": true, "bin": { "jiti": "lib/jiti-cli.mjs" } @@ -8149,6 +8201,15 @@ "node": ">= 0.8" } }, + "node_modules/on-headers": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/on-headers/-/on-headers-1.1.0.tgz", + "integrity": "sha512-737ZY3yNnXy37FHkQxPzt4UZ2UWPWiCZWLvFZ4fu5cueciegX0zGPnrlY6bwRg4FdQOe9YU8MkmJwGhoMybl8A==", + "license": "MIT", + "engines": { + "node": ">= 0.8" + } + }, "node_modules/once": { "version": "1.4.0", "resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz", @@ -8364,6 +8425,7 @@ "resolved": "https://registry.npmjs.org/pg/-/pg-8.22.0.tgz", "integrity": "sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==", "license": "MIT", + "peer": true, "dependencies": { "pg-connection-string": "^2.14.0", "pg-pool": "^3.14.0", @@ -8878,6 +8940,15 @@ "integrity": "sha512-tYC1Q1hgyRuHgloV/YXs2w15unPVh8qfu/qCTfhTYamaw7fyhumKa2yGpdSo87vY32rIclj+4fWYQXUMs9EHvg==", "license": "MIT" }, + "node_modules/random-bytes": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/random-bytes/-/random-bytes-1.0.0.tgz", + "integrity": "sha512-iv7LhNVO047HzYR3InF6pUcUsPQiHTM1Qal51DcGSuZFBil1aBBWG5eHPNek7bvILMaYJ/8RU1e8w1AMdHmLQQ==", + "license": "MIT", + "engines": { + "node": ">= 0.8" + } + }, "node_modules/randombytes": { "version": "2.1.0", "resolved": "https://registry.npmjs.org/randombytes/-/randombytes-2.1.0.tgz", @@ -9928,6 +9999,7 @@ "integrity": "sha512-RvwwcruNjI1ncT5xRakeyS9Lf8lcItv34KD+aif+VH9kduAyfYBipGh12274xtenIPZ119/R9BdTBa8gAwSh0A==", "dev": true, "license": "MIT", + "peer": true, "engines": { "node": ">=12" }, @@ -10089,6 +10161,7 @@ "integrity": "sha512-f0FFpIdcHgn8zcPSbf1dRevwt047YMnaiJM3u2w2RewrB+fob/zePZcrOyQoLMMO7aBIddLcQIEK5dYjkLnGrQ==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "@cspotcode/source-map-support": "^0.8.0", "@tsconfig/node10": "^1.0.7", @@ -10792,6 +10865,7 @@ "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", "dev": true, "license": "Apache-2.0", + "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -10838,6 +10912,18 @@ "node": ">=0.8.0" } }, + "node_modules/uid-safe": { + "version": "2.1.5", + "resolved": "https://registry.npmjs.org/uid-safe/-/uid-safe-2.1.5.tgz", + "integrity": "sha512-KPHm4VL5dDXKz01UuEd88Df+KzynaohSL9fBh096KWAxSKZQDI2uBrVqtvRM4rwrIrRRKsdLNML/lnaaVSRioA==", + "license": "MIT", + "dependencies": { + "random-bytes": "~1.0.0" + }, + "engines": { + "node": ">= 0.8" + } + }, "node_modules/undici": { "version": "8.9.0", "resolved": "https://registry.npmjs.org/undici/-/undici-8.9.0.tgz", @@ -11207,6 +11293,7 @@ "resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz", "integrity": "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==", "license": "MIT", + "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" } diff --git a/src/db/schema.ts b/src/db/schema.ts index 786c8044..8a7c4903 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -180,7 +180,6 @@ export const predictions = pgTable("predictions", { claimTxHash: text("claim_tx_hash"), /** Timestamp when the claim transaction was submitted. Null until claimed. */ claimedAt: timestamp("claimed_at", { withTimezone: true }), - createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), createdAt: timestamp("created_at", { withTimezone: true }) .notNull() .defaultNow(), @@ -612,6 +611,58 @@ export const scheduledReports = pgTable( export type ScheduledReport = typeof scheduledReports.$inferSelect; export type NewScheduledReport = typeof scheduledReports.$inferInsert; +/** + * One durable execution attempt for a scheduled report period. + * + * `scheduleKey` is deliberately stored instead of inferred by workers. The + * unique pair (scheduledReportId, runFor) is the database-level idempotency + * boundary: a queue redelivery, process restart, or two dispatchers may all + * ask for the same period without creating a second output. + */ +export const scheduledReportRuns = pgTable( + "scheduled_report_runs", + { + id: uuid("id").primaryKey().defaultRandom(), + scheduledReportId: uuid("scheduled_report_id") + .notNull() + .references(() => scheduledReports.id, { onDelete: "cascade" }), + scheduleKey: text("schedule_key").notNull(), + runFor: timestamp("run_for", { withTimezone: true }).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 }), + outputRef: text("output_ref"), + lastError: text("last_error"), + createdAt: timestamp("created_at", { withTimezone: true }) + .notNull() + .defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (t) => ({ + scheduledReportRunsIdentityIdx: index("scheduled_report_runs_identity_idx").on( + t.scheduledReportId, + t.runFor, + ), + scheduledReportRunsReadyIdx: index("scheduled_report_runs_ready_idx").on( + t.status, + t.nextAttemptAt, + ), + scheduledReportRunsLeaseIdx: index("scheduled_report_runs_lease_idx").on( + t.status, + t.leaseUntil, + ), + }), +); + +export type ScheduledReportRun = typeof scheduledReportRuns.$inferSelect; +export type NewScheduledReportRun = typeof scheduledReportRuns.$inferInsert; + // --------------------------------------------------------------------------- // Market Watchers // --------------------------------------------------------------------------- @@ -677,4 +728,3 @@ export const referrals = pgTable( export type Referral = typeof referrals.$inferSelect; export type NewReferral = typeof referrals.$inferInsert; - diff --git a/src/metrics/registry.ts b/src/metrics/registry.ts index 3198bf37..5fbce17e 100644 --- a/src/metrics/registry.ts +++ b/src/metrics/registry.ts @@ -58,6 +58,26 @@ export const indexerLagLedgers = new Gauge({ registers: [register], }); +export const scheduledReportRunsTotal = new Counter({ + name: "scheduled_report_runs_total", + help: "Scheduled report runs by terminal outcome", + labelNames: ["status"] as const, + registers: [register], +}); + +export const scheduledReportRetriesTotal = new Counter({ + name: "scheduled_report_retries_total", + help: "Scheduled report retry attempts by reason", + labelNames: ["reason"] as const, + registers: [register], +}); + +export const scheduledReportLeaseConflictsTotal = new Counter({ + name: "scheduled_report_lease_conflicts_total", + help: "Scheduled report jobs skipped because another worker owns the lease", + 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 7e82e4f1..a46d41b9 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -15,6 +15,7 @@ export const webhookQueueName = "webhook-deliveries"; export const backupVerificationQueueName = "backup-verification"; export const reconciliationQueueName = "reconciliation"; export const marketResolutionQueueName = "market-resolution"; +export const scheduledReportQueueName = "scheduled-report-runs"; export const webhookQueue = new Queue(webhookQueueName, { // IORedis types conflict with BullMQ @@ -36,4 +37,9 @@ export const marketResolutionQueue = new Queue(marketResolutionQueueName, { connection: redisConnection, }); +export const scheduledReportQueue = new Queue(scheduledReportQueueName, { + // IORedis types conflict with BullMQ + connection: redisConnection, +}); + export { Queue, Worker, QueueEvents }; diff --git a/src/routes/users.ts b/src/routes/users.ts index ed28cab9..72d650d4 100644 --- a/src/routes/users.ts +++ b/src/routes/users.ts @@ -326,20 +326,6 @@ usersRouter.get( } const { address } = paramsParse.data; - // Validate and coerce query parameters with zod. - { reqId, stellarAddress: req.params.stellarAddress }, - "user_profile_validation_failed", - ); - return res.status(400).json({ - error: { - code: "validation_error", - message: "invalid stellar address", - requestId: reqId, - }, - }); - } - - // ── 2. Service call ────────────────────────────────────────────────── // Validate and coerce query parameters with zod. const queryParse = userPredictionsQuerySchema.safeParse(req.query); if (!queryParse.success) { @@ -355,6 +341,7 @@ usersRouter.get( requestId: reqId, }, }); + } const { status, cursor, limit: rawLimit } = queryParse.data; // clampLimit is a belt-and-suspenders guard; zod already enforces 1–100. const limit = clampLimit(rawLimit); @@ -367,13 +354,6 @@ usersRouter.get( return res.status(404).json({ error: { code: "not_found", requestId: reqId } }); } - const page = await getUserPredictions(user.id, { status, limit, cursor }); - const user = await getUserByAddress(address); - if (!user) { - logger.debug({ reqId, address }, "predictions_user_not_found"); - return res.status(404).json({ error: { code: "not_found", requestId: reqId } }); - } - const page = await getUserPredictions(user.id, { status, limit, cursor }); logger.info( diff --git a/src/services/scheduledReportJobService.ts b/src/services/scheduledReportJobService.ts new file mode 100644 index 00000000..e7bb47db --- /dev/null +++ b/src/services/scheduledReportJobService.ts @@ -0,0 +1,259 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, isNull, lte, lt, or, sql } from "drizzle-orm"; +import type { Db } from "../db"; +import { db as defaultDb } from "../db"; +import { + scheduledReportRuns, + type ScheduledReportRun, +} from "../db/schema"; +import { + scheduledReportLeaseConflictsTotal, + scheduledReportRetriesTotal, + scheduledReportRunsTotal, +} from "../metrics/registry"; +import { logger } from "../config/logger"; +import { scheduledReportQueue } 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 ScheduledReportRunStatus = "pending" | "running" | "retryable" | "succeeded" | "failed"; + +export interface ReportRunJobData { + runId: string; + scheduledReportId: string; + runFor: string; + scheduleKey: string; +} + +export interface GeneratedReport { + /** Stable reference to the stored output, never the report bytes themselves. */ + outputRef: string; +} + +export interface ReportGeneratorInput { + runId: string; + scheduledReportId: string; + runFor: Date; + scheduleKey: string; + attempt: number; +} + +export type ReportGenerator = (input: ReportGeneratorInput) => Promise; + +export interface ReportRunRepository { + createOrGetRun(input: { + scheduledReportId: string; + scheduleKey: string; + runFor: Date; + }): Promise; + claimRun(runId: string, leaseToken: string, now: Date, leaseMs: number): Promise; + markSucceeded(runId: string, leaseToken: string, outputRef: string, now: Date): Promise; + markFailed(input: { + runId: string; + leaseToken: string; + error: string; + now: Date; + maxAttempts: number; + nextAttemptAt: Date; + }): Promise<{ status: "retryable" | "failed"; attempt: number } | null>; +} + +type QueueLike = { + add(name: string, data: ReportRunJobData, options?: Record): Promise; +}; + +/** + * Production repository. Every state transition includes the lease token, so + * a timed-out worker can never complete a run after a replacement owns it. + */ +export class DrizzleReportRunRepository implements ReportRunRepository { + constructor(private readonly database: Db = defaultDb) {} + + async createOrGetRun(input: { + scheduledReportId: string; + scheduleKey: string; + runFor: Date; + }): Promise { + await this.database + .insert(scheduledReportRuns) + .values({ + scheduledReportId: input.scheduledReportId, + scheduleKey: input.scheduleKey, + runFor: input.runFor, + status: "pending", + nextAttemptAt: input.runFor, + }) + .onConflictDoNothing({ + target: [scheduledReportRuns.scheduledReportId, scheduledReportRuns.runFor], + }); + + const [run] = await this.database + .select() + .from(scheduledReportRuns) + .where( + and( + eq(scheduledReportRuns.scheduledReportId, input.scheduledReportId), + eq(scheduledReportRuns.runFor, input.runFor), + ), + ) + .limit(1); + + if (!run) throw new Error("scheduled report run could not be created or loaded"); + return run; + } + + async claimRun(runId: string, leaseToken: string, now: Date, leaseMs: number): Promise { + const leaseUntil = new Date(now.getTime() + leaseMs); + const [run] = await this.database + .update(scheduledReportRuns) + .set({ + status: "running", + attempt: sql`${scheduledReportRuns.attempt} + 1`, + leaseToken, + leaseUntil, + startedAt: sql`COALESCE(${scheduledReportRuns.startedAt}, ${now})`, + updatedAt: now, + }) + .where( + and( + eq(scheduledReportRuns.id, runId), + or( + eq(scheduledReportRuns.status, "pending"), + and(eq(scheduledReportRuns.status, "retryable"), or(isNull(scheduledReportRuns.nextAttemptAt), lte(scheduledReportRuns.nextAttemptAt, now))), + and(eq(scheduledReportRuns.status, "running"), or(isNull(scheduledReportRuns.leaseUntil), lt(scheduledReportRuns.leaseUntil, now))), + ), + ), + ) + .returning(); + + if (!run) scheduledReportLeaseConflictsTotal.inc(); + return run ?? null; + } + + async markSucceeded(runId: string, leaseToken: string, outputRef: string, now: Date): Promise { + const result = await this.database + .update(scheduledReportRuns) + .set({ status: "succeeded", outputRef, completedAt: now, leaseToken: null, leaseUntil: null, updatedAt: now }) + .where(and(eq(scheduledReportRuns.id, runId), eq(scheduledReportRuns.leaseToken, leaseToken), eq(scheduledReportRuns.status, "running"))) + .returning({ id: scheduledReportRuns.id }); + const succeeded = result.length === 1; + if (succeeded) scheduledReportRunsTotal.inc({ status: "succeeded" }); + return succeeded; + } + + async markFailed(input: { + runId: string; + leaseToken: string; + error: string; + now: Date; + maxAttempts: number; + nextAttemptAt: Date; + }): Promise<{ status: "retryable" | "failed"; attempt: number } | null> { + const [run] = await this.database + .update(scheduledReportRuns) + .set({ + status: sql`CASE WHEN ${scheduledReportRuns.attempt} >= ${input.maxAttempts} THEN 'failed' ELSE 'retryable' END`, + lastError: input.error.slice(0, 2_000), + nextAttemptAt: input.nextAttemptAt, + completedAt: sql`CASE WHEN ${scheduledReportRuns.attempt} >= ${input.maxAttempts} THEN ${input.now} ELSE NULL END`, + leaseToken: null, + leaseUntil: null, + updatedAt: input.now, + }) + .where(and(eq(scheduledReportRuns.id, input.runId), eq(scheduledReportRuns.leaseToken, input.leaseToken), eq(scheduledReportRuns.status, "running"))) + .returning({ status: scheduledReportRuns.status, attempt: scheduledReportRuns.attempt }); + + if (!run) return null; + const status = run.status as "retryable" | "failed"; + scheduledReportRetriesTotal.inc({ reason: status === "retryable" ? "error" : "exhausted" }); + if (status === "failed") scheduledReportRunsTotal.inc({ status }); + return { status, attempt: run.attempt }; + } +} + +export function buildScheduleKey(scheduledReportId: string, runFor: Date): string { + if (!scheduledReportId || Number.isNaN(runFor.getTime())) throw new Error("valid schedule identity and run time are required"); + return `${scheduledReportId}:${runFor.toISOString()}`; +} + +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)); +} + +export async function enqueueScheduledReportRun( + scheduledReportId: string, + runFor: Date, + repository: ReportRunRepository = new DrizzleReportRunRepository(), + queue: QueueLike = scheduledReportQueue, +): Promise<{ run: ScheduledReportRun; enqueued: boolean }> { + const scheduleKey = buildScheduleKey(scheduledReportId, runFor); + const run = await repository.createOrGetRun({ scheduledReportId, scheduleKey, runFor }); + if (run.status === "succeeded" || run.status === "failed") return { run, enqueued: false }; + + await queue.add("generate", { + runId: run.id, + scheduledReportId, + runFor: runFor.toISOString(), + scheduleKey, + }, { jobId: scheduleKey, removeOnComplete: false, removeOnFail: false }); + return { run, enqueued: true }; +} + +export interface RunCoordinatorOptions { + leaseMs?: number; + maxAttempts?: number; + retryBaseMs?: number; + now?: () => Date; + queue?: QueueLike; +} + +export class ScheduledReportRunCoordinator { + private readonly leaseMs: number; + private readonly maxAttempts: number; + private readonly retryBaseMs: number; + private readonly now: () => Date; + private readonly queue: QueueLike; + + constructor(private readonly repository: ReportRunRepository, options: RunCoordinatorOptions = {}) { + 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 ?? scheduledReportQueue; + if (!Number.isInteger(this.maxAttempts) || this.maxAttempts < 1) throw new Error("maxAttempts must be positive"); + } + + async process(job: ReportRunJobData, generate: ReportGenerator): Promise<"skipped" | "succeeded" | "retryable" | "failed"> { + const leaseToken = randomUUID(); + const claimed = await this.repository.claimRun(job.runId, leaseToken, this.now(), this.leaseMs); + if (!claimed) return "skipped"; + + try { + const output = await generate({ + runId: claimed.id, + scheduledReportId: claimed.scheduledReportId, + runFor: claimed.runFor, + scheduleKey: claimed.scheduleKey, + attempt: claimed.attempt, + }); + if (!output.outputRef || output.outputRef.length > 2_000) throw new Error("report generator returned an invalid output reference"); + const accepted = await this.repository.markSucceeded(claimed.id, leaseToken, output.outputRef, this.now()); + return accepted ? "succeeded" : "skipped"; + } catch (error) { + const message = error instanceof Error ? error.message : "unknown report generation failure"; + const retryAt = new Date(this.now().getTime() + retryDelayMs(claimed.attempt, this.retryBaseMs)); + const failed = await this.repository.markFailed({ runId: 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("generate", job, { jobId: `${job.scheduleKey}:retry:${failed.attempt}`, delay: retryAt.getTime() - this.now().getTime(), removeOnComplete: false, removeOnFail: false }); + } + logger.warn({ runId: job.runId, attempt: failed.attempt, status: failed.status }, "scheduled report generation failed"); + return failed.status; + } + } +} diff --git a/src/workers/scheduledReportWorker.ts b/src/workers/scheduledReportWorker.ts new file mode 100644 index 00000000..1dc99281 --- /dev/null +++ b/src/workers/scheduledReportWorker.ts @@ -0,0 +1,68 @@ +import { Job, Worker } from "bullmq"; +import { logger } from "../config/logger"; +import { + DrizzleReportRunRepository, + type ReportGenerator, + type ReportRunJobData, + ScheduledReportRunCoordinator, + type ReportRunRepository, +} from "../services/scheduledReportJobService"; +import { redisConnection, scheduledReportQueueName } from "../queue"; + +export interface ScheduledReportWorkerOptions { + concurrency?: number; + leaseMs?: number; + maxAttempts?: number; + retryBaseMs?: number; + queue?: { add(name: string, data: ReportRunJobData, options?: Record): Promise }; +} + +/** + * BullMQ adapter for the durable run coordinator. The coordinator owns all + * correctness decisions; this class only translates queue jobs into calls. + */ +export class ScheduledReportWorker { + private worker: Worker | null = null; + private readonly coordinator: ScheduledReportRunCoordinator; + private readonly concurrency: number; + + constructor( + repository: ReportRunRepository = new DrizzleReportRunRepository(), + generator: ReportGenerator = async ({ scheduleKey }) => ({ outputRef: `pending://${scheduleKey}` }), + options: ScheduledReportWorkerOptions = {}, + ) { + this.coordinator = new ScheduledReportRunCoordinator(repository, options); + this.generator = generator; + this.concurrency = options.concurrency ?? 4; + } + + private readonly generator: ReportGenerator; + + start(): void { + if (this.worker) return; + this.worker = new Worker( + scheduledReportQueueName, + async (job: Job) => this.coordinator.process(job.data, this.generator), + { connection: redisConnection, concurrency: this.concurrency }, + ); + this.worker.on("failed", (job, error) => { + logger.error({ jobId: job?.id, runId: job?.data.runId, err: error.message }, "scheduled report queue job failed"); + }); + logger.info({ concurrency: this.concurrency }, "scheduled report worker started"); + } + + async stop(): Promise { + if (!this.worker) return; + await this.worker.close(); + this.worker = null; + logger.info("scheduled report worker stopped"); + } +} + +export function createScheduledReportWorker( + repository?: ReportRunRepository, + generator?: ReportGenerator, + options?: ScheduledReportWorkerOptions, +): ScheduledReportWorker { + return new ScheduledReportWorker(repository, generator, options); +} diff --git a/tests/scheduledReportJobService.test.ts b/tests/scheduledReportJobService.test.ts new file mode 100644 index 00000000..d791f83a --- /dev/null +++ b/tests/scheduledReportJobService.test.ts @@ -0,0 +1,161 @@ +jest.mock("../src/queue", () => ({ + scheduledReportQueue: { add: jest.fn().mockResolvedValue(undefined) }, +})); + +import { + buildScheduleKey, + DEFAULT_MAX_ATTEMPTS, + retryDelayMs, + ScheduledReportRunCoordinator, + type ReportRunJobData, + type ReportRunRepository, +} from "../src/services/scheduledReportJobService"; +import type { ScheduledReportRun } from "../src/db/schema"; + +function makeRun(overrides: Partial = {}): ScheduledReportRun { + const now = new Date("2026-08-24T10:00:00.000Z"); + return { + id: "run-1", + scheduledReportId: "schedule-1", + scheduleKey: "schedule-1:2026-08-24T10:00:00.000Z", + runFor: now, + status: "pending", + attempt: 0, + leaseToken: null, + leaseUntil: null, + startedAt: null, + completedAt: null, + nextAttemptAt: now, + outputRef: null, + lastError: null, + createdAt: now, + updatedAt: now, + ...overrides, + }; +} + +class FakeRepository implements ReportRunRepository { + run = makeRun(); + claimCalls: Array<{ runId: string; token: string }> = []; + successes: Array<{ runId: string; token: string; outputRef: string }> = []; + failures: Array<{ runId: string; token: string; error: string }> = []; + claimEnabled = true; + failStatus: "retryable" | "failed" = "retryable"; + + async createOrGetRun(): Promise { return this.run; } + + async claimRun(runId: string, token: string): Promise { + this.claimCalls.push({ runId, token }); + if (!this.claimEnabled) return null; + this.run = { ...this.run, status: "running", attempt: this.run.attempt + 1, leaseToken: token }; + return this.run; + } + + async markSucceeded(runId: string, token: string, outputRef: string): Promise { + this.successes.push({ runId, token, outputRef }); + if (this.run.leaseToken !== token || this.run.status !== "running") return false; + this.run = { ...this.run, status: "succeeded", outputRef, leaseToken: null }; + return true; + } + + async markFailed(input: { runId: string; leaseToken: string; error: string; maxAttempts: number }): Promise<{ status: "retryable" | "failed"; attempt: number } | null> { + this.failures.push({ runId: input.runId, token: input.leaseToken, error: input.error }); + if (this.run.leaseToken !== input.leaseToken || this.run.status !== "running") return null; + this.run = { ...this.run, status: this.failStatus, leaseToken: null, lastError: input.error }; + return { status: this.failStatus, attempt: this.run.attempt }; + } +} + +const job: ReportRunJobData = { + runId: "run-1", + scheduledReportId: "schedule-1", + runFor: "2026-08-24T10:00:00.000Z", + scheduleKey: "schedule-1:2026-08-24T10:00:00.000Z", +}; + +describe("scheduled report run identity", () => { + it("is stable for equivalent UTC instants", () => { + expect(buildScheduleKey("schedule-1", new Date("2026-08-24T10:00:00Z"))).toBe(job.scheduleKey); + expect(buildScheduleKey("schedule-1", new Date("2026-08-24T05:00:00-05:00"))).toBe(job.scheduleKey); + }); + + it("rejects missing identity and invalid dates", () => { + expect(() => buildScheduleKey("", new Date())).toThrow("valid schedule identity"); + expect(() => buildScheduleKey("schedule-1", new Date("invalid"))).toThrow("valid schedule identity"); + }); + + it("uses bounded exponential retry delays", () => { + expect(retryDelayMs(1, 100)).toBe(100); + expect(retryDelayMs(4, 100)).toBe(800); + expect(retryDelayMs(30, 100)).toBeLessThanOrEqual(60 * 60 * 1000); + expect(() => retryDelayMs(0)).toThrow("positive integer"); + expect(() => retryDelayMs(1, -1)).toThrow("non-negative"); + }); +}); + +describe("ScheduledReportRunCoordinator", () => { + function makeCoordinator(repository: FakeRepository, queue = { add: jest.fn().mockResolvedValue(undefined) }) { + return { coordinator: new ScheduledReportRunCoordinator(repository, { now: () => new Date("2026-08-24T10:00:00Z"), queue, retryBaseMs: 10 }), queue }; + } + + it("claims, generates, and completes exactly once", async () => { + const repository = new FakeRepository(); + const { coordinator, queue } = makeCoordinator(repository); + const result = await coordinator.process(job, async (input) => ({ outputRef: `blob://${input.scheduleKey}` })); + expect(result).toBe("succeeded"); + expect(repository.claimCalls).toHaveLength(1); + expect(repository.successes[0].outputRef).toContain(job.scheduleKey); + expect(queue.add).not.toHaveBeenCalled(); + }); + + it("skips an overlapping worker when the lease is owned elsewhere", async () => { + const repository = new FakeRepository(); + repository.claimEnabled = false; + const { coordinator } = makeCoordinator(repository); + const generator = jest.fn().mockResolvedValue({ outputRef: "blob://never" }); + await expect(coordinator.process(job, generator)).resolves.toBe("skipped"); + expect(generator).not.toHaveBeenCalled(); + }); + + it("re-enqueues a retry with a distinct queue id", async () => { + const repository = new FakeRepository(); + const { coordinator, queue } = makeCoordinator(repository); + const result = await coordinator.process(job, async () => { throw new Error("temporary timeout"); }); + expect(result).toBe("retryable"); + expect(repository.failures[0].error).toBe("temporary timeout"); + expect(queue.add).toHaveBeenCalledWith("generate", job, expect.objectContaining({ jobId: `${job.scheduleKey}:retry:1` })); + }); + + it("marks retry exhaustion terminal and does not enqueue more work", async () => { + const repository = new FakeRepository(); + repository.failStatus = "failed"; + const { coordinator, queue } = makeCoordinator(repository); + await expect(coordinator.process(job, async () => { throw new Error("permanent failure"); })).resolves.toBe("failed"); + expect(queue.add).not.toHaveBeenCalled(); + expect(repository.run.status).toBe("failed"); + }); + + it("does not acknowledge output when lease ownership is lost", async () => { + const repository = new FakeRepository(); + const { coordinator, queue } = makeCoordinator(repository); + const generator = jest.fn(async () => { + repository.run = { ...repository.run, leaseToken: "replacement-token", status: "running" }; + return { outputRef: "blob://old-worker" }; + }); + await expect(coordinator.process(job, generator)).resolves.toBe("skipped"); + expect(repository.successes).toHaveLength(1); + expect(queue.add).not.toHaveBeenCalled(); + }); + + it("rejects an invalid generator output and routes it through retry policy", async () => { + const repository = new FakeRepository(); + const { coordinator, queue } = makeCoordinator(repository); + await expect(coordinator.process(job, async () => ({ outputRef: "" }))).resolves.toBe("retryable"); + expect(repository.failures[0].error).toContain("invalid output reference"); + expect(queue.add).toHaveBeenCalledTimes(1); + }); + + it("keeps the default retry policy bounded", () => { + expect(DEFAULT_MAX_ATTEMPTS).toBe(3); + }); +}); diff --git a/tests/scheduledReportWorker.test.ts b/tests/scheduledReportWorker.test.ts new file mode 100644 index 00000000..036e6951 --- /dev/null +++ b/tests/scheduledReportWorker.test.ts @@ -0,0 +1,34 @@ +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: {}, + scheduledReportQueueName: "scheduled-report-runs", +})); + +import { Worker } from "bullmq"; +import { ScheduledReportWorker } from "../src/workers/scheduledReportWorker"; +import { ReportRunRepository } from "../src/services/scheduledReportJobService"; + +describe("ScheduledReportWorker", () => { + it("starts one BullMQ worker and is idempotent", () => { + const worker = new ScheduledReportWorker({} as ReportRunRepository); + worker.start(); + worker.start(); + expect(Worker).toHaveBeenCalledTimes(1); + }); + + it("closes the worker during stop and tolerates repeated stop", async () => { + const worker = new ScheduledReportWorker({} as ReportRunRepository); + 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); + }); +});