feat(backend): per-network ledger cursor, gap alerts, reindex reset

backend/prisma/migrations/20260327120000_ledger_cursors/migration.sql

@@ -0,0 +1,24 @@
+-- CreateTable
+CREATE TABLE "ledger_cursors" (
+    "network" TEXT NOT NULL,
+    "last_processed_ledger" INTEGER NOT NULL,
+    "updated_at" TIMESTAMP(3) NOT NULL,
+
+    CONSTRAINT "ledger_cursors_pkey" PRIMARY KEY ("network")
+);
+
+-- CreateTable
+CREATE TABLE "ledger_gap_alert_dedup" (
+    "network" TEXT NOT NULL,
+    "last_fired_at" TIMESTAMP(3) NOT NULL,
+    "last_gap_size" INTEGER,
+    "last_processed_ledger" INTEGER,
+    "latest_ledger" INTEGER,
+
+    CONSTRAINT "ledger_gap_alert_dedup_pkey" PRIMARY KEY ("network")
+);
+
+-- Seed default network from legacy indexer_state (if present)
+INSERT INTO "ledger_cursors" ("network", "last_processed_ledger", "updated_at")
+SELECT 'testnet', COALESCE((SELECT MAX("last_ledger") FROM "indexer_state"), 0), NOW()
+WHERE NOT EXISTS (SELECT 1 FROM "ledger_cursors" WHERE "network" = 'testnet');

backend/prisma/migrations/migration_lock.toml

@@ -0,0 +1,3 @@
+# Please do not edit this file manually
+# It should be added in your version-control system (i.e. Git)
+provider = "postgresql"

backend/prisma/schema.prisma

@@ -110,6 +110,28 @@ model IndexerState {
   @@map("indexer_state")
 }
 
+/// Per-Stellar-network indexer progress. `last_processed_ledger` is the last ledger
+/// sequence whose relevant contract events have been applied (cursor advances in the
+/// same DB transaction as raw event / projection upserts).
+model LedgerCursor {
+  network               String   @id
+  lastProcessedLedger   Int      @map("last_processed_ledger")
+  updatedAt             DateTime @updatedAt @map("updated_at")
+
+  @@map("ledger_cursors")
+}
+
+/// One row per network: last time a ledger gap alert was emitted (dedup / cooldown).
+model LedgerGapAlertDedup {
+  network               String   @id
+  lastFiredAt           DateTime @map("last_fired_at")
+  lastGapSize           Int?     @map("last_gap_size")
+  lastProcessedLedger   Int?     @map("last_processed_ledger")
+  latestLedger          Int?     @map("latest_ledger")
+
+  @@map("ledger_gap_alert_dedup")
+}
+
 model RawEvent {
   id              Int      @id @default(autoincrement())
   txHash          String

backend/src/admin/admin.controller.spec.ts

@@ -1,14 +1,20 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { ExecutionContext, ForbiddenException } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
 import { Request } from 'express';
 import { AdminController } from './admin.controller';
 import { AdminService } from './admin.service';
 import { AuditService } from './audit.service';
 import { AdminRoleGuard } from './guards/admin-role.guard';
 import { JwtAuthGuard } from '../auth/guards/jwt-auth.guard';
+import { PrivacyService } from '../maintenance/privacy.service';
+import { RateLimitService } from '../rate-limit/rate-limit.service';
 
 const mockAdminService = { enqueueReindex: jest.fn(), setFeatureFlag: jest.fn(), getFeatureFlags: jest.fn() };
 const mockAuditService = { write: jest.fn(), findAll: jest.fn() };
+const mockConfigService = {
+  get: jest.fn((key: string, def?: string) => (key === 'STELLAR_NETWORK' ? 'testnet' : def)),
+};
 
 const adminReq = (role = 'admin') => ({ user: { walletAddress: 'GADMIN', role }, ip: '127.0.0.1' });
 const toExecutionContext = (role?: string): ExecutionContext =>
@@ -26,6 +32,9 @@ describe('AdminController', () => {
       providers: [
         { provide: AdminService, useValue: mockAdminService },
         { provide: AuditService, useValue: mockAuditService },
+        { provide: ConfigService, useValue: mockConfigService },
+        { provide: PrivacyService, useValue: {} },
+        { provide: RateLimitService, useValue: {} },
       ],
     })
       .overrideGuard(JwtAuthGuard).useValue({ canActivate: () => true })
@@ -43,12 +52,30 @@ describe('AdminController', () => {
     it('enqueues job and writes audit row', async () => {
       mockAdminService.enqueueReindex.mockResolvedValue('job-123');
       const result = await controller.reindex({ fromLedger: 500 }, adminReq() as unknown as Request);
-      expect(result).toEqual({ jobId: 'job-123', fromLedger: 500, status: 'queued' });
-      expect(mockAdminService.enqueueReindex).toHaveBeenCalledWith(500);
+      expect(result).toEqual({
+        jobId: 'job-123',
+        fromLedger: 500,
+        network: 'testnet',
+        status: 'queued',
+      });
+      expect(mockAdminService.enqueueReindex).toHaveBeenCalledWith(500, 'testnet');
       expect(mockAuditService.write).toHaveBeenCalledWith(
-        expect.objectContaining({ actor: 'GADMIN', action: 'reindex', payload: expect.objectContaining({ fromLedger: 500 }) }),
+        expect.objectContaining({
+          actor: 'GADMIN',
+          action: 'reindex',
+          payload: expect.objectContaining({ fromLedger: 500, network: 'testnet' }),
+        }),
       );
     });
+
+    it('passes explicit network to enqueue', async () => {
+      mockAdminService.enqueueReindex.mockResolvedValue('job-456');
+      await controller.reindex(
+        { fromLedger: 100, network: 'public' },
+        adminReq() as unknown as Request,
+      );
+      expect(mockAdminService.enqueueReindex).toHaveBeenCalledWith(100, 'public');
+    });
   });
 
   describe('GET /admin/audits', () => {

backend/src/admin/admin.controller.ts

@@ -12,6 +12,7 @@ import {
   HttpCode,
   HttpStatus,
 } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
 import { ApiBearerAuth, ApiOperation, ApiTags } from '@nestjs/swagger';
 import { IsEnum, IsOptional, IsString } from 'class-validator';
 import { Request } from 'express';
@@ -66,14 +67,16 @@ export class AdminController {
   @ApiOperation({ summary: 'Enqueue a ledger reindex job from a given ledger' })
   async reindex(@Body() dto: ReindexDto, @Req() req: AdminRequest) {
     const actor = req.user?.walletAddress ?? 'unknown';
-    const jobId = await this.adminService.enqueueReindex(dto.fromLedger);
+    const network =
+      dto.network ?? this.configService.get<string>('STELLAR_NETWORK', 'testnet');
+    const jobId = await this.adminService.enqueueReindex(dto.fromLedger, network);
     await this.auditService.write({
       actor,
       action: 'reindex',
-      payload: { fromLedger: dto.fromLedger, jobId },
+      payload: { fromLedger: dto.fromLedger, network, jobId },
       ipAddress: req.ip,
     });
-    return { jobId, fromLedger: dto.fromLedger, status: 'queued' };
+    return { jobId, fromLedger: dto.fromLedger, network, status: 'queued' };
   }
 
   /**

backend/src/admin/admin.service.spec.ts

@@ -0,0 +1,62 @@
+const mockQueueAdd = jest.fn().mockResolvedValue({ id: 'queued-job-id' });
+
+jest.mock('bullmq', () => ({
+  Queue: jest.fn().mockImplementation(() => ({
+    add: (...args: unknown[]) => mockQueueAdd(...args),
+  })),
+}));
+
+jest.mock('../redis/client', () => ({
+  getBullMQConnection: () => ({}),
+}));
+
+import { AdminService } from './admin.service';
+
+describe('AdminService', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+    mockQueueAdd.mockResolvedValue({ id: 'queued-job-id' });
+  });
+
+  describe('enqueueReindex', () => {
+    it('sets last_processed_ledger to fromLedger-1 and enqueues with network', async () => {
+      const upsert = jest.fn();
+      const prisma = {
+        $transaction: jest.fn(async (fn: (t: { ledgerCursor: { upsert: jest.Mock } }) => Promise<void>) =>
+          fn({ ledgerCursor: { upsert } })),
+      };
+
+      const svc = new AdminService(prisma as never);
+      const jobId = await svc.enqueueReindex(500, 'testnet');
+
+      expect(jobId).toBe('queued-job-id');
+      expect(upsert).toHaveBeenCalledWith({
+        where: { network: 'testnet' },
+        create: { network: 'testnet', lastProcessedLedger: 499 },
+        update: { lastProcessedLedger: 499 },
+      });
+      expect(mockQueueAdd).toHaveBeenCalledWith(
+        'reindex',
+        { fromLedger: 500, network: 'testnet' },
+        expect.objectContaining({
+          jobId: expect.stringMatching(/^reindex-testnet-500-/),
+        }),
+      );
+    });
+
+    it('clamps at 0 when fromLedger is 0', async () => {
+      const upsert = jest.fn();
+      const prisma = {
+        $transaction: jest.fn(async (fn: (t: { ledgerCursor: { upsert: jest.Mock } }) => Promise<void>) =>
+          fn({ ledgerCursor: { upsert } })),
+      };
+      const svc = new AdminService(prisma as never);
+      await svc.enqueueReindex(0, 'public');
+      expect(upsert).toHaveBeenCalledWith(
+        expect.objectContaining({
+          create: expect.objectContaining({ lastProcessedLedger: 0 }),
+        }),
+      );
+    });
+  });
+});

backend/src/admin/admin.service.ts

@@ -20,9 +20,25 @@ export class AdminService {
     });
   }
 
-  async enqueueReindex(fromLedger: number): Promise<string> {
-    const job = await this.reindexQueue.add('reindex', { fromLedger }, { jobId: `reindex-${fromLedger}-${Date.now()}` });
-    this.logger.log(`Reindex job enqueued: ${job.id} from ledger ${fromLedger}`);
+  /**
+   * Reset per-network cursor so the next indexer pass starts at `fromLedger`,
+   * then enqueue a BullMQ job to drive catch-up (see ReindexWorkerService).
+   */
+  async enqueueReindex(fromLedger: number, network: string): Promise<string> {
+    const lastProcessed = Math.max(0, fromLedger - 1);
+    await this.prisma.$transaction(async (tx) => {
+      await tx.ledgerCursor.upsert({
+        where: { network },
+        create: { network, lastProcessedLedger: lastProcessed },
+        update: { lastProcessedLedger: lastProcessed },
+      });
+    });
+    const job = await this.reindexQueue.add(
+      'reindex',
+      { fromLedger, network },
+      { jobId: `reindex-${network}-${fromLedger}-${Date.now()}` },
+    );
+    this.logger.log(`Reindex job enqueued: ${job.id} network=${network} fromLedger=${fromLedger}`);
     return job.id!;
   }
 

backend/src/admin/dto/reindex.dto.ts

@@ -1,9 +1,19 @@
-import { IsInt, Min } from 'class-validator';
-import { ApiProperty } from '@nestjs/swagger';
+import { IsInt, IsOptional, IsString, Matches, Min } from 'class-validator';
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
 
 export class ReindexDto {
   @ApiProperty({ description: 'Ledger sequence to reindex from', minimum: 0 })
   @IsInt()
   @Min(0)
   fromLedger!: number;
+
+  @ApiPropertyOptional({
+    description:
+      'Stellar logical network id (must match STELLAR_NETWORK / indexer cursor row). Defaults to server config.',
+    example: 'testnet',
+  })
+  @IsOptional()
+  @IsString()
+  @Matches(/^[a-z0-9][a-z0-9_-]{0,62}$/i)
+  network?: string;
 }

backend/src/claims/claims.service.ts

@@ -44,6 +44,7 @@ export class ClaimsService {
   private readonly cacheTtl: number;
   private readonly ipfsGateway: string;
   private readonly maxAcceptableLag = 5;
+  private readonly indexerNetwork: string;
 
   constructor(
     private readonly prisma: PrismaService,
@@ -55,6 +56,7 @@ export class ClaimsService {
   ) {
     this.cacheTtl = this.config.get<number>('CACHE_TTL_SECONDS', 60);
     this.ipfsGateway = this.config.get<string>('IPFS_GATEWAY', 'https://ipfs.io');
+    this.indexerNetwork = this.config.get<string>('STELLAR_NETWORK', 'testnet');
   }
 
   async listClaims(params: ListClaimsParams): Promise<ClaimsListResponseDto> {
@@ -183,10 +185,16 @@ export class ClaimsService {
   }
 
   private async getLastLedger(): Promise<number> {
-    const indexerState = await this.prisma.indexerState.findFirst({
+    const cursor = await this.prisma.ledgerCursor.findUnique({
+      where: { network: this.indexerNetwork },
+    });
+    if (cursor) {
+      return cursor.lastProcessedLedger;
+    }
+    const legacy = await this.prisma.indexerState.findFirst({
       orderBy: { lastLedger: 'desc' },
     });
-    return indexerState?.lastLedger || 0;
+    return legacy?.lastLedger ?? 0;
   }
 
   private transformClaim(claim: ClaimWithVotes, lastLedger: number): ClaimDetailResponseDto {

backend/src/common/middleware/idempotency.middleware.ts

@@ -46,8 +46,8 @@
 import { Injectable, NestMiddleware, BadRequestException, Logger } from '@nestjs/common';
 import { Request, Response, NextFunction } from 'express';
 import { createHash } from 'crypto';
-import { getIdempotencyEntry, setIdempotencyEntry } from '../redis/cache';
-import { TTL } from '../redis/config';
+import { getIdempotencyEntry, setIdempotencyEntry } from '../../redis/cache';
+import { TTL } from '../../redis/config';
 
 /** Bump this when any covered endpoint's response schema changes. */
 export const IDEMPOTENCY_VERSION = 1;
@@ -82,8 +82,13 @@ export class IdempotencyMiddleware implements NestMiddleware {
       .update(`${req.method}:${req.path}:${rawKey}:${subject}`)
       .digest('hex');
 
-    // Cache hit — replay stored response
-    const cached = await getIdempotencyEntry(cacheKey, IDEMPOTENCY_VERSION);
+    // Cache hit — replay stored response (fail open if Redis/cache layer errors)
+    let cached: Awaited<ReturnType<typeof getIdempotencyEntry>> = null;
+    try {
+      cached = await getIdempotencyEntry(cacheKey, IDEMPOTENCY_VERSION);
+    } catch (err) {
+      this.logger.warn(`Idempotency lookup failed (fail open): ${String(err)}`);
+    }
     if (cached) {
       this.logger.debug(`Idempotency replay: key=${rawKey} subject=${subject} path=${req.path}`);
       res.setHeader('Idempotency-Replayed', 'true');

backend/src/config/env.validation.ts

@@ -10,6 +10,19 @@ export const validationSchema = Joi.object({
     .description("PostgreSQL connection URL"),
   REDIS_URL: Joi.string().required().description("Redis connection URL"),
   SOROBAN_RPC_URL: Joi.string().required().description("Soroban RPC endpoint"),
+  STELLAR_NETWORK: Joi.string()
+    .default("testnet")
+    .description("Logical network id for indexer cursor isolation (e.g. testnet, public)"),
+  INDEXER_GAP_ALERT_THRESHOLD_LEDGERS: Joi.number()
+    .integer()
+    .min(1)
+    .default(100)
+    .description("Alert when chain head minus last_processed exceeds this"),
+  INDEXER_GAP_ALERT_COOLDOWN_MS: Joi.number()
+    .integer()
+    .min(60_000)
+    .default(3_600_000)
+    .description("Minimum milliseconds between gap alerts per network"),
   // IPFS Configuration
   IPFS_PROVIDER: Joi.string()
     .valid("mock", "pinata")

backend/src/indexer/indexer.module.ts

@@ -1,12 +1,14 @@
 import { Module } from '@nestjs/common';
+import { ConfigModule } from '@nestjs/config';
 import { IndexerService } from './indexer.service';
 import { IndexerWorker } from './indexer.worker';
+import { ReindexWorkerService } from './reindex.worker';
 import { PrismaModule } from '../prisma/prisma.module';
 import { RpcModule } from '../rpc/rpc.module';
 
 @Module({
-  imports: [PrismaModule, RpcModule],
-  providers: [IndexerService, IndexerWorker],
+  imports: [PrismaModule, RpcModule, ConfigModule],
+  providers: [IndexerService, IndexerWorker, ReindexWorkerService],
   exports: [IndexerService],
 })
 export class IndexerModule {}