admin: add ledger-range Horizon replay endpoint and docs

docs/admin-api.md

@@ -243,6 +243,50 @@ This action is logged with:
 
 ---
 
+### Replay Horizon Ledger Range
+
+**POST** `/api/admin/events/replay-range`
+
+Replay raw Horizon operations between two ledger sequence numbers (inclusive). This endpoint performs a best-effort mapping of operations to internal replay handlers (`bond_creation`, `withdrawal`, `attestation`) and invokes those handlers to re-process historical events. Use with caution; the operation is admin-gated and audit-logged.
+
+#### Request Body
+
+```json
+{
+  "fromLedger": 123456,
+  "toLedger": 123460
+}
+```
+
+#### Example Request
+
+```bash
+curl -X POST http://localhost:3000/api/admin/events/replay-range \
+  -H "Authorization: Bearer <ADMIN_API_KEY_RAW>" \
+  -H "Content-Type: application/json" \
+  -d '{"fromLedger":123456,"toLedger":123460}'
+```
+
+#### Example Response (200 OK)
+
+```json
+{
+  "success": true,
+  "data": {
+    "success": true,
+    "processed": 10,
+    "errors": 0
+  }
+}
+```
+
+#### Notes
+
+- The endpoint validates that `fromLedger <= toLedger` and that both are non-negative integers.
+- Processing is best-effort: operations that cannot be mapped to a known handler are skipped; handler failures are captured to the failed-events queue for later inspection.
+- All replay attempts are recorded in the audit log with action `REPLAY_LEDGER_RANGE`.
+
+
 ### Issue Impersonation Token
 
 **POST** `/api/admin/impersonate`

src/routes/admin/index.ts

@@ -27,6 +27,8 @@ import { registerAllReplayHandlers } from "../../services/replayHandlers.js";
 import { IdentityRepository } from "../../db/repositories/identityRepository.js";
 import { BondsRepository } from "../../db/repositories/bondsRepository.js";
 import { pool } from "../../db/pool.js";
+import { validate } from '../../middleware/validate.js'
+import { z } from 'zod'
 
 /**
  * Create the admin router with role and user management endpoints
@@ -380,6 +382,37 @@ export function createAdminRouter(): Router {
     }
   })
 
+  /**
+   * POST /api/admin/events/replay-range
+   * Replay raw Horizon events between `fromLedger` and `toLedger` (inclusive).
+   */
+  router.post(
+    '/events/replay-range',
+    requireUserAuth,
+    requireAdminRole,
+    validate({ body: z.object({ fromLedger: z.coerce.number().int().min(0), toLedger: z.coerce.number().int().min(0) }) }),
+    async (req: Request, res: Response, next: NextFunction) => {
+      try {
+        const authReq = req as AuthenticatedRequest
+        const admin = authReq.user!
+        const { fromLedger, toLedger } = req.body as { fromLedger: number; toLedger: number }
+
+        const result = await replayService.replayLedgerRange(
+          fromLedger,
+          toLedger,
+          admin.id,
+          admin.email,
+          admin.tenantId,
+          req.ip
+        )
+
+        res.status(200).json({ success: true, data: result })
+      } catch (error: any) {
+        res.status(400).json({ error: 'ReplayFailed', message: error.message })
+      }
+    }
+  )
+
   /**
    * POST /api/admin/replay
    * Replays a request by requestId against captured snapshot and returns diff.

src/services/replayService.ts

@@ -2,6 +2,7 @@ import { FailedInboundEventsRepository, FailedInboundEvent } from '../db/reposit
 import { auditLogService } from './audit/index.js'
 import { cache } from '../cache/redis.js'
 import { invalidateCache } from '../cache/invalidation.js'
+import { Horizon } from '@stellar/stellar-sdk'
 
 const FAILED_EVENT_CACHE_TTL = 300 // 5 minutes
 
@@ -144,4 +145,113 @@ export class ReplayService {
   async listFailedEvents(filters: { status?: any; type?: string }, limit = 50, offset = 0) {
     return this.repository.list(filters, limit, offset)
   }
+
+  /**
+   * Replay raw Horizon events between ledger sequence numbers (inclusive).
+   * This performs a best-effort mapping of operations to registered handlers
+   * (e.g. `bond_creation`, `withdrawal`, `attestation`) and invokes handlers
+   * with parsed event payloads where possible. Errors for individual events
+   * are captured via `captureFailure` and logged.
+   */
+  async replayLedgerRange(
+    fromLedger: number,
+    toLedger: number,
+    adminId: string,
+    adminEmail: string,
+    tenantId: string,
+    ipAddress?: string
+  ): Promise<{ success: boolean; processed: number; errors: number }> {
+    const HORIZON_URL = process.env.HORIZON_URL || 'https://horizon-testnet.stellar.org'
+    const server = new Horizon.Server(HORIZON_URL)
+
+    if (fromLedger > toLedger) {
+      throw new Error('fromLedger must be <= toLedger')
+    }
+
+    let processed = 0
+    let errors = 0
+
+    for (let seq = fromLedger; seq <= toLedger; seq++) {
+      try {
+        const res = await server.operations().forLedger(seq).limit(200).call()
+        for (const op of res.records) {
+          const anyOp: any = op
+          try {
+            // Map operation types to registered handler keys
+            if (anyOp.type === 'create_bond' && this.handlers.has('bond_creation')) {
+              const parsed = {
+                identity: { id: anyOp.source_account },
+                bond: { id: anyOp.id, address: anyOp.source_account, amount: anyOp.amount, duration: anyOp.duration ?? null },
+              }
+              await this.handlers.get('bond_creation')!.handle(parsed)
+              processed++
+              continue
+            }
+
+            if (anyOp.type === 'payment' && this.handlers.has('withdrawal')) {
+              const payment = anyOp
+              const parsed = {
+                id: anyOp.id,
+                pagingToken: anyOp.paging_token,
+                type: anyOp.type,
+                createdAt: new Date(anyOp.created_at),
+                bondId: `${payment.from || payment.source_account}-${anyOp.transaction_hash}`,
+                account: payment.from || payment.source_account,
+                amount: payment.amount,
+                assetType: payment.asset_type,
+                assetCode: payment.asset_code,
+                assetIssuer: payment.asset_issuer,
+                transactionHash: anyOp.transaction_hash || '',
+                operationIndex: Number.parseInt(anyOp.id.split('-').pop() ?? '0', 10) || 0,
+              }
+              await this.handlers.get('withdrawal')!.handle(parsed)
+              processed++
+              continue
+            }
+
+            // Best-effort attestation mapping
+            if ((anyOp.type && anyOp.type.toString().toLowerCase().includes('attest')) && this.handlers.has('attestation')) {
+              await this.handlers.get('attestation')!.handle(anyOp)
+              processed++
+              continue
+            }
+
+            // Unknown/unsupported op - skip
+          } catch (err: any) {
+            errors++
+            await this.captureFailure('replay_range_op_failure', { ledger: seq, op }, err?.message || 'handler failure')
+          }
+        }
+      } catch (err: any) {
+        errors++
+        await auditLogService.logAction(
+          tenantId,
+          adminId,
+          adminEmail,
+          'REPLAY_LEDGER_RANGE' as any,
+          `${fromLedger}-${toLedger}`,
+          'system',
+          { ledger: seq, error: err?.message },
+          'failure',
+          err?.message,
+          ipAddress
+        )
+      }
+    }
+
+    await auditLogService.logAction(
+      tenantId,
+      adminId,
+      adminEmail,
+      'REPLAY_LEDGER_RANGE' as any,
+      `${fromLedger}-${toLedger}`,
+      'system',
+      { fromLedger, toLedger, processed, errors },
+      errors === 0 ? 'success' : 'failure',
+      undefined,
+      ipAddress
+    )
+
+    return { success: errors === 0, processed, errors }
+  }
 }