Match Result Submission Idempotency for Duplicate Oracle Calls

backend/src/migrations/1788100000000-AddOracleSubmissionIdempotency.ts

@@ -0,0 +1,37 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+/**
+ * Makes oracle match-result submissions idempotent: a duplicate webhook call
+ * (network retry, oracle re-delivery) resolves to the original submission
+ * instead of creating a second record and a second on-chain attempt.
+ */
+export class AddOracleSubmissionIdempotency1788100000000 implements MigrationInterface {
+  async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      ALTER TABLE "oracle_submissions"
+        ADD COLUMN "job_id" varchar(64),
+        ADD COLUMN "idempotency_key" varchar(255)
+    `);
+    await queryRunner.query(`
+      CREATE UNIQUE INDEX "UQ_oracle_submissions_idempotency_key"
+        ON "oracle_submissions" ("idempotency_key")
+        WHERE "idempotency_key" IS NOT NULL
+    `);
+    await queryRunner.query(`
+      CREATE INDEX "IDX_oracle_submissions_job_id"
+        ON "oracle_submissions" ("job_id")
+    `);
+  }
+
+  async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`DROP INDEX "IDX_oracle_submissions_job_id"`);
+    await queryRunner.query(
+      `DROP INDEX "UQ_oracle_submissions_idempotency_key"`,
+    );
+    await queryRunner.query(`
+      ALTER TABLE "oracle_submissions"
+        DROP COLUMN "idempotency_key",
+        DROP COLUMN "job_id"
+    `);
+  }
+}

backend/src/oracle/dto/webhook-match-result.dto.ts

@@ -1,4 +1,12 @@
-import { IsString, IsInt, IsOptional, IsEnum, Min, Max } from 'class-validator';
+import {
+  IsString,
+  IsInt,
+  IsOptional,
+  IsEnum,
+  Min,
+  Max,
+  MaxLength,
+} from 'class-validator';
 import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
 
 export enum WinningTeam {
@@ -54,6 +62,17 @@ export class WebhookMatchResultDto {
   })
   @IsOptional()
   metadata?: Record<string, unknown>;
+
+  @ApiPropertyOptional({
+    description:
+      'Client-supplied key identifying this result delivery. Repeating a call with the same key returns the original job instead of submitting again. When omitted, a key is derived from match_id, data_source, winning_team, confidence_score and timestamp.',
+    example: 'sports-data-123-final',
+    maxLength: 255,
+  })
+  @IsOptional()
+  @IsString()
+  @MaxLength(255)
+  idempotency_key?: string;
 }
 
 export class WebhookResponseDto {
@@ -74,4 +93,16 @@ export class WebhookResponseDto {
     example: 'Match result queued for submission',
   })
   message: string;
+
+  @ApiPropertyOptional({
+    description:
+      'True when this call duplicated an earlier one and no new submission was made',
+    example: false,
+  })
+  duplicate?: boolean;
+
+  @ApiPropertyOptional({
+    description: 'ID of the oracle submission record backing this job',
+  })
+  submission_id?: string;
 }

backend/src/oracle/entities/oracle-submission.entity.ts

@@ -53,6 +53,22 @@ export class OracleSubmission {
   @ApiProperty()
   match_id: string;
 
+  /** Queue job ID returned to the oracle; replayed on duplicate calls. */
+  @Column({ type: 'varchar', length: 64, nullable: true })
+  @Index()
+  @ApiPropertyOptional()
+  job_id?: string;
+
+  /**
+   * Deduplication key for the webhook call that created this submission —
+   * either supplied by the oracle or derived from the result payload. Unique,
+   * so a duplicate call resolves to this record instead of a new one.
+   */
+  @Column({ type: 'varchar', length: 255, nullable: true })
+  @Index({ unique: true, where: '"idempotency_key" IS NOT NULL' })
+  @ApiPropertyOptional()
+  idempotency_key?: string;
+
   @Column({ type: 'varchar', length: 100 })
   @ApiProperty()
   team_a: string;

backend/src/oracle/webhook.service.ts

@@ -4,6 +4,7 @@
   NotFoundException,
   ConflictException,
 } from '@nestjs/common';
+import { createHash } from 'crypto';
 import { InjectRepository } from '@nestjs/typeorm';
 import { Repository } from 'typeorm';
 import { CreatorEventMatch } from '../creator-events/entities/creator-event-match.entity';
@@ -26,6 +27,16 @@
   ReviewResultResponse,
 } from './dto/anomaly-detection.dto';
 
+const POSTGRES_UNIQUE_VIOLATION = '23505';
+
+function isUniqueViolation(error: unknown): boolean {
+  return (
+    !!error &&
+    typeof error === 'object' &&
+    (error as { code?: string }).code === POSTGRES_UNIQUE_VIOLATION
+  );
+}
+
 interface QueuedSubmission {
   id: string;
   matchId: string;
@@ -61,6 +72,17 @@
   async processMatchResult(
     dto: WebhookMatchResultDto,
   ): Promise<WebhookResponseDto> {
+    // A duplicate delivery of a result we already accepted is answered with
+    // the original job, before the resolved-match check, so an oracle retrying
+    // after a successful submission gets the same answer instead of a 409.
+    const idempotencyKey = this.resolveIdempotencyKey(dto);
+    const existing = await this.submissionRepository.findOne({
+      where: { idempotency_key: idempotencyKey },
+    });
+    if (existing) {
+      return this.replayDuplicate(existing, dto);
+    }
+
     const jobId = this.generateJobId();
 
     // Validate match exists
@@ -103,13 +125,30 @@
       nextRetryAt: new Date(),
     };
 
-    this.submissionQueue.set(jobId, submission);
+    // Save to database for history tracking. The unique idempotency key makes
+    // this the serialization point for concurrent duplicate calls: only one
+    // insert wins, the others replay the winner.
+    let savedSubmission: OracleSubmission | null;
+    try {
+      savedSubmission = await this.saveSubmissionToDatabase(
+        submission,
+        match,
+        idempotencyKey,
+      );
+    } catch (error) {
+      if (!isUniqueViolation(error)) {
+        throw error;
+      }
+      const winner = await this.submissionRepository.findOne({
+        where: { idempotency_key: idempotencyKey },
+      });
+      if (!winner) {
+        throw error;
+      }
+      return this.replayDuplicate(winner, dto);
+    }
 
-    // Save to database for history tracking.
-    const savedSubmission = await this.saveSubmissionToDatabase(
-      submission,
-      match,
-    );
+    this.submissionQueue.set(jobId, submission);
 
     // Screen for statistical anomalies before any on-chain use (#1364). A
     // flagged submission is recorded; when holding is enabled it is withheld
@@ -129,6 +168,8 @@
           status: 'held',
           message:
             'Submission flagged as a statistical anomaly and held for manual review',
+          duplicate: false,
+          submission_id: savedSubmission.id,
         };
       }
     }
@@ -144,9 +185,91 @@
       job_id: jobId,
       status: 'accepted',
       message: 'Match result queued for submission',
+      duplicate: false,
+      submission_id: savedSubmission?.id,
+    };
+  }
+
+  /**
+   * Key identifying one result delivery. An oracle-supplied key is scoped to
+   * its data source so two oracles cannot collide; otherwise the key is
+   * derived from the fields that define the result, so a byte-identical
+   * re-delivery maps to the same key. Both are hashed to a fixed length.
+   */
+  private resolveIdempotencyKey(dto: WebhookMatchResultDto): string {
+    const hash = (parts: unknown[]): string =>
+      createHash('sha256').update(JSON.stringify(parts)).digest('hex');
+
+    if (dto.idempotency_key) {
+      return `key:${hash([dto.data_source, dto.idempotency_key])}`;
+    }
+
+    const parsed = new Date(dto.timestamp);
+    const timestamp = Number.isNaN(parsed.getTime())
+      ? dto.timestamp
+      : parsed.toISOString();
+
+    return `payload:${hash([
+      dto.match_id,
+      dto.data_source,
+      dto.winning_team,
+      dto.confidence_score,
+      timestamp,
+    ])}`;
+  }
+
+  /**
+   * Answer a duplicate call with the original submission's state. Nothing is
+   * saved, queued, or submitted. Reusing an oracle-supplied key for a
+   * different result is a client error, not a duplicate.
+   */
+  private replayDuplicate(
+    existing: OracleSubmission,
+    dto: WebhookMatchResultDto,
+  ): WebhookResponseDto {
+    const samePayload =
+      existing.match_id === dto.match_id &&
+      existing.data_source === dto.data_source &&
+      existing.winning_team === (dto.winning_team as string) &&
+      existing.confidence_score === dto.confidence_score;
+
+    if (!samePayload) {
+      throw new ConflictException(
+        `Idempotency key already used for a different result on match ${existing.match_id}`,
+      );
+    }
+
+    this.logger.log(
+      `Duplicate match result ignored: match_id=${dto.match_id}, submission_id=${existing.id}, job_id=${existing.job_id ?? 'unknown'}`,
+    );
+
+    return {
+      job_id: existing.job_id ?? existing.id,
+      status: this.describeSubmissionState(existing),
+      message:
+        'Duplicate result delivery; returning the original submission without resubmitting',
+      duplicate: true,
+      submission_id: existing.id,
     };
   }
 
+  private describeSubmissionState(submission: OracleSubmission): string {
+    if (submission.review_status === SubmissionReviewStatus.HELD) {
+      return 'held';
+    }
+    if (submission.review_status === SubmissionReviewStatus.REJECTED) {
+      return 'rejected';
+    }
+    switch (submission.status) {
+      case SubmissionStatus.SUBMITTED:
+        return 'submitted';
+      case SubmissionStatus.FAILED:
+        return 'failed';
+      default:
+        return 'accepted';
+    }
+  }
+
   private async submitToOracle(submission: QueuedSubmission): Promise<void> {
     try {
       // In a real implementation, this would submit to the Soroban contract
@@ -275,12 +398,20 @@
     this.logger.log('Webhook retry processor started');
   }
 
+  /**
+   * Persist the submission. Storage failures are logged and swallowed so the
+   * result is still attempted, except a unique violation on the idempotency
+   * key, which is rethrown so the caller can replay the original submission.
+   */
   private async saveSubmissionToDatabase(
     submission: QueuedSubmission,
     match: CreatorEventMatch,
+    idempotencyKey: string,
   ): Promise<OracleSubmission | null> {
     try {
       const dbSubmission = this.submissionRepository.create({
+        job_id: submission.id,
+        idempotency_key: idempotencyKey,
         match_id: submission.matchId,
         team_a: match.team_a,
         team_b: match.team_b,
@@ -297,6 +428,9 @@
       this.logger.log(`Submission saved to database: job_id=${submission.id}`);
       return saved;
     } catch (error) {
+      if (isUniqueViolation(error)) {
+        throw error;
+      }
       this.logger.error(
         `Failed to save submission to database: ${error instanceof Error ? error.message : 'Unknown'}`,
       );
@@ -354,16 +488,20 @@
       };
     }
 
+    // Requeue the approved submission for on-chain use. The record is
+    // re-pointed at the new job so status updates and duplicate replays
+    // track it.
+    const requeuedJobId = this.generateJobId();
     submission.review_status = SubmissionReviewStatus.APPROVED;
+    submission.job_id = requeuedJobId;
     await this.submissionRepository.save(submission);
 
     this.logger.log(
       `Held submission approved: submission_id=${submissionId}, reviewer=${dto.reviewer ?? 'unknown'}`,
     );
 
-    // Requeue the approved submission for on-chain use.
     const requeued: QueuedSubmission = {
-      id: this.generateJobId(),
+      id: requeuedJobId,
       matchId: submission.match_id,
       winningTeam: submission.winning_team,
       confidenceScore: submission.confidence_score,
@@ -391,7 +529,7 @@
   ): Promise<void> {
     try {
       const submission = await this.submissionRepository.findOne({
-        where: { match_id: jobId.replace('job_', '').split('_')[0] },
+        where: { job_id: jobId },
       });
 
       if (submission) {