diff --git a/drizzle/0026_add_predictions_confirmer.sql b/drizzle/0026_add_predictions_confirmer.sql new file mode 100644 index 00000000..99c580ed --- /dev/null +++ b/drizzle/0026_add_predictions_confirmer.sql @@ -0,0 +1,8 @@ +-- Migration: add confirm_attempts and last_error columns to predictions +-- These columns support the predictionsConfirmer worker which transitions +-- pending predictions to confirmed (or failed after max attempts) by joining +-- against indexer_events on txHash. + +ALTER TABLE "predictions" + ADD COLUMN IF NOT EXISTS "confirm_attempts" integer NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS "last_error" text; diff --git a/drizzle/meta/0003_snapshot.json b/drizzle/meta/0003_snapshot.json new file mode 100644 index 00000000..18011caf --- /dev/null +++ b/drizzle/meta/0003_snapshot.json @@ -0,0 +1,13 @@ +{ + "id": "f5a6b7c8-d9e0-1f2a-3b4c-5d6e7f8a9b0c", + "prevId": "d4e5f6a7-b8c9-0d1e-2f3a-4b5c6d7e8f90", + "version": "7", + "dialect": "postgresql", + "tables": {}, + "schemas": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 3f08e4b6..c25cc95d 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -36,6 +36,13 @@ "when": 1753718153907, "tag": "0025_users_filter_idx", "breakpoints": true + }, + { + "idx": 26, + "version": "7", + "when": 1785335000000, + "tag": "0026_add_predictions_confirmer", + "breakpoints": true } ] } diff --git a/src/config/env-schema.ts b/src/config/env-schema.ts index 8ec26e25..096a5ed1 100644 --- a/src/config/env-schema.ts +++ b/src/config/env-schema.ts @@ -101,6 +101,11 @@ const baseSchema = z.object({ 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(""), diff --git a/src/db/schema.ts b/src/db/schema.ts index 5f63780d..e0ef3450 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -184,6 +184,17 @@ export const predictions = pgTable("predictions", { createdAt: timestamp("created_at", { withTimezone: true }) .notNull() .defaultNow(), + /** + * Number of confirmation attempts made by the predictionsConfirmer worker. + * Incremented each tick when no matching indexer event is found. + * After MAX_CONFIRM_ATTEMPTS (3) the prediction is marked as failed. + */ + confirmAttempts: integer("confirm_attempts").notNull().default(0), + /** + * Error message from the most recent failed confirmation attempt. + * Set when the prediction transitions to failed after exhausting all attempts. + */ + lastError: text("last_error"), }); /** diff --git a/src/server.ts b/src/server.ts index 9f37c330..b2941e11 100644 --- a/src/server.ts +++ b/src/server.ts @@ -9,12 +9,14 @@ import { marketResolverWorker } from "./workers/marketResolver"; import { backupVerificationWorker } from "./workers/backupVerificationWorker"; import { reconciliationWorker } from "./workers/reconciliationWorker"; import { startSlowQueryAlerter } from "./workers/slowQueryAlerter"; +import { startPredictionsConfirmer } from "./workers/predictionsConfirmer"; import { drainSearchRequests } from "./routes/search"; import { drainExportsRequests } from "./routes/exports"; const app = createApp(); let webhookWorker: WebhookWorker | null = null; let probeHandle: ReturnType | null = null; +let predictionsConfirmerHandle: ReturnType | null = null; connectWithRetry() .then(() => { @@ -24,6 +26,7 @@ connectWithRetry() backupVerificationWorker.start(); reconciliationWorker.start(); startSlowQueryAlerter(); + predictionsConfirmerHandle = startPredictionsConfirmer(); probeHandle = startIndexerHealthProbe(); const server = app.listen(env.PORT, () => { diff --git a/src/services/webhookCatalog.ts b/src/services/webhookCatalog.ts index 72361e85..2831f268 100644 --- a/src/services/webhookCatalog.ts +++ b/src/services/webhookCatalog.ts @@ -113,6 +113,35 @@ export const DisputeOpenedPayloadSchema = WebhookEnvelopeSchema.extend({ export type DisputeOpenedPayload = z.infer; +// --------------------------------------------------------------------------- +// prediction.confirmed +// --------------------------------------------------------------------------- + +/** + * Emitted when the predictionsConfirmer worker detects a matching indexer + * event for a pending prediction and transitions it to confirmed. + * + * The dispatcher fans this event out to every subscription which subscribes to + * "prediction.confirmed" or "*". + */ +export const PredictionConfirmedPayloadSchema = WebhookEnvelopeSchema.extend({ + event: z.literal("prediction.confirmed"), + /** UUID of the confirmed prediction. */ + predictionId: z.string().uuid(), + /** Primary key of the market the prediction belongs to. */ + marketId: z.string(), + /** UUID of the user who made the prediction. */ + userId: z.string().uuid(), + /** The outcome the user predicted (e.g. "yes" or "no"). */ + outcome: z.string(), + /** The amount staked on the prediction (string to preserve precision). */ + amount: z.string(), + /** On-chain transaction hash of the bet_placed event. */ + txHash: z.string(), +}); + +export type PredictionConfirmedPayload = z.infer; + // --------------------------------------------------------------------------- // Catalog registry // --------------------------------------------------------------------------- @@ -130,6 +159,7 @@ export type DisputeOpenedPayload = z.infer; export const WEBHOOK_EVENT_SCHEMAS = { "market.resolved": MarketResolvedPayloadSchema, "dispute.opened": DisputeOpenedPayloadSchema, + "prediction.confirmed": PredictionConfirmedPayloadSchema, } as const; /** All registered event-type strings, derived from the catalog at compile time. */ @@ -141,7 +171,8 @@ export type WebhookEventType = keyof typeof WEBHOOK_EVENT_SCHEMAS; /** Union of every concrete payload type. */ export type WebhookPayload = | MarketResolvedPayload - | DisputeOpenedPayload; + | DisputeOpenedPayload + | PredictionConfirmedPayload; // --------------------------------------------------------------------------- // Validation helpers diff --git a/src/workers/predictionsConfirmer.ts b/src/workers/predictionsConfirmer.ts new file mode 100644 index 00000000..63ef2259 --- /dev/null +++ b/src/workers/predictionsConfirmer.ts @@ -0,0 +1,541 @@ +/** + * predictionsConfirmer.ts + * + * Background worker that transitions pending predictions to confirmed (or + * failed) by joining against indexer_events on txHash. + * + * After the indexer ingests a bet_placed event, the corresponding pending row + * in predictions must transition to confirmed. This worker: + * + * 1. Polls on a configurable interval (PREDICTION_CONFIRM_INTERVAL_MS). + * 2. Finds pending predictions where a matching indexer_event exists (same + * txHash) and batch-updates them to "confirmed". + * 3. For unmatched pending predictions, increments `confirmAttempts`. After + * MAX_CONFIRM_ATTEMPTS (default 3) the row transitions to "failed" with a + * descriptive lastError. + * 4. Emits a `prediction.confirmed` webhook event for each confirmed row. + * + * Design decisions + * ──────────────── + * • Batches are processed in a single database transaction per tick so that + * partial failures never leave inconsistent state. + * • Failed rows (attempts exhausted) do not block sibling rows — each + * row's fate is determined independently via bulk SQL. + * • The webhook is emitted exactly once per confirmed prediction via the + * `dispatchEvent` path; it fans out to all matching subscriptions. + * • The webhook emitter is injected as a constructor dependency so unit + * tests can mock the webhook path without module-level imports. + * • The repo uses DISTINCT when joining indexer_events to guard against + * duplicate prediction IDs when a txHash has multiple event rows. + * • The worker is structured as an injectable service class (for unit tests) + * and a standalone main() entry point (for production). + */ + +import { randomUUID } from "node:crypto"; +import { and, eq, inArray, sql } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/node-postgres"; +import { env } from "../config/env"; +import { logger } from "../config/logger"; +import { pool } from "../db/client"; +import { predictions, indexerEvents } from "../db/schema"; +import type { Db } from "../db"; + +// ─── Constants ─────────────────────────────────────────────────────────────── + +/** + * Maximum number of confirmation attempts before a prediction is marked as + * permanently failed. Used as the default in the Drizzle repo; the service + * class accepts an override via constructor opts. + */ +export const DEFAULT_MAX_ATTEMPTS = 3; + +// ─── Webhook emitter type ───────────────────────────────────────────────────── + +/** + * Function signature for emitting a webhook event. Injected into the service + * so it can be mocked in unit tests without module-level mocking. + */ +export type WebhookEmitter = ( + eventType: string, + payload: Record, +) => Promise; + +/** Default emitter that delegates to the production webhook dispatcher. */ +export const defaultWebhookEmitter: WebhookEmitter = async ( + eventType: string, + payload: Record, +): Promise => { + // We need a db handle; in production the service creates one internally. + // This function signature matches what dispatchEvent expects minus the db arg. + // For the production path the caller provides the full db reference. + throw new Error( + "defaultWebhookEmitter should not be called directly — use the full signature", + ); +}; + +// ─── Repository interface ──────────────────────────────────────────────────── + +/** + * Thin repository to decouple the confirmer service from Drizzle so it can + * be unit-tested without a live database. + */ +export interface PredictionsConfirmerRepo { + /** + * Returns the unique IDs of pending predictions that have a matching txHash + * in indexer_events. These should be transitioned to "confirmed". + * Uses DISTINCT to avoid duplicates when a txHash has multiple event rows. + */ + findConfirmedCandidateIds(batchSize: number): Promise; + + /** + * For pending predictions without a matching indexer event, increment + * confirmAttempts. Only targets rows with confirmAttempts < maxAttempts. + * Returns the number of rows updated. + */ + incrementAttempts(maxAttempts: number, batchSize: number): Promise; + + /** + * Mark predictions whose confirmAttempts >= maxAttempts as "failed" with + * a descriptive lastError. Returns the IDs of the failed predictions. + */ + markFailed(maxAttempts: number, batchSize: number): Promise; + + /** + * Batch-update the given prediction IDs to status "confirmed", + * reset confirmAttempts to 0, and clear lastError. + * Returns the full row data for the confirmed predictions. + */ + confirmBatch(predictionIds: string[]): Promise; + + /** + * Execute all three mutations (confirm, increment, fail) inside a single + * database transaction. Returns the aggregated results. + */ + transactionalTick( + batchSize: number, + maxAttempts: number, + ): Promise<{ + confirmed: ConfirmedPrediction[]; + incremented: number; + failedIds: string[]; + }>; +} + +/** Shape of a prediction row returned after confirmation. */ +export interface ConfirmedPrediction { + id: string; + marketId: string; + userId: string; + outcome: string; + amount: string; + txHash: string; + status: string; +} + +// ─── Poll result ───────────────────────────────────────────────────────────── + +export interface PollResult { + processed: number; + confirmed: number; + incremented: number; + failed: number; + webhookErrors: number; +} + +// ─── Service ───────────────────────────────────────────────────────────────── + +export class PredictionsConfirmerService { + private readonly repo: PredictionsConfirmerRepo; + private readonly batchSize: number; + private readonly maxAttempts: number; + private readonly emitWebhook: (db: Db, eventType: string, payload: Record) => Promise; + private readonly db: Db; + + constructor( + repo: PredictionsConfirmerRepo, + opts?: { + batchSize?: number; + maxAttempts?: number; + db?: Db; + /** Webhook emitter injected for testability. Defaults to dispatchEvent. */ + emitWebhook?: (db: Db, eventType: string, payload: Record) => Promise; + }, + ) { + this.repo = repo; + this.batchSize = opts?.batchSize ?? env.PREDICTION_CONFIRM_BATCH_SIZE; + this.maxAttempts = opts?.maxAttempts ?? env.PREDICTION_CONFIRM_MAX_ATTEMPTS; + this.db = opts?.db ?? ({} as Db); + // Default emitter uses dispatchEvent; lazily imported to avoid + // pulling in BullMQ/Redis at load time (which causes test hangs + // when the queue module is not mocked). + this.emitWebhook = + opts?.emitWebhook ?? + (async (db: Db, eventType: string, payload: Record) => { + const { dispatchEvent } = await import("../services/webhookDispatcher"); + await dispatchEvent(db, eventType, payload); + }); + } + + /** + * One poll cycle — all DB work inside a single transactional tick: + * 1. Confirm pending predictions with matching indexer_events. + * 2. Increment confirmAttempts for unmatched pending predictions. + * 3. Mark predictions as failed when attempts >= maxAttempts. + * 4. Emit `prediction.confirmed` webhooks for confirmed rows. + */ + async pollOnce(): Promise { + const correlationId = randomUUID(); + const start = Date.now(); + + logger.info({ correlationId }, "predictions_confirmer: poll starting"); + + // ── Single transactional DB tick ────────────────────────────────────────── + const { confirmed, incremented, failedIds } = + await this.repo.transactionalTick(this.batchSize, this.maxAttempts); + + const confirmedCount = confirmed.length; + const failedCount = failedIds.length; + + if (confirmedCount > 0) { + logger.info( + { correlationId, confirmedCount }, + "predictions_confirmer: confirmed predictions", + ); + } + if (incremented > 0) { + logger.info( + { correlationId, incremented }, + "predictions_confirmer: incremented attempts", + ); + } + if (failedCount > 0) { + logger.info( + { correlationId, failedCount }, + "predictions_confirmer: marked failed", + ); + } + + // ── Emit webhooks ───────────────────────────────────────────────────────── + let webhookErrors = 0; + for (const row of confirmed) { + try { + await this.emitWebhook(this.db, "prediction.confirmed", { + predictionId: row.id, + marketId: row.marketId, + userId: row.userId, + outcome: row.outcome, + amount: row.amount, + txHash: row.txHash, + }); + } catch (err) { + logger.error( + { correlationId, predictionId: row.id, err }, + "predictions_confirmer: webhook dispatch failed", + ); + webhookErrors++; + } + } + + const durationMs = Date.now() - start; + const totalProcessed = confirmedCount + incremented + failedCount; + + logger.info( + { + correlationId, + processed: totalProcessed, + confirmed: confirmedCount, + incremented, + failed: failedCount, + webhookErrors, + durationMs, + }, + "predictions_confirmer: poll complete", + ); + + return { + processed: totalProcessed, + confirmed: confirmedCount, + incremented, + failed: failedCount, + webhookErrors, + }; + } +} + +// ─── Drizzle repository ──────────────────────────────────────────────────── + +/** + * Production implementation of PredictionsConfirmerRepo backed by Drizzle ORM. + * All three batch mutations run inside a single Drizzle transaction so partial + * failures never leave the database in an inconsistent state. + */ +export class DrizzlePredictionsConfirmerRepo implements PredictionsConfirmerRepo { + private readonly db: ReturnType; + + constructor(db: ReturnType) { + this.db = db; + } + + async findConfirmedCandidateIds(batchSize: number): Promise { + // DISTINCT ensures a prediction ID appears at most once even when its + // txHash has multiple indexer_event rows (different op_index values). + const rows = await this.db + .selectDistinct({ id: predictions.id }) + .from(predictions) + .innerJoin( + indexerEvents, + eq(predictions.txHash, indexerEvents.txHash), + ) + .where( + and( + eq(predictions.status, "pending"), + sql`${predictions.txHash} != ''`, + ), + ) + .limit(batchSize); + + return rows.map((r) => r.id); + } + + async confirmBatch(predictionIds: string[]): Promise { + if (predictionIds.length === 0) return []; + + const rows = await this.db + .update(predictions) + .set({ + status: "confirmed", + confirmAttempts: 0, + lastError: null, + }) + .where(inArray(predictions.id, predictionIds)) + .returning({ + id: predictions.id, + marketId: predictions.marketId, + userId: predictions.userId, + outcome: predictions.outcome, + amount: predictions.amount, + txHash: predictions.txHash, + status: predictions.status, + }); + + return rows; + } + + async incrementAttempts( + maxAttempts: number, + _batchSize: number, + ): Promise { + const result = await this.db + .update(predictions) + .set({ + confirmAttempts: sql`${predictions.confirmAttempts} + 1`, + }) + .where( + and( + eq(predictions.status, "pending"), + sql`${predictions.txHash} != ''`, + sql`${predictions.confirmAttempts} < ${maxAttempts}`, + sql`NOT EXISTS ( + SELECT 1 FROM ${indexerEvents} + WHERE ${indexerEvents.txHash} = ${predictions.txHash} + )`, + ), + ); + + return result.rowCount ?? 0; + } + + async markFailed( + maxAttempts: number, + _batchSize: number, + ): Promise { + const rows = await this.db + .update(predictions) + .set({ + status: "failed", + lastError: `No matching indexer event found after ${maxAttempts} attempts`, + }) + .where( + and( + eq(predictions.status, "pending"), + sql`${predictions.txHash} != ''`, + sql`${predictions.confirmAttempts} >= ${maxAttempts}`, + sql`NOT EXISTS ( + SELECT 1 FROM ${indexerEvents} + WHERE ${indexerEvents.txHash} = ${predictions.txHash} + )`, + ), + ) + .returning({ id: predictions.id }); + + return rows.map((r) => r.id); + } + + async getPredictionsByIds(ids: string[]): Promise { + if (ids.length === 0) return []; + + const rows = await this.db + .select({ + id: predictions.id, + marketId: predictions.marketId, + userId: predictions.userId, + outcome: predictions.outcome, + amount: predictions.amount, + txHash: predictions.txHash, + status: predictions.status, + }) + .from(predictions) + .where(inArray(predictions.id, ids)); + + return rows; + } + + /** + * Execute the full tick inside a single Drizzle transaction: + * 1. Find candidate IDs (matching indexer_events) → confirm batch. + * 2. Increment attempts for unmatched pending predictions. + * 3. Mark attempts-exhausted predictions as failed. + * + * If the transaction rolls back, none of the mutations are persisted. + */ + async transactionalTick( + batchSize: number, + maxAttempts: number, + ): Promise<{ + confirmed: ConfirmedPrediction[]; + incremented: number; + failedIds: string[]; + }> { + return this.db.transaction(async (tx) => { + // Use the same tx instance for all repo operations. + const txRepo = new DrizzlePredictionsConfirmerRepo( + tx as unknown as ReturnType, + ); + + // Phase 1: Find and confirm matched predictions. + const candidateIds = await txRepo.findConfirmedCandidateIds(batchSize); + const confirmed = await txRepo.confirmBatch(candidateIds); + + // Phase 2: Increment attempts for unmatched predictions. + const incremented = await txRepo.incrementAttempts(maxAttempts, batchSize); + + // Phase 3: Mark exhausted predictions as failed. + const failedIds = await txRepo.markFailed(maxAttempts, batchSize); + + return { confirmed, incremented, failedIds }; + }); + } +} + +// ─── Standalone entry point ─────────────────────────────────────────────────── + +async function main(): Promise { + const db = drizzle(pool); + const repo = new DrizzlePredictionsConfirmerRepo(db); + const service = new PredictionsConfirmerService(repo, { + db: db as unknown as Db, + }); + const intervalMs = env.PREDICTION_CONFIRM_INTERVAL_MS; + + let shuttingDown = false; + let activeTick: Promise = Promise.resolve(); + + const requestShutdown = (signal: NodeJS.Signals): void => { + if (shuttingDown) return; + shuttingDown = true; + logger.info( + { signal }, + "predictions_confirmer: shutdown requested; draining current tick", + ); + }; + + process.on("SIGTERM", requestShutdown); + process.on("SIGINT", requestShutdown); + + logger.info( + { + intervalMs, + batchSize: env.PREDICTION_CONFIRM_BATCH_SIZE, + maxAttempts: env.PREDICTION_CONFIRM_MAX_ATTEMPTS, + }, + "predictions_confirmer: worker started", + ); + + const sleep = (ms: number): Promise => + new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + const onSignal = (): void => { + clearTimeout(timer); + resolve(); + }; + process.once("SIGTERM", onSignal); + process.once("SIGINT", onSignal); + }); + + while (!shuttingDown) { + try { + activeTick = service.pollOnce(); + await activeTick; + } catch (err) { + logger.error({ err }, "predictions_confirmer: tick failed"); + } + if (shuttingDown) break; + await sleep(intervalMs); + } + + await activeTick.catch(() => undefined); + await pool.end(); + logger.info({}, "predictions_confirmer: worker stopped"); +} + +if (require.main === module) { + main() + .then(() => process.exit(0)) + .catch((err) => { + logger.error({ err }, "predictions_confirmer: worker crashed"); + process.exit(1); + }); +} + +// ─── Server integration ─────────────────────────────────────────────────────── + +/** + * Start the predictions confirmer worker on the configured interval. + * + * Returns the interval handle so callers can cancel it during shutdown. + * This function is imported and called from src/server.ts. + */ +export function startPredictionsConfirmer( + intervalMs: number = env.PREDICTION_CONFIRM_INTERVAL_MS, +): NodeJS.Timeout { + const db = drizzle(pool); + const repo = new DrizzlePredictionsConfirmerRepo(db); + const service = new PredictionsConfirmerService(repo, { + db: db as unknown as Db, + }); + + // Run immediately on start, then on the interval. + const tick = async (): Promise => { + try { + await service.pollOnce(); + } catch (err) { + logger.error({ err }, "predictions_confirmer: tick failed"); + } + }; + + tick(); + + const id = setInterval(tick, intervalMs); + id.unref(); + + logger.info( + { + intervalMs, + batchSize: env.PREDICTION_CONFIRM_BATCH_SIZE, + maxAttempts: env.PREDICTION_CONFIRM_MAX_ATTEMPTS, + }, + "predictions_confirmer: worker started (server mode)", + ); + + return id; +} diff --git a/tests/predictionsConfirmer.test.ts b/tests/predictionsConfirmer.test.ts new file mode 100644 index 00000000..ebbdda26 --- /dev/null +++ b/tests/predictionsConfirmer.test.ts @@ -0,0 +1,355 @@ +/** + * predictionsConfirmer.test.ts + * + * Unit tests for the PredictionsConfirmerService using a mock repository + * and a mock webhook emitter. The service is tested in isolation from the + * database layer so tests are fast and deterministic. + * + * Coverage: + * - Happy path: matching indexer event → prediction confirmed + webhook emitted + * - Batch confirmations with multiple predictions + * - Incrementing attempts for unmatched predictions + * - Marking predictions as failed after max attempts + * - Mixed outcomes: some confirmed, some incremented, some failed + * - Empty state (no pending predictions) + * - Webhook dispatch errors do not crash the tick + * - Idempotency: already-confirmed predictions are not re-processed + * - Configurable maxAttempts and batchSize + */ + +import { + PredictionsConfirmerService, + DEFAULT_MAX_ATTEMPTS, + type PredictionsConfirmerRepo, + type ConfirmedPrediction, + type PollResult, + type WebhookEmitter, +} from "../src/workers/predictionsConfirmer"; + +// ─── Mock repository factory ────────────────────────────────────────────── + +function makeRepo( + overrides?: Partial, +): jest.Mocked { + return { + findConfirmedCandidateIds: jest + .fn, [number]>() + .mockResolvedValue([]), + confirmBatch: jest + .fn, [string[]]>() + .mockResolvedValue([]), + incrementAttempts: jest + .fn, [number, number]>() + .mockResolvedValue(0), + markFailed: jest + .fn, [number, number]>() + .mockResolvedValue([]), + getPredictionsByIds: jest + .fn, [string[]]>() + .mockResolvedValue([]), + transactionalTick: jest + .fn< + Promise<{ + confirmed: ConfirmedPrediction[]; + incremented: number; + failedIds: string[]; + }>, + [number, number] + >() + .mockResolvedValue({ + confirmed: [], + incremented: 0, + failedIds: [], + }), + ...overrides, + }; +} + +// ─── Helpers ────────────────────────────────────────────────────────────── + +function makeService( + repoOverrides?: Partial, + opts?: { + batchSize?: number; + maxAttempts?: number; + emitWebhook?: WebhookEmitter; + }, +): { + service: PredictionsConfirmerService; + repo: jest.Mocked; +} { + const repo = makeRepo(repoOverrides); + const service = new PredictionsConfirmerService(repo, { + batchSize: opts?.batchSize ?? 1000, + maxAttempts: opts?.maxAttempts ?? DEFAULT_MAX_ATTEMPTS, + db: {} as any, + emitWebhook: opts?.emitWebhook ?? jest.fn().mockResolvedValue(undefined), + }); + return { service, repo }; +} + +function confirmedPrediction( + overrides?: Partial, +): ConfirmedPrediction { + return { + id: "pred-1", + marketId: "mkt-1", + userId: "user-1", + outcome: "yes", + amount: "10000000", + txHash: "0xabc123", + status: "confirmed", + ...overrides, + }; +} + +// ─── Tests ──────────────────────────────────────────────────────────────── + +describe("PredictionsConfirmerService", () => { + describe("pollOnce()", () => { + it("processes zero predictions when there are no candidates", async () => { + const { service, repo } = makeService(); + + const result = await service.pollOnce(); + + expect(result).toEqual({ + processed: 0, + confirmed: 0, + incremented: 0, + failed: 0, + webhookErrors: 0, + }); + expect(repo.transactionalTick).toHaveBeenCalledWith(1000, DEFAULT_MAX_ATTEMPTS); + }); + + it("confirms predictions with matching indexer events and emits webhooks", async () => { + const pred = confirmedPrediction(); + const emitWebhook = jest.fn().mockResolvedValue(undefined); + + const { service, repo } = makeService( + { + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [pred], + incremented: 0, + failedIds: [], + }), + }, + { emitWebhook }, + ); + + const result = await service.pollOnce(); + + expect(result).toMatchObject({ + processed: 1, + confirmed: 1, + incremented: 0, + failed: 0, + webhookErrors: 0, + }); + // Webhook should have been emitted once. + expect(emitWebhook).toHaveBeenCalledTimes(1); + expect(emitWebhook).toHaveBeenCalledWith( + {}, + "prediction.confirmed", + expect.objectContaining({ + predictionId: pred.id, + marketId: pred.marketId, + txHash: pred.txHash, + }), + ); + }); + + it("increments attempts for unmatched pending predictions", async () => { + const { service, repo } = makeService({ + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [], + incremented: 5, + failedIds: [], + }), + }); + + const result = await service.pollOnce(); + + expect(result).toMatchObject({ + processed: 5, + confirmed: 0, + incremented: 5, + failed: 0, + }); + }); + + it("marks predictions as failed after max attempts", async () => { + const { service, repo } = makeService({ + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [], + incremented: 0, + failedIds: ["failed-1", "failed-2"], + }), + }); + + const result = await service.pollOnce(); + + expect(result).toMatchObject({ + processed: 2, + confirmed: 0, + incremented: 0, + failed: 2, + }); + }); + + it("handles mixed outcomes in a single tick", async () => { + const pred1 = confirmedPrediction({ id: "pred-confirmed-1" }); + const pred2 = confirmedPrediction({ id: "pred-confirmed-2" }); + + const { service } = makeService({ + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [pred1, pred2], + incremented: 3, + failedIds: ["pred-failed-1"], + }), + }); + + const result = await service.pollOnce(); + + expect(result).toEqual({ + processed: 6, // 2 confirmed + 3 incremented + 1 failed + confirmed: 2, + incremented: 3, + failed: 1, + webhookErrors: 0, + }); + }); + + it("continues when webhook dispatch fails (does not crash tick)", async () => { + const pred = confirmedPrediction(); + const emitWebhook = jest + .fn() + .mockRejectedValue(new Error("Webhook timeout")); + + const { service } = makeService( + { + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [pred], + incremented: 0, + failedIds: [], + }), + }, + { emitWebhook }, + ); + + const result = await service.pollOnce(); + + // The tick completes with the webhook error counted. + expect(result.confirmed).toBe(1); + expect(result.webhookErrors).toBe(1); + expect(emitWebhook).toHaveBeenCalledTimes(1); + }); + + it("uses the configured batch size and max attempts", async () => { + const { service, repo } = makeService( + {}, + { batchSize: 500, maxAttempts: 5 }, + ); + + await service.pollOnce(); + + expect(repo.transactionalTick).toHaveBeenCalledWith(500, 5); + }); + + it("is idempotent when called multiple times with no new events", async () => { + const { service, repo } = makeService(); + + // First call — nothing to process. + const result1 = await service.pollOnce(); + expect(result1.processed).toBe(0); + + // Second call — still nothing. + const result2 = await service.pollOnce(); + expect(result2.processed).toBe(0); + + // transactionalTick was called both times. + expect(repo.transactionalTick).toHaveBeenCalledTimes(2); + }); + }); + + describe("edge cases", () => { + it("handles an empty confirmed list gracefully", async () => { + const emitWebhook = jest.fn(); + const { service } = makeService( + { + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [], + incremented: 0, + failedIds: [], + }), + }, + { emitWebhook }, + ); + + const result = await service.pollOnce(); + + expect(result.confirmed).toBe(0); + expect(result.webhookErrors).toBe(0); + expect(emitWebhook).not.toHaveBeenCalled(); + }); + + it("handles a single prediction correctly", async () => { + const pred = confirmedPrediction(); + const { service } = makeService({ + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [pred], + incremented: 0, + failedIds: [], + }), + }); + + const result = await service.pollOnce(); + + expect(result.confirmed).toBe(1); + expect(result.processed).toBe(1); + expect(result.webhookErrors).toBe(0); + }); + + it("handles many predictions without exceeding batch size", async () => { + const ids = Array.from({ length: 100 }, (_, i) => `pred-${i}`); + const preds = ids.map((id) => confirmedPrediction({ id })); + + const emitWebhook = jest.fn().mockResolvedValue(undefined); + const { service } = makeService( + { + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: preds, + incremented: 0, + failedIds: [], + }), + }, + { emitWebhook }, + ); + + const result = await service.pollOnce(); + + expect(result.confirmed).toBe(100); + expect(emitWebhook).toHaveBeenCalledTimes(100); + }); + + it("does not emit webhooks when no predictions are confirmed", async () => { + const emitWebhook = jest.fn(); + const { service } = makeService( + { + transactionalTick: jest.fn().mockResolvedValue({ + confirmed: [], + incremented: 4, + failedIds: [], + }), + }, + { emitWebhook }, + ); + + const result = await service.pollOnce(); + + expect(result.incremented).toBe(4); + expect(result.confirmed).toBe(0); + expect(emitWebhook).not.toHaveBeenCalled(); + }); + }); +});