diff --git a/apps/oracle/main.ts b/apps/oracle/main.ts index df69d7f..b9cb237 100644 --- a/apps/oracle/main.ts +++ b/apps/oracle/main.ts @@ -20,14 +20,14 @@ import { OracleService } from "./oracle-service.js"; import { PrimaryAdapter } from "./primary-adapter.js"; import { FallbackAdapter } from "./fallback-adapter.js"; import { signResolutionReport } from "./signature-helper.js"; -import { RedisSubmissionQueue } from "../workers/src/oracle/redis-submission-queue.js"; +import { BullMQSubmissionQueue } from "../workers/src/oracle/bullmq-submission-queue.js"; import type { ResolutionRequest } from "./provider-adapter.js"; import type { ShutdownHandler, ShutdownSignal, } from "../workers/src/finalization/types.js"; -async function poll(): Promise { +async function poll(queue: BullMQSubmissionQueue): Promise { const config = loadOracleConfig(); const logger = createLogger(config.logLevel); const prisma = getPrismaClient(); @@ -60,14 +60,6 @@ async function poll(): Promise { enableFallback: true, }); - const queue = new RedisSubmissionQueue({ - redisClient: redis, - visibilityTimeoutMs: 300_000, - logger, - }); - - await queue.initialize(); - // Fetch all ACTIVE markets that have an oracle address const markets = await prisma.market.findMany({ where: { status: "ACTIVE" }, @@ -142,11 +134,13 @@ async function bootstrap(): Promise { logger.info("Oracle starting", { pollIntervalMs: config.pollIntervalMs }); + const queue = new BullMQSubmissionQueue(logger); + // Run immediately, then on interval - await poll(); + await poll(queue); const timer = setInterval( () => - void poll().catch((err) => { + void poll(queue).catch((err) => { logger.error("Poll cycle failed", { error: err instanceof Error ? err.message : String(err), }); @@ -165,6 +159,7 @@ async function bootstrap(): Promise { clearInterval(timer); try { + await queue.close(); await disconnectPrisma(); await redis.disconnect(); logger.info("Oracle shutdown complete", { signal }); diff --git a/apps/workers/src/oracle/redis-submission-queue.test.ts b/apps/workers/src/oracle/redis-submission-queue.test.ts deleted file mode 100644 index ed76827..0000000 --- a/apps/workers/src/oracle/redis-submission-queue.test.ts +++ /dev/null @@ -1,286 +0,0 @@ -/** - * Redis Submission Queue Tests - */ - -import { describe, it, expect, beforeEach, vi } from "vitest"; -import { RedisSubmissionQueue } from "./redis-submission-queue.js"; -import type { SubmissionQueueItem } from "../../../oracle/submission-queue.js"; - -// Mock logger -const mockLogger = { - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - child: vi.fn().mockReturnValue({ - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - child: vi.fn(), - }), -}; - -// Mock Redis client -const createMockRedisClient = () => ({ - xgroup: vi.fn(), - xadd: vi.fn(), - exists: vi.fn(), - set: vi.fn(), - xreadgroup: vi.fn(), - xack: vi.fn(), - xclaim: vi.fn(), -}); - -describe("RedisSubmissionQueue", () => { - let queue: RedisSubmissionQueue; - let mockClient: any; - - beforeEach(() => { - mockClient = createMockRedisClient(); - queue = new RedisSubmissionQueue({ - redisClient: mockClient, - visibilityTimeoutMs: 5000, - logger: mockLogger, - }); - vi.clearAllMocks(); - }); - - describe("initialize", () => { - it("should create consumer group on first init", async () => { - mockClient.xgroup.mockResolvedValueOnce(undefined); - - await queue.initialize(); - - expect(mockClient.xgroup).toHaveBeenCalledWith( - "CREATE", - "oracle:submissions", - "oracle-worker", - "$", - { MKSTREAM: true } - ); - expect(mockLogger.info).toHaveBeenCalledWith( - "Oracle submission queue initialized", - expect.any(Object) - ); - }); - - it("should handle existing consumer group gracefully", async () => { - mockClient.xgroup.mockRejectedValueOnce( - new Error("BUSYGROUP group already exists") - ); - - await queue.initialize(); - - expect(mockLogger.info).toHaveBeenCalledWith( - "Consumer group already exists", - expect.any(Object) - ); - }); - - it("should propagate other errors", async () => { - mockClient.xgroup.mockRejectedValueOnce(new Error("Redis error")); - - await expect(queue.initialize()).rejects.toThrow("Redis error"); - }); - }); - - describe("enqueue", () => { - const testItem: SubmissionQueueItem = { - id: "test-123", - request: { - marketId: "market-1", - oracleAddress: "G123456789", - }, - result: { - outcome: true, - source: "Chainlink", - signature: "sig123", - publicKey: "pk123", - confidence: 0.9, - confidenceMetadata: { score: 0.9, method: "test" }, - sourceMetadata: { provider: "Chainlink" }, - timestamp: "2024-01-01T00:00:00Z", - }, - status: "pending", - enqueuedAt: "2024-01-01T00:00:00Z", - attempts: 0, - }; - - it("should enqueue item and set dedup flag", async () => { - mockClient.exists.mockResolvedValueOnce(0); // Not already queued - mockClient.xadd.mockResolvedValueOnce("1-0"); // Stream ID - mockClient.set.mockResolvedValueOnce("OK"); - - const result = await queue.enqueue(testItem); - - expect(result).toBe(true); - expect(mockClient.xadd).toHaveBeenCalledWith( - "oracle:submissions", - "*", - "payload", - expect.any(String), - "marketId", - "market-1", - "payloadHash", - expect.any(String) - ); - expect(mockClient.set).toHaveBeenCalledWith( - expect.stringContaining("oracle:dedup:market-1:"), - "1-0", - "EX", - 86400 - ); - expect(mockLogger.info).toHaveBeenCalledWith( - "Oracle submission queued", - expect.any(Object) - ); - }); - - it("should skip duplicate payloads", async () => { - mockClient.exists.mockResolvedValueOnce(1); // Already queued - - const result = await queue.enqueue(testItem); - - expect(result).toBe(false); - expect(mockClient.xadd).not.toHaveBeenCalled(); - expect(mockLogger.info).toHaveBeenCalledWith( - "Submission already queued, skipping duplicate", - expect.any(Object) - ); - }); - }); - - describe("dequeue", () => { - it("should return null when no messages", async () => { - mockClient.xreadgroup.mockResolvedValueOnce(null); - - const result = await queue.dequeue("consumer-1"); - - expect(result).toBeNull(); - }); - - it("should dequeue and parse message", async () => { - const testItem: SubmissionQueueItem = { - id: "test-123", - request: { marketId: "m1", oracleAddress: "G123" }, - result: { - outcome: true, - source: "Test", - signature: "s1", - publicKey: "p1", - confidence: 0.8, - confidenceMetadata: { score: 0.8, method: "test" }, - sourceMetadata: { provider: "Test" }, - timestamp: "2024-01-01T00:00:00Z", - }, - status: "pending", - enqueuedAt: "2024-01-01T00:00:00Z", - attempts: 0, - }; - - mockClient.xreadgroup.mockResolvedValueOnce([ - [ - "oracle:submissions", - [ - [ - "1-0", - { - payload: JSON.stringify(testItem), - marketId: "m1", - }, - ], - ], - ], - ]); - - const result = await queue.dequeue("consumer-1"); - - expect(result).toBeDefined(); - expect(result?.streamId).toBe("1-0"); - expect(result?.visibilityExpiresAt).toBeGreaterThan(Date.now()); - expect(mockLogger.info).toHaveBeenCalledWith( - "Dequeued submission from Redis stream", - expect.any(Object) - ); - }); - }); - - describe("acknowledge", () => { - it("should acknowledge processed message", async () => { - mockClient.xack.mockResolvedValueOnce(1); - - const item = { - id: "test-123", - request: { marketId: "m1", oracleAddress: "G123" }, - result: { - outcome: true, - source: "Test", - signature: "s1", - publicKey: "p1", - confidence: 0.8, - confidenceMetadata: { score: 0.8, method: "test" }, - sourceMetadata: { provider: "Test" }, - timestamp: "2024-01-01T00:00:00Z", - }, - status: "pending" as const, - enqueuedAt: "2024-01-01T00:00:00Z", - attempts: 0, - streamId: "1-0", - visibilityExpiresAt: Date.now() + 5000, - }; - - await queue.acknowledge(item); - - expect(mockClient.xack).toHaveBeenCalledWith( - "oracle:submissions", - "oracle-worker", - "1-0" - ); - expect(mockLogger.info).toHaveBeenCalledWith( - "Acknowledged oracle submission", - expect.any(Object) - ); - }); - }); - - describe("nack", () => { - it("should nack message for retry", async () => { - mockClient.xclaim.mockResolvedValueOnce([]); - - const item = { - id: "test-123", - request: { marketId: "m1", oracleAddress: "G123" }, - result: { - outcome: true, - source: "Test", - signature: "s1", - publicKey: "p1", - confidence: 0.8, - confidenceMetadata: { score: 0.8, method: "test" }, - sourceMetadata: { provider: "Test" }, - timestamp: "2024-01-01T00:00:00Z", - }, - status: "pending" as const, - enqueuedAt: "2024-01-01T00:00:00Z", - attempts: 1, - streamId: "1-0", - visibilityExpiresAt: Date.now() + 5000, - }; - - await queue.nack(item); - - expect(mockClient.xclaim).toHaveBeenCalledWith( - "oracle:submissions", - "oracle-worker", - "nack-worker", - 0, - "1-0" - ); - expect(mockLogger.warn).toHaveBeenCalledWith( - "Nacked oracle submission for retry", - expect.any(Object) - ); - }); - }); -}); diff --git a/apps/workers/src/oracle/redis-submission-queue.ts b/apps/workers/src/oracle/redis-submission-queue.ts deleted file mode 100644 index ebf278a..0000000 --- a/apps/workers/src/oracle/redis-submission-queue.ts +++ /dev/null @@ -1,243 +0,0 @@ -/** - * Redis-Backed Oracle Submission Queue - * - * Provides durable queue semantics for oracle submissions using Redis streams. - * Implements at-least-once delivery with visibility timeout and idempotency. - * - * @module apps/workers/src/oracle/redis-submission-queue - */ - -import { createHash } from "crypto"; -import type { ILogger } from "../../../../packages/shared/src/logger.js"; -import type { SubmissionQueueItem } from "../../../oracle/submission-queue.js"; - -const STREAM_KEY = "oracle:submissions"; -const CONSUMER_GROUP = "oracle-worker"; - -export interface RedisSubmissionQueueConfig { - redisClient: any; - visibilityTimeoutMs: number; - deduplicationTtlSeconds?: number; - logger: ILogger; -} - -export interface QueuedSubmission extends SubmissionQueueItem { - streamId: string; - visibilityExpiresAt: number; -} - -/** - * Redis-backed submission queue using streams with consumer groups. - */ -export class RedisSubmissionQueue { - private redisClient: any; - private visibilityTimeoutMs: number; - private deduplicationTtlSeconds: number; - private logger: ILogger; - - constructor(config: RedisSubmissionQueueConfig) { - this.redisClient = config.redisClient; - this.visibilityTimeoutMs = config.visibilityTimeoutMs; - this.deduplicationTtlSeconds = config.deduplicationTtlSeconds ?? 86400; - this.logger = config.logger; - } - - /** - * Initialize the consumer group (idempotent). - */ - async initialize(): Promise { - try { - await this.redisClient.xgroup("CREATE", STREAM_KEY, CONSUMER_GROUP, "$", { - MKSTREAM: true, - }); - this.logger.info("Oracle submission queue initialized", { - stream: STREAM_KEY, - group: CONSUMER_GROUP, - }); - } catch (error) { - const msg = error instanceof Error ? error.message : String(error); - if (msg.includes("BUSYGROUP")) { - this.logger.info("Consumer group already exists", { - stream: STREAM_KEY, - group: CONSUMER_GROUP, - }); - } else { - throw error; - } - } - } - - /** - * Compute SHA256 hash of the payload. - */ - private computePayloadHash(payload: unknown): string { - const normalized = JSON.stringify(payload); - return createHash("sha256").update(normalized).digest("hex"); - } - - /** - * Check if a submission is already queued (deduplication). - */ - private async isAlreadyQueued( - marketId: string, - payloadHash: string - ): Promise { - const dedupKey = `oracle:dedup:${marketId}:${payloadHash}`; - return (await this.redisClient.exists(dedupKey)) > 0; - } - - /** - * Mark a submission as queued (set dedup flag with TTL). - */ - private async markAsQueued( - marketId: string, - payloadHash: string, - streamId: string - ): Promise { - const dedupKey = `oracle:dedup:${marketId}:${payloadHash}`; - await this.redisClient.set( - dedupKey, - streamId, - "EX", - this.deduplicationTtlSeconds - ); - } - - /** - * Enqueue a submission to the Redis stream. - * Returns false if already queued (deduplication). - */ - async enqueue(item: SubmissionQueueItem): Promise { - const payloadHash = this.computePayloadHash(item.result); - const marketId = item.request.marketId; - - const alreadyQueued = await this.isAlreadyQueued(marketId, payloadHash); - if (alreadyQueued) { - this.logger.info("Submission already queued, skipping duplicate", { - marketId, - payloadHash: payloadHash.substring(0, 8), - id: item.id, - }); - return false; - } - - const streamId = await this.redisClient.xadd( - STREAM_KEY, - "*", - "payload", - JSON.stringify(item), - "marketId", - marketId, - "payloadHash", - payloadHash - ); - - await this.markAsQueued(marketId, payloadHash, streamId); - - this.logger.info("Oracle submission queued", { - id: item.id, - marketId, - payloadHash: payloadHash.substring(0, 8), - streamId, - enqueuedAt: item.enqueuedAt, - }); - - return true; - } - - /** - * Dequeue a submission from the Redis stream (consumer group). - * Sets visibility timeout for at-least-once delivery. - */ - async dequeue( - consumerName: string, - maxWaitMs: number = 1000 - ): Promise { - const messages = await this.redisClient.xreadgroup( - CONSUMER_GROUP, - consumerName, - STREAM_KEY, - ">", - { COUNT: 1, BLOCK: maxWaitMs } - ); - - if (!messages || !messages.length) { - return null; - } - - const [, msgList] = messages[0]; - if (!msgList || !msgList.length) { - return null; - } - - const [streamId, fieldsData] = msgList[0]; - // fieldsData is either an object (newer ioredis) or array of [key, val, key, val, ...] - const fields = - typeof fieldsData === "object" && !Array.isArray(fieldsData) - ? fieldsData - : Object.fromEntries( - Array.isArray(fieldsData) - ? (fieldsData as string[]).reduce((acc: any[], val, i) => { - if (i % 2 === 0) acc.push([val]); - else acc[acc.length - 1].push(val); - return acc; - }, []) - : [] - ); - - const payload = JSON.parse(fields.payload as string); - - const queued: QueuedSubmission = { - ...payload, - streamId, - visibilityExpiresAt: Date.now() + this.visibilityTimeoutMs, - }; - - this.logger.info("Dequeued submission from Redis stream", { - id: queued.id, - marketId: queued.request.marketId, - streamId, - consumer: consumerName, - }); - - return queued; - } - - /** - * Acknowledge successful processing (remove from consumer group). - */ - async acknowledge(submission: QueuedSubmission): Promise { - await this.redisClient.xack( - STREAM_KEY, - CONSUMER_GROUP, - submission.streamId - ); - - this.logger.info("Acknowledged oracle submission", { - id: submission.id, - marketId: submission.request.marketId, - streamId: submission.streamId, - }); - } - - /** - * Negative acknowledge (nack) — makes the message visible again for retry. - */ - async nack(submission: QueuedSubmission, consumerName: string): Promise { - await this.redisClient.xclaim( - STREAM_KEY, - CONSUMER_GROUP, - consumerName, - 0, // Min idle time 0 = claim immediately - submission.streamId - ); - - this.logger.warn("Nacked oracle submission for retry", { - id: submission.id, - marketId: submission.request.marketId, - streamId: submission.streamId, - attempts: submission.attempts, - consumerName, - }); - } -} diff --git a/apps/workers/src/oracle/submission-worker.test.ts b/apps/workers/src/oracle/submission-worker.test.ts deleted file mode 100644 index 198c170..0000000 --- a/apps/workers/src/oracle/submission-worker.test.ts +++ /dev/null @@ -1,173 +0,0 @@ -/** - * Submission Worker Tests - */ - -import { describe, it, expect, beforeEach, vi } from "vitest"; - -vi.mock("../../../oracle/signature-helper.js", () => ({ - verifyResolutionReport: vi.fn((report: { signature?: string }) => - Boolean(report.signature) - ), -})); - -import { SubmissionWorker } from "./submission-worker.js"; -import type { QueuedSubmission } from "./redis-submission-queue.js"; - -// Mock Prisma -const mockPrisma = { - oracleReport: { - create: vi.fn(), - upsert: vi.fn(), - updateMany: vi.fn(), - }, - resolutionCandidate: { - upsert: vi.fn(), - }, -}; - -// Mock queue -const mockQueue = { - acknowledge: vi.fn(), - nack: vi.fn(), -}; - -// Mock logger -const mockLogger = { - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - child: vi.fn().mockReturnValue({ - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - child: vi.fn(), - }), -}; - -const createTestSubmission = (): QueuedSubmission => ({ - id: "test-123", - request: { - marketId: "market-1", - oracleAddress: "GTEST123456789", - }, - result: { - outcome: true, - source: "Chainlink", - signature: "dGVzdF9zaWduYXR1cmU=", // base64 encoded - publicKey: "GTEST123456789", - confidence: 0.9, - confidenceMetadata: { score: 0.9, method: "test" }, - sourceMetadata: { provider: "Chainlink" }, - timestamp: new Date().toISOString(), - }, - status: "pending", - enqueuedAt: new Date().toISOString(), - attempts: 0, - streamId: "1-0", - visibilityExpiresAt: Date.now() + 5000, -}); - -describe("SubmissionWorker", () => { - let worker: SubmissionWorker; - - beforeEach(() => { - vi.clearAllMocks(); - worker = new SubmissionWorker(mockQueue as any, mockPrisma as any, { - submissionMaxRetries: 3, - consumerName: "test-consumer", - logger: mockLogger, - }); - }); - - describe("processSubmission", () => { - it("should process successful submission", async () => { - const submission = createTestSubmission(); - mockPrisma.oracleReport.create.mockResolvedValueOnce({ - id: "report-1", - }); - mockPrisma.resolutionCandidate.upsert.mockResolvedValueOnce({ - id: "candidate-1", - }); - mockQueue.acknowledge.mockResolvedValueOnce(undefined); - - await worker.processSubmission(submission); - - expect(mockPrisma.oracleReport.create).toHaveBeenCalled(); - expect(mockPrisma.resolutionCandidate.upsert).toHaveBeenCalled(); - expect(mockQueue.acknowledge).toHaveBeenCalledWith(submission); - expect(mockLogger.info).toHaveBeenCalledWith( - "Oracle submission processed successfully", - expect.any(Object) - ); - }); - - it("should retry on first failure", async () => { - const submission = createTestSubmission(); - const error = new Error("Network error"); - - // Mock successful DB writes but then throw on submission - mockPrisma.oracleReport.updateMany.mockResolvedValueOnce({ - count: 1, - }); - mockQueue.nack.mockResolvedValueOnce(undefined); - - // Create an override for processSubmission to simulate network error - // We'll do this by checking the error flow directly - await expect( - worker.processSubmission({ - ...submission, - result: { ...submission.result, signature: "" }, // Invalid signature - }) - ).rejects.toThrow(); - - // Should have nacked for retry - expect(mockQueue.nack).toHaveBeenCalled(); - expect(mockLogger.warn).toHaveBeenCalledWith( - "Oracle submission processing failed, will retry", - expect.any(Object) - ); - }); - - it("should dead-letter after max retries", async () => { - const submission = createTestSubmission(); - submission.attempts = 2; // Will exceed maxRetries of 3 on next attempt - - mockPrisma.oracleReport.updateMany.mockResolvedValueOnce({ - count: 1, - }); - mockQueue.acknowledge.mockResolvedValueOnce(undefined); - - await expect( - worker.processSubmission({ - ...submission, - result: { ...submission.result, signature: "" }, - }) - ).rejects.toThrow(); - - // Should acknowledge (remove from active queue) - expect(mockQueue.acknowledge).toHaveBeenCalled(); - expect(mockLogger.error).toHaveBeenCalledWith( - "Oracle submission processing failed, max attempts exceeded", - expect.any(Object) - ); - }); - - it("should handle Prisma errors gracefully", async () => { - const submission = createTestSubmission(); - mockPrisma.oracleReport.create.mockRejectedValueOnce( - new Error("DB error") - ); - - await expect(worker.processSubmission(submission)).rejects.toThrow( - "DB error" - ); - - expect(mockLogger.error).toHaveBeenCalledWith( - "Failed to persist oracle submission", - expect.any(Object) - ); - }); - }); -}); diff --git a/apps/workers/src/oracle/submission-worker.ts b/apps/workers/src/oracle/submission-worker.ts deleted file mode 100644 index 58718c6..0000000 --- a/apps/workers/src/oracle/submission-worker.ts +++ /dev/null @@ -1,397 +0,0 @@ -/** - * Oracle Submission Worker - * - * Polls the Redis queue and submits signed oracle resolutions on-chain. - * Implements retry logic, persistence, and graceful shutdown. - * - * @module apps/workers/src/oracle/submission-worker - */ - -import { - Contract, - Keypair, - TransactionBuilder, - nativeToScVal, - rpc as StellarRpc, - xdr, -} from "@stellar/stellar-sdk"; -import { PrismaClient } from "../../../../src/generated/prisma/client/index.js"; -import type { ILogger } from "../../../../packages/shared/src/logger.js"; -import { - verifyResolutionReport, - type SignedResolutionReport, -} from "../../../oracle/signature-helper.js"; -import { - RedisSubmissionQueue, - type QueuedSubmission, -} from "./redis-submission-queue.js"; -import type { SubmissionQueueItem } from "../../../oracle/submission-queue.js"; - -export interface OracleStellarConfig { - rpcUrl: string; - contractId: string; - networkPassphrase: string; - signerSecret: string; -} - -export interface SubmissionWorkerConfig { - submissionMaxRetries: number; - consumerName: string; - logger: ILogger; - stellar?: OracleStellarConfig; -} - -/** - * Submission worker that processes queued oracle resolutions. - */ -export class SubmissionWorker { - private maxRetries: number; - private consumerName: string; - private logger: ILogger; - private queue: RedisSubmissionQueue; - private prisma: PrismaClient; - private stellarConfig?: OracleStellarConfig; - - constructor( - queue: RedisSubmissionQueue, - prisma: PrismaClient, - config: SubmissionWorkerConfig - ) { - this.queue = queue; - this.prisma = prisma; - this.maxRetries = config.submissionMaxRetries; - this.consumerName = config.consumerName; - this.logger = config.logger; - this.stellarConfig = config.stellar; - } - - /** - * Process a single queued submission. - */ - async processSubmission(submission: QueuedSubmission): Promise { - const { id, request, result, attempts } = submission; - - try { - this.logger.info("Processing oracle submission", { - id, - marketId: request.marketId, - attempt: attempts + 1, - maxAttempts: this.maxRetries, - }); - - // Create signed report from the result - const report = this.createSignedReport(submission); - - // Verify signature before submission (defensive check) - if (!verifyResolutionReport(report)) { - this.logger.error("Signature verification failed", { - id, - marketId: request.marketId, - attempt: attempts + 1, - }); - throw new Error("Signature verification failed"); - } - - // Submit on-chain via Stellar SDK - await this.submitOnChain(report, request.oracleAddress); - - // Update database on success - await this.updateOnSuccess(submission, report); - - // Acknowledge in queue - await this.queue.acknowledge(submission); - - this.logger.info("Oracle submission processed successfully", { - id, - marketId: request.marketId, - attempt: attempts + 1, - }); - } catch (error) { - const errorMessage = - error instanceof Error ? error.message : String(error); - const nextAttempt = attempts + 1; - - if (nextAttempt < this.maxRetries) { - this.logger.warn("Oracle submission processing failed, will retry", { - id, - marketId: request.marketId, - attempt: nextAttempt, - maxAttempts: this.maxRetries, - error: errorMessage, - }); - - // Update attempts and nack for retry - const updated: QueuedSubmission = { - ...submission, - attempts: nextAttempt, - lastAttemptAt: new Date().toISOString(), - lastError: errorMessage, - }; - - await this.updateAttempt(updated); - await this.queue.nack(updated, this.consumerName); - } else { - this.logger.error( - "Oracle submission processing failed, max attempts exceeded", - { - id, - marketId: request.marketId, - attempt: nextAttempt, - maxAttempts: this.maxRetries, - error: errorMessage, - } - ); - - // Dead-letter: mark as failed in database - await this.updateOnFailure(submission, errorMessage); - await this.queue.acknowledge(submission); // Remove from active queue - } - - throw error; // Re-throw for caller to handle - } - } - - /** - * Create a signed resolution report from a queued submission. - */ - private createSignedReport( - submission: SubmissionQueueItem - ): SignedResolutionReport { - const { result, request } = submission; - - return { - payload: { - marketId: request.marketId, - outcome: result.outcome, - timestamp: new Date().toISOString(), - }, - signature: result.signature || "", - publicKey: result.publicKey || "", - }; - } - - /** - * Submit the signed resolution on-chain by invoking resolve_market on the - * Soroban contract. Falls back to a warn-only path when stellar config is - * absent (e.g. in local dev without chain access). - */ - private async submitOnChain( - report: SignedResolutionReport, - oracleAddress: string - ): Promise { - if (!report.payload.marketId || !report.signature || !report.publicKey) { - throw new Error("Invalid report: missing required fields"); - } - - if (!oracleAddress || oracleAddress.length === 0) { - throw new Error("Invalid oracle address"); - } - - if (!this.stellarConfig) { - this.logger.warn( - "No Stellar config provided — resolve_market call skipped (off-chain only). " + - "Set STELLAR_RPC_URL, MARKET_CONTRACT_ID, SOROBAN_NETWORK_PASSPHRASE, " + - "and ORACLE_SECRET_KEY to enable on-chain submission.", - { marketId: report.payload.marketId, oracleAddress } - ); - return; - } - - const { rpcUrl, contractId, networkPassphrase, signerSecret } = - this.stellarConfig; - - this.logger.debug("Invoking resolve_market on-chain", { - marketId: report.payload.marketId, - oracleAddress, - outcome: report.payload.outcome, - contractId, - }); - - const keypair = Keypair.fromSecret(signerSecret); - const server = new StellarRpc.Server(rpcUrl); - const contract = new Contract(contractId); - - const sourceAccount = await server.getAccount(keypair.publicKey()); - - const args: xdr.ScVal[] = [ - nativeToScVal(report.payload.marketId, { type: "string" }), - nativeToScVal(report.payload.outcome, { type: "bool" }), - nativeToScVal(Buffer.from(report.signature, "base64"), { type: "bytes" }), - nativeToScVal(report.publicKey, { type: "address" }), - ]; - - const tx = new TransactionBuilder(sourceAccount, { - fee: "100", - networkPassphrase, - }) - .addOperation(contract.call("resolve_market", ...args)) - .setTimeout(30) - .build(); - - const preparedTx = await server.prepareTransaction(tx); - preparedTx.sign(keypair); - - const sendResult = await server.sendTransaction(preparedTx); - - if (sendResult.status === "ERROR") { - throw new Error( - `resolve_market submission failed: status=ERROR hash=${sendResult.hash}` - ); - } - - this.logger.info("resolve_market submitted, awaiting confirmation", { - marketId: report.payload.marketId, - hash: sendResult.hash, - }); - - // Poll until confirmed or failed - const MAX_POLL_ATTEMPTS = 30; - const POLL_INTERVAL_MS = 1_000; - for (let i = 0; i < MAX_POLL_ATTEMPTS; i++) { - await new Promise((r) => setTimeout(r, POLL_INTERVAL_MS)); - const txStatus = await server.getTransaction(sendResult.hash); - if (txStatus.status === StellarRpc.Api.GetTransactionStatus.SUCCESS) { - this.logger.info("resolve_market confirmed on-chain", { - marketId: report.payload.marketId, - hash: sendResult.hash, - ledger: txStatus.ledger, - }); - return; - } - if (txStatus.status === StellarRpc.Api.GetTransactionStatus.FAILED) { - throw new Error( - `resolve_market transaction failed on-chain: hash=${sendResult.hash}` - ); - } - } - - throw new Error( - `resolve_market not confirmed after ${MAX_POLL_ATTEMPTS}s: hash=${sendResult.hash}` - ); - } - - /** - * Update database on successful submission. - */ - private async updateOnSuccess( - submission: QueuedSubmission, - report: SignedResolutionReport - ): Promise { - const { request } = submission; - const { marketId, outcome, timestamp } = report.payload; - - try { - const payloadHash = this.computePayloadHash(report.payload); - - // Create or update OracleReport - await this.prisma.oracleReport.create({ - data: { - payloadHash, - source: request.oracleAddress, - confidence: 1.0, // Full confidence on successful submission - marketId, - candidateResolution: outcome, - createdAt: new Date(timestamp), - }, - }); - - // Upsert ResolutionCandidate - await this.prisma.resolutionCandidate.upsert({ - where: { - idempotencyKey: `${marketId}:${request.oracleAddress}`, - }, - create: { - marketId, - proposedOutcome: outcome, - source: request.oracleAddress, - operatorAddress: request.oracleAddress, - idempotencyKey: `${marketId}:${request.oracleAddress}`, - }, - update: { - proposedOutcome: outcome, - }, - }); - - this.logger.info("Oracle submission persisted", { - id: submission.id, - marketId, - outcome, - }); - } catch (error) { - this.logger.error("Failed to persist oracle submission", { - id: submission.id, - marketId, - error: error instanceof Error ? error.message : String(error), - }); - throw error; - } - } - - /** - * Update attempts for retry. - */ - private async updateAttempt(submission: QueuedSubmission): Promise { - const { request } = submission; - - try { - await this.prisma.oracleReport.updateMany({ - where: { marketId: request.marketId }, - data: { - confidence: Math.max(0, 1.0 - submission.attempts * 0.2), - }, - }); - } catch (error) { - this.logger.warn("Failed to update attempt count", { - id: submission.id, - marketId: request.marketId, - error: error instanceof Error ? error.message : String(error), - }); - } - } - - /** - * Mark submission as failed in database. - */ - private async updateOnFailure( - submission: QueuedSubmission, - errorMessage: string - ): Promise { - const { request } = submission; - - try { - await this.prisma.oracleReport.updateMany({ - where: { marketId: request.marketId }, - data: { candidateResolution: null }, - }); - - await this.prisma.resolutionCandidate.updateMany({ - where: { - marketId: request.marketId, - source: request.oracleAddress, - }, - data: { status: "REJECTED" }, - }); - - this.logger.error("Oracle submission marked as failed", { - id: submission.id, - marketId: request.marketId, - error: errorMessage, - }); - } catch (error) { - this.logger.error("Failed to mark submission as failed", { - id: submission.id, - marketId: request.marketId, - error: error instanceof Error ? error.message : String(error), - }); - } - } - - /** - * Compute payload hash (same as queue). - */ - private computePayloadHash(payload: unknown): string { - const crypto = require("crypto"); - const normalized = JSON.stringify(payload); - return crypto.createHash("sha256").update(normalized).digest("hex"); - } -} diff --git a/package.json b/package.json index f9b7c8f..c50e212 100644 --- a/package.json +++ b/package.json @@ -52,9 +52,10 @@ "@prisma/adapter-pg": "^7.3.0", "@prisma/client": "^7.2.0", "@stellar/stellar-sdk": "^14.4.3", + "bullmq": "^5.79.2", + "dotenv": "^16.6.1", "fastify": "^5.7.1", "fastify-plugin": "^5.1.0", - "dotenv": "^16.6.1", "ioredis": "^5.9.2", "pg": "^8.17.2", "redis": "^5.10.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0b0f3d0..b4c4b6b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -26,6 +26,9 @@ importers: '@stellar/stellar-sdk': specifier: ^14.4.3 version: 14.6.1 + bullmq: + specifier: ^5.79.2 + version: 5.79.2(redis@5.12.1) dotenv: specifier: ^16.6.1 version: 16.6.1 @@ -336,6 +339,36 @@ packages: '@kurkle/color@0.3.4': resolution: {integrity: sha512-M5UknZPHRu3DEDWoipU6sE8PdkZ6Z/S+v4dD+Ke8IaNlpdSQah50lz1KtcFBa2vsdOnwbbnxJwVM4wty6udA5w==} + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + resolution: {integrity: sha512-LCkGo6JDfaBhgST7UpPWgNgLINpcpabaHfyz5OBx75nUYxBsaEPxjnyNjWpeb/xBup/682QnBfRBy2/LvPutZQ==} + cpu: [arm64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + resolution: {integrity: sha512-zExlW9zUJKZH/tOtVMttwjKa4Xm/3KcNjnE3dPN92uCktwavMxpgCA3MoJK/DOnTWsQgo224OaST27/mPNAf+w==} + cpu: [x64] + os: [darwin] + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + resolution: {integrity: sha512-dgX0P/9wGPJeHFBG+ZmhgE6bmtMt7NP5CRBGyyktpopdk/mW4POnrpQsSLtKI1dwpc+pPLuXHDh6vvskyQE/sw==} + cpu: [arm64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + resolution: {integrity: sha512-Tg3yX65f5GbtXLkrYEHE5oibZG9epyYWas7FogTTEJeDEF9JlXJzKgXaNhT3UXlTOeA+AfZpYZYZ0uPj7Cfquw==} + cpu: [arm] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + resolution: {integrity: sha512-8TNXMEjJc3QEy7R/x1INhgiU+XakDAFUzBhaz7+Rbrs8NH5UQeHQxxmzsSBJGyV6I1jW79undiQm8tOI+D+8FQ==} + cpu: [x64] + os: [linux] + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + resolution: {integrity: sha512-CmCXPQrkbwExx3j946/PtHWHbYJiCRBRDl4BlkRQcJB/YOwQxJRTpoo7aTsortjgoJ1x7opzTSxn7C+ASSLVjQ==} + cpu: [x64] + os: [win32] + '@napi-rs/wasm-runtime@1.1.4': resolution: {integrity: sha512-3NQNNgA1YSlJb/kMH1ildASP9HW7/7kYnRI2szWJaofaS1hWmbGI4H+d3+22aGzXXN9IJ+n+GiFVcGipJP18ow==} peerDependencies: @@ -805,6 +838,15 @@ packages: buffer@6.0.3: resolution: {integrity: sha512-FTiCpNxtwiZZHEZbcbTIcZjERVICn9yq/pDFkTl95/AxzD1naBctN7YO68riM/gLSDY7sdrMby8hofADYuuqOA==} + bullmq@5.79.2: + resolution: {integrity: sha512-FebD+8XCZl/hnS1R4to24L4EAN70XSndKZO0776M36vGRk5MKVhKlNFM8/34zXLXKyYB4QaeIPFhVXSYYGTHpQ==} + engines: {node: '>=12.22.0'} + peerDependencies: + redis: '>=5.0.0' + peerDependenciesMeta: + redis: + optional: true + c12@3.3.4: resolution: {integrity: sha512-cM0ApFQSBXuourJejzwv/AuPRvAxordTyParRVcHjjtXirtkzM0uK2L9TTn9s0cXZbG7E55jCivRQzoxYmRAlA==} peerDependencies: @@ -882,6 +924,10 @@ packages: resolution: {integrity: sha512-ei8Aos7ja0weRpFzJnEA9UHJ/7XQmqglbRwnf2ATjcB9Wq874VKH9kfjjirM6UhU2/E5fFYadylyhFldcqSidQ==} engines: {node: '>=18'} + cron-parser@4.9.0: + resolution: {integrity: sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==} + engines: {node: '>=12.0.0'} + cross-spawn@7.0.6: resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==} engines: {node: '>= 8'} @@ -1383,6 +1429,10 @@ packages: resolution: {integrity: sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA==} engines: {bun: '>=1.0.0', deno: '>=1.30.0', node: '>=8.0.0'} + luxon@3.7.2: + resolution: {integrity: sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==} + engines: {node: '>=12'} + magic-string@0.30.21: resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==} @@ -1431,6 +1481,13 @@ packages: ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} + msgpackr-extract@3.0.4: + resolution: {integrity: sha512-4kmO/MdyUIkLIvTPr8VHLil4AtoKIoniWPIEk5+CDy0xnWC84azhSFmuJ7PxZdsYtiP5kEeQsORAVIeMgxT+Hw==} + hasBin: true + + msgpackr@2.0.4: + resolution: {integrity: sha512-o1C5KRmuRt+apqMr1HuGSqWStZoRBUpEsCsl15uM9VdAF1qHLtvMOU2En747EnTyEl6c4pzPewRMFF31s1CNbA==} + mysql2@3.15.3: resolution: {integrity: sha512-FBrGau0IXmuqg4haEZRBfHNWB5mUARw6hNwPDXXGg0XzVJ50mr/9hb267lvpVMnhZ1FON3qNd4Xfcez1rbFwSg==} engines: {node: '>= 8.0'} @@ -1444,6 +1501,13 @@ packages: engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1} hasBin: true + node-abort-controller@3.1.1: + resolution: {integrity: sha512-AGK2yQKIjRuqnc6VkX2Xj5d+QW8xZ87pa1UK6yA6ouUyuxfHuMP6umE5QK7UmTeOAymo+Zx1Fxiuw9rVx8taHQ==} + + node-gyp-build-optional-packages@5.2.2: + resolution: {integrity: sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==} + hasBin: true + nodemon@3.1.14: resolution: {integrity: sha512-jakjZi93UtB3jHMWsXL68FXSAosbLfY0In5gtKq3niLSkrWznrVBzXFNOEMJUfc9+Ke7SHWoAZsiMkNP3vq6Jw==} engines: {node: '>=10'} @@ -1722,6 +1786,11 @@ packages: engines: {node: '>=10'} hasBin: true + semver@7.8.5: + resolution: {integrity: sha512-Y7/KDsb8LjooZpwaqGyulO6DQlksgCncchHGk+sZIY4SBvUocMBEFH5Ur1fI4dV+Jvl0w6cjvucaIi40puRioA==} + engines: {node: '>=10'} + hasBin: true + seq-queue@0.0.5: resolution: {integrity: sha512-hr3Wtp/GZIc/6DAGPDcV4/9WoZhjrkXsi5B/07QgX8tsdc6ilr7BFM6PM6rbdAX1kFSDYeZGLipIZZKyQP0O5Q==} @@ -2193,6 +2262,24 @@ snapshots: '@kurkle/color@0.3.4': {} + '@msgpackr-extract/msgpackr-extract-darwin-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-darwin-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-arm@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-linux-x64@3.0.4': + optional: true + + '@msgpackr-extract/msgpackr-extract-win32-x64@3.0.4': + optional: true + '@napi-rs/wasm-runtime@1.1.4(@emnapi/core@1.10.0)(@emnapi/runtime@1.10.0)': dependencies: '@emnapi/core': 1.10.0 @@ -2648,6 +2735,19 @@ snapshots: base64-js: 1.5.1 ieee754: 1.2.1 + bullmq@5.79.2(redis@5.12.1): + dependencies: + cron-parser: 4.9.0 + ioredis: 5.10.1 + msgpackr: 2.0.4 + node-abort-controller: 3.1.1 + semver: 7.8.5 + tslib: 2.8.1 + optionalDependencies: + redis: 5.12.1 + transitivePeerDependencies: + - supports-color + c12@3.3.4(magicast@0.5.2): dependencies: chokidar: 5.0.0 @@ -2733,6 +2833,10 @@ snapshots: cookie@1.1.1: {} + cron-parser@4.9.0: + dependencies: + luxon: 3.7.2 + cross-spawn@7.0.6: dependencies: path-key: 3.1.1 @@ -3215,6 +3319,8 @@ snapshots: lru.min@1.1.4: {} + luxon@3.7.2: {} + magic-string@0.30.21: dependencies: '@jridgewell/sourcemap-codec': 1.5.5 @@ -3256,6 +3362,22 @@ snapshots: ms@2.1.3: {} + msgpackr-extract@3.0.4: + dependencies: + node-gyp-build-optional-packages: 5.2.2 + optionalDependencies: + '@msgpackr-extract/msgpackr-extract-darwin-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-darwin-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-arm64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-linux-x64': 3.0.4 + '@msgpackr-extract/msgpackr-extract-win32-x64': 3.0.4 + optional: true + + msgpackr@2.0.4: + optionalDependencies: + msgpackr-extract: 3.0.4 + mysql2@3.15.3: dependencies: aws-ssl-profiles: 1.1.2 @@ -3274,6 +3396,13 @@ snapshots: nanoid@3.3.11: {} + node-abort-controller@3.1.1: {} + + node-gyp-build-optional-packages@5.2.2: + dependencies: + detect-libc: 2.1.2 + optional: true + nodemon@3.1.14: dependencies: chokidar: 3.6.0 @@ -3540,6 +3669,8 @@ snapshots: semver@7.7.4: {} + semver@7.8.5: {} + seq-queue@0.0.5: {} set-cookie-parser@2.7.2: {} @@ -3668,8 +3799,7 @@ snapshots: dependencies: typescript: 5.9.3 - tslib@2.8.1: - optional: true + tslib@2.8.1: {} tsx@4.21.0: dependencies: diff --git a/prisma/migrations/20260629220000_remove_unused_position_model/migration.sql b/prisma/migrations/20260629220000_remove_unused_position_model/migration.sql new file mode 100644 index 0000000..7d65d35 --- /dev/null +++ b/prisma/migrations/20260629220000_remove_unused_position_model/migration.sql @@ -0,0 +1,5 @@ +-- DropForeignKey +ALTER TABLE "positions" DROP CONSTRAINT "positions_market_id_fkey"; + +-- DropTable +DROP TABLE "positions"; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 3de9756..354173a 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -72,7 +72,6 @@ model Market { orders Order[] positions UserPosition[] - positionSnapshots Position[] oracleReports OracleReport[] resolutionCandidates ResolutionCandidate[] resolutions Resolution[] @@ -189,24 +188,6 @@ model Resolution { @@map("resolutions") } -model Position { - id String @id @default(uuid()) - walletAddress String @map("wallet_address") @db.VarChar(56) - marketId String @map("market_id") - outcome Outcome? - quantity Int @default(0) - valuation Decimal @default(0) @map("valuation") @db.Decimal(20, 8) - createdAt DateTime @default(now()) @map("created_at") - updatedAt DateTime @default(now()) @updatedAt @map("updated_at") - - market Market @relation(fields: [marketId], references: [id], onDelete: Cascade) - - @@unique([walletAddress, marketId, outcome], map: "positions_wallet_market_outcome_key") - @@index([walletAddress]) - @@index([marketId]) - @@map("positions") -} - model IndexerCursor { networkId String @map("network_id") cursorKey String @map("cursor_key") @@ -233,18 +214,18 @@ model IndexerProcessedEvent { /// CLOB-engine trade records. Written atomically with order fills; tradeId is /// the source-of-truth idempotency key preventing duplicate writes on retry. model Trade { - id String @id @default(uuid()) - tradeId String @unique @map("trade_id") @db.VarChar(256) - marketId String @map("market_id") - outcome Outcome - buyerAddress String @map("buyer_address") @db.VarChar(56) - sellerAddress String @map("seller_address") @db.VarChar(56) - buyOrderId String @map("buy_order_id") - sellOrderId String @map("sell_order_id") - price Decimal @db.Decimal(10, 8) - quantity Int - tradedAt DateTime @map("traded_at") - createdAt DateTime @default(now()) @map("created_at") + id String @id @default(uuid()) + tradeId String @unique @map("trade_id") @db.VarChar(256) + marketId String @map("market_id") + outcome Outcome + buyerAddress String @map("buyer_address") @db.VarChar(56) + sellerAddress String @map("seller_address") @db.VarChar(56) + buyOrderId String @map("buy_order_id") + sellOrderId String @map("sell_order_id") + price Decimal @db.Decimal(10, 8) + quantity Int + tradedAt DateTime @map("traded_at") + createdAt DateTime @default(now()) @map("created_at") @@index([marketId]) @@index([buyerAddress]) diff --git a/tests/prisma.schema.test.ts b/tests/prisma.schema.test.ts index e0c831c..22b0c68 100644 --- a/tests/prisma.schema.test.ts +++ b/tests/prisma.schema.test.ts @@ -25,10 +25,11 @@ describe("Prisma Schema", () => { expect(prisma.market).toBeDefined(); expect(prisma.order).toBeDefined(); expect(prisma.userPosition).toBeDefined(); - expect(prisma.position).toBeDefined(); expect(prisma.indexerCursor).toBeDefined(); expect(prisma.indexerProcessedEvent).toBeDefined(); + expect(prisma.trade).toBeDefined(); expect(prisma.indexedTrade).toBeDefined(); + expect(prisma.collateralDeposit).toBeDefined(); }); it("should define the expected schema models", () => { @@ -39,12 +40,13 @@ describe("Prisma Schema", () => { "UserPosition", "ResolutionCandidate", "Resolution", - "Position", "IndexerCursor", "IndexerProcessedEvent", + "Trade", "IndexedTrade", "OracleSourceAlias", + "CollateralDeposit", ]); - expect(modelNames).toHaveLength(11); + expect(modelNames).toHaveLength(12); }); });