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

- Add ledger_cursors (network PK, last_processed_ledger, updated_at) and
  ledger_gap_alert_dedup for cooldown-based gap notifications
- Indexer advances cursor inside same Prisma tx as raw_events + projections
- Gap alert when latestLedger - lastProcessed > threshold; dedupe via cooldown
- Admin POST /admin/reindex resets cursor (fromLedger-1), enqueues Bull job;
  ReindexWorkerService runs processUntilCaughtUp (disabled in test / DISABLE_REINDEX_WORKER)
- STELLAR_NETWORK + INDEXER_GAP_ALERT_* env; ClaimsService reads cursor by network
- Migration seeds testnet row from legacy indexer_state
- Tests: cursor resume, tx-bound advance, gap dedupe, admin reindex
- Fix idempotency middleware Redis import path + fail-open on lookup errors
- Jest: include *.spec.ts in testMatch

Made-with: Cursor

Author: aji70

backend/jest.config.js

@@ -2,7 +2,7 @@
 module.exports = {
   testEnvironment: "node",
   roots: ["<rootDir>/src", "<rootDir>/tests"],
-  testMatch: ["**/__tests__/**/*.test.ts", "**/*.test.ts"],
+  testMatch: ["**/__tests__/**/*.test.ts", "**/*.test.ts", "**/*.spec.ts"],
   transform: {
     "^.+\\.tsx?$": ["ts-jest", { tsconfig: "tsconfig.test.json" }],
   },

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

@@ -99,6 +99,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';
@@ -48,6 +49,7 @@ export class AdminController {
     private readonly auditService: AuditService,
     private readonly privacyService: PrivacyService,
     private readonly rateLimitService: RateLimitService,
+    private readonly configService: ConfigService,
   ) {}
 
   /**
@@ -64,14 +66,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.module.ts

@@ -1,4 +1,5 @@
 import { Module } from '@nestjs/common';
+import { ConfigModule } from '@nestjs/config';
 import { AdminController } from './admin.controller';
 import { AdminService } from './admin.service';
 import { AuditService } from './audit.service';
@@ -8,7 +9,7 @@ import { MaintenanceModule } from '../maintenance/maintenance.module';
 import { RateLimitModule } from '../rate-limit/rate-limit.module';
 
 @Module({
-  imports: [PrismaModule, AuthModule, MaintenanceModule, RateLimitModule],
+  imports: [ConfigModule, PrismaModule, AuthModule, MaintenanceModule, RateLimitModule],
   controllers: [AdminController],
   providers: [AdminService, AuditService],
   exports: [AuditService],

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;
 }