Merge PR #755

Author: greatest0fallt1me

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;

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": {}
+  }
+}

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
     }
   ]
 }

src/config/env-schema.ts

@@ -101,16 +101,10 @@ 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),
 
-  // ── Per-user concurrency limiting ─────────────────────────
-  /**
-   * Maximum number of in-flight (concurrent) requests allowed for a single
-   * authenticated user (or IP for anonymous callers) at any point in time.
-   * Requests that exceed this cap receive HTTP 429 with a `Retry-After: 1`
-   * header.  Set to a large value (e.g. 1000) to effectively disable the
-   * limit without removing the middleware.
-   * Default: 10.
-   */
-  MAX_CONCURRENT_REQUESTS_PER_USER: z.coerce.number().int().positive().default(10),
+  // ── 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. */

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"),
 });
 
 /**

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<typeof setInterval> | null = null;
+let predictionsConfirmerHandle: ReturnType<typeof setInterval> | null = null;
 
 connectWithRetry()
   .then(() => {
@@ -24,6 +26,7 @@ connectWithRetry()
     backupVerificationWorker.start();
     reconciliationWorker.start();
     startSlowQueryAlerter();
+    predictionsConfirmerHandle = startPredictionsConfirmer();
     probeHandle = startIndexerHealthProbe();
 
     const server = app.listen(env.PORT, () => {

src/services/webhookCatalog.ts

@@ -113,6 +113,35 @@ export const DisputeOpenedPayloadSchema = WebhookEnvelopeSchema.extend({
 
 export type DisputeOpenedPayload = z.infer<typeof DisputeOpenedPayloadSchema>;
 
+// ---------------------------------------------------------------------------
+// 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<typeof PredictionConfirmedPayloadSchema>;
+
 // ---------------------------------------------------------------------------
 // Catalog registry
 // ---------------------------------------------------------------------------
@@ -130,6 +159,7 @@ export type DisputeOpenedPayload = z.infer<typeof DisputeOpenedPayloadSchema>;
 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