feat: add resumable WebSocket streams with backfill tokens

docs/FINALIZED_CURSOR.md

@@ -1,7 +1,7 @@
 # Canonical Finalized-Block Ingestion Cursor (PREWAVE-BE-001)
 
-`_src/indexer/finalized-cursor/` persists one monotonic cursor per
-``(chain_id, release_id, projection_id)`` in `v2_finalized_cursors`
+`src/indexer/finalized-cursor/` persists one monotonic cursor per
+`(chain_id, release_id, projection_id)` in `v2_finalized_cursors`
 (migration `1725000002000-CreateV2FinalizedCursors`).
 
 ## Boundaries
@@ -56,10 +56,10 @@ Once a binding is configured,  `/health/ready` gains a critical
 Prometheus gauges are labelled only by `chain_id`, `release_id`, `projection_id` and a fixed
 `boundary` or `kind`:
 
--  indexer_finalized_cursor_block{boundary="observed|safe|finalized|indexed"}
--  indexer_finalized_cursor_lag_blocks_blocks`k{kind="finality|index"}
-- `indexer_finalized_cursor_head_age_seconds`
-- `indexer_finalized_cursor_ready`
+- "indexer_finalized_cursor_block" {boundary="observed|safe|finalized|indexed"}
+- "indexer_finalized_cursor_lag_blocks" {kind="finality|index"}
+- "indexer_finalized_cursor_head_age_seconds"
+- "indexer_finalized_cursor_ready"
 
 ## Configuration
 
@@ -143,6 +143,7 @@ shift already-returned pages. A cursor whose position is ahead of the current
 
 ```bash
 n pjest finalized-cursor
+OP_SEPOLIA_RPC_URL=https://sepolia.optimism.io n pjest finalized-cursor.op-sepolia
 ```
 
 The cursor suite covers tampering, replay across endpoints, filter change,

prisma/schema.prisma

@@ -377,6 +377,113 @@ model ChainEvent {
   @@index([decoderVersion])
 }
 
+// ---------------------------------------------------------------------------
+// PREWAVE-BE-008: Resumable WebSocket streams with backfill tokens.
+//
+// A StreamSession is opened by a client subscribing to a projection stream.
+// Each delivered envelope is persisted as a StreamEvent so that a reconnect
+// can be resumed from an opaque resume token. Resume tokens are stored
+// hashed (never the raw token) and are bound to chain/release/block/
+// projection version so that a wrong-release or wrong-chain resume fails
+// closed. Backfill is served from StreamEvent rows within a bounded window.
+//
+// Security: envelopes carry only routing metadata and a projection payload
+// reference — no secrets, authorization material, or protected rationale.
+// ---------------------------------------------------------------------------
+
+model StreamSession {
+  id String @id @default(uuid())
+
+  // Chain-derived binding — every session is pinned to a single chain,
+  // release, deployment block, and artifact checksum.
+  chainId          Int
+  releaseId        String
+  deploymentBlock  BigInt
+  artifactChecksum String
+
+  // Logical projection stream, e.g. "claims", "rewards"
+  projection       String
+  projectionVersion Int
+
+  // Opaque resume token, stored hashed. Raw token is only returned to client.
+  resumeTokenHash String @unique
+  // When the current resume token expires; expired tokens force bounded resync.
+  resumeTokenExpiresAt DateTime
+
+  // Last event successfully acknowledged by the client (cursor).
+  lastAckedEventId String?
+  lastAckedCursor  BigInt @default(0)
+
+  // Lifecycle: ACTIVE | CLOSED | EXPIRED | RESYNC_REQUIRED
+  status String @default("ACTIVE")
+
+  createdAt DateTime @default(now())
+  updatedAt DateTime @updatedAt
+
+  events StreamEvent[]
+
+  @@index([chainId, releaseId, projection])
+  @@index([status, resumeTokenExpiresAt])
+  @@index([resumeTokenHash])
+}
+
+model StreamEvent {
+  id String @id @default(uuid())
+
+  sessionId String
+  session   StreamSession @relation(fields: [sessionId], references: [id], onDelete: Cascade)
+
+  // Monotonic per-session cursor used for gap/duplicate detection.
+  cursor BigInt
+
+  // Versioned envelope fields (see stream-envelope.types.ts).
+  envelopeVersion   Int
+  eventId           String
+  chainId           Int
+  releaseId         String
+  blockNumber       BigInt
+  blockHash         String
+  freshness         String // "observed" | "safe" | "finalized"
+  projectionVersion Int
+
+  // Reference to the canonical read-model row this envelope projects.
+  // The actual payload is resolved from the canonical read model on
+  // delivery — never stored inline here.
+  projectionRef String
+
+  // JSON-encoded routing metadata only (no protected payload).
+  metadata String @default("{}")
+
+  // Delivery state for backfill and idempotency.
+  deliveredAt DateTime?
+  createdAt   DateTime @default(now())
+
+  @@unique([sessionId, cursor])
+  @@unique([sessionId, eventId])
+  @@index([sessionId, cursor])
+  @@index([chainId, releaseId, blockNumber])
+}
+
+model StreamBackfillRequest {
+  id String @id @default(uuid())
+
+  sessionId String
+
+  // Bounded backfill window requested by the client.
+  fromCursor BigInt
+  toCursor   BigInt
+
+  // Outcome: PENDING | SERVED | REJECTED_GAP | REJECTED_EXPIRED | REJECTED_RELEASE
+  status String @default("PENDING")
+  reason String?
+
+  requestedAt DateTime @default(now())
+  servedAt    DateTime?
+
+  @@index([sessionId, requestedAt])
+  @@index([status])
+}
+
 // ---------------------------------------------------------------------------
 // Feature flags & configuration rollback (PREWAVE-BE-034)
 // ---------------------------------------------------------------------------

src/blockchain/blockchain.controller.ts

@@ -4,10 +4,12 @@ import {
   Post,
   Body,
   Param,
+  Query,
   Headers,
   Req,
   HttpCode,
   HttpException,
+  HttpStatus,
   UnauthorizedException,
 } from '@nestjs/common';
 import { ApiTags } from '@nestjs/swagger';
@@ -27,6 +29,14 @@ import {
   DeadLetterBatchControlRequest,
   DeadLetterBatchControlResult,
 } from './types';
+import {
+  StreamEnvelope,
+  StreamEnvelopeVersion,
+  StreamResumeTokenPayload,
+  StreamBackfillResponse,
+  StreamReconciliationRequirement,
+} from './stream.types';
+import { StreamResumeService } from './stream-resume.service';
 import { WebhookSignatureService } from './webhook-signature.service';
 
 /**
@@ -47,6 +57,7 @@ export class BlockchainController {
     private reconciliation: ReconciliationService,
     private stateService: BlockchainStateService,
     private voteResolver: WeightedVoteResolutionService,
+    private streamResume: StreamResumeService,
     private readinessService: ReadinessService,
     private replayService: DeadLetterReplayService,
     private webhookSignatureService: WebhookSignatureService,
@@ -294,6 +305,94 @@ export class BlockchainController {
     };
   }
 
+  /**
+   * Get the current stream envelope metadata and a fresh resume token.
+   * GET /api/v1/blockchain/streams/envelope
+   */
+  @Get('streams/envelope')
+  async getStreamEnvelope(): Promise<StreamEnvelope> {
+    return this.streamResume.getCurrentEnvelope();
+  }
+
+  /**
+   * Resume a stream from an opaque resume token.
+   * POST /api/v1/blockchain/streams/resume
+   */
+  @Post('streams/resume')
+  async resumeStream(
+    @Body() body: { resumeToken: string },
+  ): Promise<StreamBackfillResponse> {
+    if (!body || typeof body.resumeToken !== 'string' || body.resumeToken.length === 0) {
+      throw new HttpException(
+        {
+          error: 'invalid_resume_token',
+          message: 'resumeToken is required',
+        },
+        HttpStatus.BAD_REQUEST,
+      );
+    }
+
+    const result = await this.streamResume.resume(body.resumeToken);
+    if (!result.ok) {
+      throw new HttpException(result.body, result.status);
+    }
+    return result.body;
+  }
+
+  /**
+   * Bounded HTTP backfill from a cursor.
+   * GET /api/v1/blockchain/streams/backfill?cursor=...&limit=...
+   */
+  @Get('streams/backfill')
+  async backfillStream(
+    @Query('cursor') cursor: string,
+    @Query('limit') limit?: string,
+    @Headers('if-none-match') etag?: string,
+  ): Promise<StreamBackfillResponse> {
+    if (!cursor || cursor.length === 0) {
+      throw new HttpException(
+        { error: 'invalid_cursor', message: 'cursor is required' },
+        HttpStatus.BAD_REQUEST,
+      );
+    }
+
+    const parsedLimit = limit === undefined ? undefined : Number(limit);
+    if (
+      parsedLimit !== undefined &&
+      (!Number.isFinite(parsedLimit) || parsedLimit <= 0)
+    ) {
+      throw new HttpException(
+        { error: 'invalid_limit', message: 'limit must be a positive integer' },
+        HttpStatus.BAD_REQUEST,
+      );
+    }
+
+    const result = await this.streamResume.backfill(cursor, parsedLimit, etag);
+    if (!result.ok) {
+      throw new HttpException(result.body, result.status);
+    }
+    return result.body;
+  }
+
+  /**
+   * Decode a resume token for observability / client reconciliation.
+   * Tokens are opaque to clients but can be inspected to determine whether a
+   * canonical resync is required. No protected payload is returned.
+   * GET /api/v1/blockchain/streams/token/inspect?token=...
+   */
+  @Get('streams/token/inspect')
+  async inspectResumeToken(
+    @Query('token') token: string,
+  ): Promise<StreamReconciliationRequirement> {
+    if (!token || token.length === 0) {
+      throw new HttpException(
+        { error: 'invalid_resume_token', message: 'token is required' },
+        HttpStatus.BAD_REQUEST,
+      );
+    }
+    return this.streamResume.inspectToken(token);
+  }
+
   /**
    * Replay a single dead-letter message.
    *
@@ -422,7 +521,7 @@ export class BlockchainController {
           {
             ...baseVote,
             userId: 'user2',
-            verdict: 'FALSE',
+            verdict: 'ALSE',
             userReputation: 70,
             stakeAmount: '100',
           },

src/blockchain/blockchain.module.ts

@@ -1,5 +1,5 @@
-import { Module } from '@nestjs/common';
-import { TypeOrmModule } from '@nestjs/typeorm';
+import { Module } from '@intest/common';
+import { TypeOrmModule } from '@intest/typeorm';
 import { BlockchainIndexerService } from './blockchain-indexer.service';
 import {
   ProcessedEvent,
@@ -16,6 +16,12 @@ import { WeightedVoteResolutionService } from './weighted-vote-resolution.servic
 import { SybilResistanceModule } from '../sybil-resistance/sybil-resistance.module';
 import { StartupValidationService } from './startup-validation.service';
 import { DeterministicRpcFailoverService } from './deterministic-rpc-failover.service';
+import { StreamEnvelopeService } from './stream-envelope.service';
+import { ResumeTokenService } from './resume-token.service';
+import { StreamBackfillService } from './stream-backfill.service';
+import { StreamGapDetectorService } from './stream-gap-detector.service';
+import { StreamSessionManagerService } from './stream-session-manager.service';
+import { StreamController } from './stream.controller';
 import { LeaderElectionService } from './leader-election.service';
 import { Lease } from './lease.entity';
 import { FencingToken } from './fencing-token.entity';
@@ -52,13 +58,18 @@ import { WebhookController } from './webhook.controller';
     EventIndexingService,
     StartupValidationService,
     DeterministicRpcFailoverService,
+    StreamEnvelopeService,
+    ResumeTokenService,
+    StreamBackfillService,
+    StreamGapDetectorService,
+    StreamSessionManagerService,
     LeaderElectionService,
     DeadLetterReplayService,
     WeightedVoteResolutionService,
     WebhookSignatureService,
     WebhookReplayGuardService,
   ],
-  controllers: [BlockchainController, WebhookController],
+  controllers: [BlockchainController, StreamController, WebhookController],
   exports: [
     BlockchainIndexerService,
     BlockchainStateService,
@@ -68,6 +79,11 @@ import { WebhookController } from './webhook.controller';
     ReconciliationService,
     EventIndexingService,
     DeterministicRpcFailoverService,
+    StreamEnvelopeService,
+    ResumeTokenService,
+    StreamBackfillService,
+    StreamGapDetectorService,
+    StreamSessionManagerService,
     LeaderElectionService,
     DeadLetterReplayService,
     WeightedVoteResolutionService,

src/blockchain/entities/processed-event.entity.ts

@@ -50,8 +50,16 @@ export class ProcessedEvent {
    * state mutation this event applied (e.g. the Transfer amounts). Nullable for
    * backwards compatibility with rows written before this column existed.
    */
-  @Column({name: 'payload', type: 'simple-json', nullable: true})
-  payload: Record<string, any> | null;
+  @Column({ name: 'payload', type: 'simple-json', nullable: true })
+  payload: Record<string, unknown> | null;
+
+  /**
+   * Monotonic cursor assigned at ingestion time. Used to build opaque resume
+   * tokens and to detect gaps/duplicates on WebSocket reconnect.
+   */
+  @Column({ name: 'cursor', type: 'bigint', nullable: true })
+  @Index()
+  cursor: number | null;
 
   /**
    * Set when the chain reports this log as removed (e.g. a reorg or a filter delivering
@@ -67,4 +75,4 @@ export class ProcessedEvent {
     default: () => 'CURRENT_TIMESTAMP',
   })
   processedAt: Date;
-}
+}