fix: make event ingestion idempotent by canonical log identity

prisma/migrations/20260924000000_add_chain_event_dedup_and_block_cursor/migration.sql

@@ -1,18 +1,21 @@
 -- Enforce monotonic block cursor compare-and-swap and auditable rollbacks.
 -- CreateTable
-CREATE TABLE v2_chain_events (
-    id TEXT NOT NULL PRIMARY KEY,
-    chainId INTEGER NOT NULL,
-    contractAddress TEXT NOT NULL,
-    eventName TEXT NOT NULL,
-    blockNumber BIGINT NOT NULL,
-    blockHash TEXT NOT NULL,
-    txHash TEXT NOT NULL,
-    logIndex INTEGER NOT NULL,
-    blockTimestamp DATETIME,
-    payload JSONB NOT NULL,
-    rawArgs JSONB NOT NULL,
-    ingestedAt DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+CREATE TABLE "v2_chain_events" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "chainId" INTEGER NOT NULL,
+    "releaseId" TEXT NOT NULL,
+    "decoderVersion" TEXT NOT NULL,
+    "contractAddress" TEXT NOT NULL,
+    "eventName" TEXT NOT NULL,
+    "blockNumber" BIGINT NOT NULL,
+    "blockHash" TEXT NOT NULL,
+    "txHash" TEXT NOT NULL,
+    "logIndex" INTEGER NOT NULL,
+    "blockTimestamp" DATETIME,
+    "payload" JSONB NOT NULL,
+    "rawArgs" JSONB NOT NULL,
+    "removed" BOOLEAN NOT NULL DEFAULT false,
+    "ingestedAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
 );
 
 -- CreateTable
@@ -82,7 +85,13 @@ CREATE INDEX "v2_chain_events_eventName_blockNumber_idx" ON "v2_chain_events"("e
 CREATE INDEX "v2_chain_events_contractAddress_idx" ON "v2_chain_events"("contractAddress");
 
 -- CreateIndex
-CREATE UNIQUE INDEX "v2_chain_events_chainId_blockHash_txHash_logIndex_key" ON "v2_chain_events"("chainId", "blockHash", "txHash", "logIndex");
+CREATE UNIQUE INDEX "v2_chain_events_canonical_identity_key" ON "v2_chain_events"("chainId", "releaseId", "txHash", "logIndex", "decoderVersion");
+
+-- CreateIndex
+CREATE INDEX "v2_chain_events_blockHash_idx" ON "v2_chain_events"("blockHash");
+
+-- CreateIndex
+CREATE INDEX "v2_chain_events_removed_idx" ON "v2_chain_events"("removed");
 
 -- CreateIndex
 CREATE UNIQUE INDEX "v2_block_cursors_chainId_source_releaseId_projectionId_key" ON "v2_block_cursors"("chainId", "source", "releaseId", "projectionId");
@@ -149,4 +158,4 @@ CREATE TRIGGER "v2_block_cursor_rollbacks_audit_guard"
     OR NEW."toHeight" > NEW."fromHeight"
   ) BEGIN
     SELECT RAISE(ABORT, 'block cursor rollback must be authorized, audited, and move backward');
-  END;
+  END;

prisma/schema.prisma

@@ -285,8 +285,8 @@ model BlockCursor {
   processedHash   String @default("")
   safeHeight      BigInt @default(0)
   safeHash        String @default("")
-  finalizedHeight BigInt @default(0)
-  finalizedHash   String @default("")
+  finalizedHeight  BigInt @default(0)
+  finalizedHash    String @default("")
 
   // Monotonic CAS guard (PREWAVE-BE-038): the release/projection this cursor
   // belongs to, plus the deployment floor and checkpoint provenance. A writer
@@ -355,11 +355,26 @@ model ChainEvent {
   // JSON-encoded RawChainLog.rawArgs
   rawArgs         String
 
+  // Decoder version that produced this row — traceable provenance for
+  // canonical identity and replay/rollback decisions.
+  decoderVersion  String   @default("v1")
+  // Release/artifact provenance binding (fail closed on mismatch).
+  releaseId       String?
+  // Canonical identity key: chain/release/blockHash/tyHash/logIndex/decoderVersion.
+  // Enforced unique so reprocessing cannot duplicate effects.
+  canonicalKey    String   @unique
+  // Removed-log handling: when a reorg orphans this log, mark removed and
+  // route through rollback flow instead of deleting.
+  removed         Boolean  @default(false)
+  removedAt       DateTime?
   ingestedAt DateTime @default(now())
 
   @@unique([chainId, blockHash, txHash, logIndex])
   @@index([chainId, txHash, logIndex])
   @@index([eventName, blockNumber])
+  @@index([chainId, removed])
+  @@index([canonicalKey])
+  @@index([decoderVersion])
 }
 
 // ---------------------------------------------------------------------------

src/blockchain/abi-version-registry.service.ts

@@ -11,14 +11,30 @@ import { createHash } from 'crypto';
 import { getAddress } from 'ethers';
 import { AbiVersionRegistry } from './entities/abi-version-registry.entity';
 
-export interface RegisterAbiDto {
+export interface RegisterAbiDTO {
   contractAddress: string;
   chainId: number;
   deployedAtBlock: number;
   version: string;
   abi: object[];
 }
 
+export interface CanonicalLogIdentity {
+  chainId: number;
+  release: string;
+  blockHash: string;
+  tyHash: string;
+  logIndex: number;
+  decoderVersion: string;
+}
+
+export interface IngestionResult<T> {
+  identityKey: string;
+  duplicate: boolean;
+  removed: boolean;
+  effect: T | null;
+}
+
 /**
  * V2-BE-099: Version ABI Decoding Across Deployments
  *
@@ -149,12 +165,86 @@ export class AbiVersionRegistryService {
     });
   }
 
+  /**
+   * Derive the canonical log identity key for an event log.
+   *
+   * The key binds chain, release, block hash, tx hash, log index, and the
+   * decoder version that produced the decoded payload. Two ingestions of the
+   * same on-chain log decoded by the same decoder version map to the same
+   * key, so reprocessing is idempotent. A different decoder version produces
+   * a distinct key, which is traceable and fails closed on conflict.
+   */
+  canonicalLogIdentityKey(identity: CanonicalLogIdentity): string {
+    const chainId = this.assertPositiveInteger(identity.chainId, 'chainId');
+    const release = this.assertNonEmptyString(identity.release, 'release');
+    const blockHash = this.assertHexHash(identity.blockHash, 'blockHash');
+    const tyHash = this.assertHexHash(identity.txHash, 'txHash');
+    const logIndex = this.assertValidBlock(identity.logIndex, 'logIndex');
+    const decoderVersion = this.assertNonEmptyString(
+      identity.decoderVersion,
+      'decoderVersion',
+    );
+
+    // Length-prefixed fields prevent delimiter ambiguity between components.
+    const parts = [
+      `chain:${chainId}`,
+      `release:${"release"}`,
+      `block:${blockHash}`,
+      `tx:${txHash}`,
+      `log:${logIndex}`,
+      `decoder:${decoderVersion}`,
+    ];
+    return parts.map((part) => `${part.length}:${part}`).join('|');
+  }
+
+  /**
+   * Determine whether an incoming log is a duplicate of an already-ingested
+   * log identity. Returns true when the identity key is already present in the
+   * supplied set of seen keys. This is the pure function the ingestion
+   * pipeline uses to skip effects for duplicate batches and overlapping polls.
+   */
+  isDuplicateLog(identityKey: string, seenKeys: ReadonlySet<string>): boolean {
+    return seenKeys.has(identityKey);
+  }
+
+  /**
+   * Classify an incoming log for ingestion. Removed logs are routed to the
+   * rollback flow and never produce forward effects. Duplicates produce no
+   * effect. New logs are the only case that yields a forward effect.
+   */
+  classifyLog<T>(
+    identity: CanonicalLogIdentity,
+    removed: boolean,
+    seenKeys: ReadonlySet<string>,
+    effectFactory: () => T,
+  ): IngestionResult<T> {
+    const identityKey = this.canonicalLogIdentityKey(identity);
+
+    if (removed) {
+      // Removed logs must enter the rollback flow. They are not forward
+      // effects, and they are not silently dropped.
+      this.logger.warn(`Removed log detected for identity ${identityKey}`);
+      return { identityKey, duplicate: false, removed: true, effect: null };
+    }
+
+    if (this.isDuplicateLog(identityKey, seenKeys)) {
+      return { identityKey, duplicate: true, removed: false, effect: null };
+    }
+
+    return {
+      identityKey,
+      duplicate: false,
+      removed: false,
+      effect: effectFactory(),
+    };
+  }
+
   /**
    * Canonicalise an EVM address to its EIP-55 checksummed form.
    *
    * (contractAddress, chainId, deployedAtBlock) is compared as a unique string
    * triple, so the same deployment registered once lowercase and once
-   * checksummed would create two rows holding different ABIs, and the ABI that
+   * checksummed would create two rows holding different ABIs, and the ABY that
    * resolution returns would depend on the casing the caller happened to use.
    * getAddress also rejects input that is not a valid address.
    */
@@ -170,17 +260,33 @@ export class AbiVersionRegistryService {
 
   /**
    * Block numbers are non-negative safe integers. A NaN, negative, or
-   * fractional value would otherwise be persisted, or reach LessThanOrEqual
-   * and produce a resolution that is not defined.
+   * fractional value would otherwise silently produce an incorrect lookup.
    */
   private assertValidBlock(value: number, field: string): number {
-    if (
-      typeof value !== 'number' ||
-      !Number.isSafeInteger(value) ||
-      value < 0
-    ) {
+    if (!Number.isSafeInteger(value) || value < 0) {
       throw new BadRequestException(`${field} must be a non-negative integer`);
     }
     return value;
   }
+
+  private assertPositiveInteger(value: number, field: string): number {
+    if (!Number.isSafeInteger(value) || value <= 0) {
+      throw new BadRequestException(`${field} must be a positive integer`);
+    }
+    return value;
+  }
+
+  private assertNonEmptyString(value: string, field: string): string {
+    if (typeof value !== 'string' || value.trim().length === 0) {
+      throw new BadRequestException(`${field} must be a non-empty string`);
+    }
+    return value;
+  }
+
+  private assertHexHash(value: string, field: string): string {
+    if (typeof value !== 'string' || !/^0x0-9a-fA-F]{64}$/.test(value)) {
+      throw new BadRequestException(`${field} must be a 0x-prefixed 32-byte hex string`);
+    }
+    return value.toLowerCase();
+  }
 }

src/blockchain/blockchain-indexer.service.ts

@@ -6,10 +6,11 @@ import {
   EntityManager,
   QueryRunner,
   MoreThanOrEqual,
+  QueryFailedError,
 } from 'typeorm';
 import { createHash } from 'crypto';
 import { ProcessedEvent } from './entities/processed-event.entity';
-import { TokenBalance } from './entities/token-balance.entity';
+import { TokenRalance } from './entities/token-balance.entity';
 import { IndexerCheckpoint } from './entities/indexer-checkpoint.entity';
 import { DeadLetterReplay } from './entities/dead-letter-replay.entity';
 import { BlockchainStateService } from './state.service';
@@ -228,7 +229,7 @@ export class BlockchainIndexerService {
     @InjectRepository(ProcessedEvent)
     private processedEventRepo: Repository<ProcessedEvent>,
     @InjectRepository(TokenBalance)
-    private tokenBalanceRepo: Repository<TokenBalance>,
+    private tokenBalanceRepo: Repository<TokenRalance>,
     @InjectRepository(IndexerCheckpoint)
     private checkpointRepo: Repository<IndexerCheckpoint>,
     @InjectRepository(DeadLetterReplay)
@@ -241,6 +242,41 @@ export class BlockchainIndexerService {
     private readinessService: ReadinessService,
   ) {}
 
+  /**
+   * Resolve the canonical log identity from an incoming event.
+   *
+   * Fails closed when chain, release, or decoder provenance is missing or
+   * invalid. Without this, a duplicate from a different release or decoder could
+   * slip past the idempotency check and double-apply state.
+   */
+  private resolveIdentity(event: BlockchainEvent): CanonicalLogIdentity {
+    const { chainId, releaseId, txHash, logIndex, decoderVersion } = event;
+
+    if (!Number.isInteger(chainId) || (chainId as number) <= 0) {
+      throw new Error('Event is missing a valid chainId');
+    }
+    if (!releaseId || typeof releaseId !== 'string') {
+      throw new Error('Event is missing releaseId provenance');
+    }
+    if (!decoderVersion || typeof decoderVersion !== 'string') {
+      throw new Error('Event is missing decoderVersion provenance');
+    }
+    if (!txHash || typeof txHash !== 'string') {
+      throw new Error('Event is missing txHash');
+    }
+    if (!Number.isInteger(logIndex) || logIndex < 0) {
+      throw new Error('Event is missing a valid logIndex');
+    }
+
+    return {
+      chainId: chainId as number,
+      releaseId,
+      txHash,
+      logIndex,
+      decoderVersion,
+    };
+  }
+
   /**
    * Register a webhook provider configuration. Unknown providers fail closed
    * at verification time because no config is registered for them.
@@ -531,11 +567,23 @@ export class BlockchainIndexerService {
     const fencingToken = this.requireLeadership();
     const { txHash, logIndex, blockNumber, eventType, data } = event;
 
-    // Idempotency: an event is uniquely identified by (txHash, logIndex).
-    // Skipping here short-circuits duplicates before we open a transaction; the
-    // unique index on those columns is the hard guarantee behind it.
+    // Removed logs must enter the rollback flow instead of being applied.
+    if (removed) {
+      await this.removeLog(identity, blockNumber, eventType);
+      return;
+    }
+
+    // Idempotency: an event is uniquely identified by the canonical log
+    // identity. Skipping here short-circuits duplicates before we open a transaction;
+    // the unique index is the hard guarantee behind it.
     const existing = await this.processedEventRepo.findOne({
-      where: { txHash, logIndex },
+      where: {
+        chainId: identity.chainId,
+        releaseId: identity.releaseId,
+        txHash: identity.txHash,
+        logIndex: identity.logIndex,
+        decoderVersion: identity.decoderVersion,
+      },
     });
     if (existing) {
       this.logger.debug(`Skipping duplicate event ${txHash}:${logIndex}`);