From 2b2ebfa9f20a83f4f49e9eaae827962479406b0b Mon Sep 17 00:00:00 2001 From: peace0git Date: Mon, 29 Jun 2026 20:08:29 +0000 Subject: [PATCH] admin: add ledger-range Horizon replay endpoint and docs --- docs/admin-api.md | 44 ++++++++++++++ package-lock.json | 62 ++----------------- src/routes/admin/index.ts | 33 ++++++++++ src/services/replayService.ts | 110 ++++++++++++++++++++++++++++++++++ 4 files changed, 191 insertions(+), 58 deletions(-) diff --git a/docs/admin-api.md b/docs/admin-api.md index 03f75c51..cb5a2e2b 100644 --- a/docs/admin-api.md +++ b/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 " \ + -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` diff --git a/package-lock.json b/package-lock.json index e4c35ed4..f4630d20 100644 --- a/package-lock.json +++ b/package-lock.json @@ -43,7 +43,7 @@ "@types/express": "^4.17.21", "@types/jest": "^30.0.0", "@types/multer": "^2.1.0", - "@types/node": "^20.19.33", + "@types/node": "^20.19.37", "@types/pg": "^8.16.0", "@types/redis": "^4.0.10", "@types/supertest": "^6.0.3", @@ -2444,59 +2444,6 @@ "node": "^20.19.0 || >=22.12.0" } }, - "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/core": { - "version": "1.10.0", - "resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.10.0.tgz", - "integrity": "sha512-yq6OkJ4p82CAfPl0u9mQebQHKPJkY7WrIuk205cTYnYe+k2Z8YBh11FrbRG/H6ihirqcacOgl2BIO8oyMQLeXw==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "@emnapi/wasi-threads": "1.2.1", - "tslib": "^2.4.0" - } - }, - "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/runtime": { - "version": "1.10.0", - "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.10.0.tgz", - "integrity": "sha512-ewvYlk86xUoGI0zQRNq/mC+16R1QeDlKQy21Ki3oSYXNgLb45GV1P6A0M+/s6nyCuNDqe5VpaY84BzXGwVbwFA==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "tslib": "^2.4.0" - } - }, - "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/wasi-threads": { - "version": "1.2.1", - "resolved": "https://registry.npmjs.org/@emnapi/wasi-threads/-/wasi-threads-1.2.1.tgz", - "integrity": "sha512-uTII7OYF+/Mes/MrcIOYp5yOtSMLBWSIoLPpcgwipoiKbli6k322tcoFsxoIIxPDqW01SQGAgko4EzZi2BNv2w==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "tslib": "^2.4.0" - } - }, - "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@napi-rs/wasm-runtime": { - "version": "1.1.4", - "resolved": "https://registry.npmjs.org/@napi-rs/wasm-runtime/-/wasm-runtime-1.1.4.tgz", - "integrity": "sha512-3NQNNgA1YSlJb/kMH1ildASP9HW7/7kYnRI2szWJaofaS1hWmbGI4H+d3+22aGzXXN9IJ+n+GiFVcGipJP18ow==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "@tybys/wasm-util": "^0.10.1" - }, - "funding": { - "type": "github", - "url": "https://github.com/sponsors/Brooooooklyn" - }, - "peerDependencies": { - "@emnapi/core": "^1.7.1", - "@emnapi/runtime": "^1.7.1" - } - }, "node_modules/@rolldown/binding-win32-arm64-msvc": { "version": "1.0.3", "resolved": "https://registry.npmjs.org/@rolldown/binding-win32-arm64-msvc/-/binding-win32-arm64-msvc-1.0.3.tgz", @@ -2885,7 +2832,7 @@ "version": "20.19.41", "resolved": "https://registry.npmjs.org/@types/node/-/node-20.19.41.tgz", "integrity": "sha512-ECymXOukMnOoVkC2bb1Vc/w/836DXncOg5m8Xj1RH7xSHZJWNYY6Zh7EH477vcnD5egKNNfy2RpNOmuChhFPgQ==", - "dev": true, + "devOptional": true, "license": "MIT", "dependencies": { "undici-types": "~6.21.0" @@ -2895,7 +2842,7 @@ "version": "8.20.0", "resolved": "https://registry.npmjs.org/@types/pg/-/pg-8.20.0.tgz", "integrity": "sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==", - "dev": true, + "devOptional": true, "license": "MIT", "dependencies": { "@types/node": "*", @@ -4264,7 +4211,6 @@ "integrity": "sha512-Z0oHEHAFDZkffN8Qc39zNZjQlMDkPJRyyyZieU1VH7u8c5S+qHZ2S8ixdKIAxEjfHO7FJxXmJWgteOghVanIsg==", "dev": true, "license": "Apache-2.0", - "peer": true, "peerDependencies": { "bare-abort-controller": "*" }, @@ -11036,7 +10982,7 @@ "version": "6.21.0", "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.21.0.tgz", "integrity": "sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==", - "dev": true, + "devOptional": true, "license": "MIT" }, "node_modules/unpipe": { diff --git a/src/routes/admin/index.ts b/src/routes/admin/index.ts index f840c077..fe40c7a6 100644 --- a/src/routes/admin/index.ts +++ b/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. diff --git a/src/services/replayService.ts b/src/services/replayService.ts index fd162866..de4d6cdb 100644 --- a/src/services/replayService.ts +++ b/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 } + } }