diff --git a/backend/SSE_README.md b/backend/SSE_README.md new file mode 100644 index 00000000..e705d2e1 --- /dev/null +++ b/backend/SSE_README.md @@ -0,0 +1,21 @@ +# FlowFi Real-Time Event Streaming & Indexer Overview + +This document provides a quick index for real-time event streaming via Server-Sent Events (SSE) and on-chain Soroban event indexing in the FlowFi backend. + +## Documentation Map + +- **[SSE Architecture Overview](docs/SSE_ARCHITECTURE.md)**: Details the architecture, system flow, connection handling, subscription filtering, horizontal scaling with Redis, event broadcasting logic, memory capacity, and security layers. +- **[Operational Runbook](docs/SSE_ARCHITECTURE.md#operational-runbook)**: **Full operational runbook** covering health check endpoints (`/health` and `/v1/admin/metrics`), lag monitoring thresholds, indexer environment variables, admin reset vs. replay procedures, deduplication guarantees, and the RPC outage recovery walkthrough. +- **[SSE Client Implementation Guide](docs/SSE_IMPLEMENTATION.md)**: Describes client integration (`GET /events/subscribe`), supported query parameters (`streams`, `users`, `all`), event types (`stream.created`, `stream.topped_up`, etc.), reconnection strategies, and code examples. + +## Quick Endpoint Reference + +| Endpoint | Method | Auth | Description | +|----------|--------|------|-------------| +| `/health` | `GET` | Public | Liveness and readiness check. Reports `indexerLag` and returns `503` if lag exceeds 60s while enabled. | +| `/v1/admin/metrics` | `GET` | Admin JWT | Detailed health metrics including `indexer.lastLedger`, `indexer.lagSeconds`, and `sse.activeConnections`. | +| `/v1/admin/indexer/reset` | `POST` | Admin JWT | Reset indexer `lastLedger` pointer for the next scheduled poll cycle. | +| `/v1/admin/indexer/replay` | `POST` | Admin JWT | Reset `lastLedger` pointer and immediately trigger event polling batch. | +| `/events/subscribe` | `GET` | Public / Auth | Connect to real-time Server-Sent Events stream. | + +For detailed operational guidance and troubleshooting, refer to the [Operational Runbook](docs/SSE_ARCHITECTURE.md#operational-runbook). diff --git a/backend/docs/SSE_ARCHITECTURE.md b/backend/docs/SSE_ARCHITECTURE.md index e6f5ab21..4edc1158 100644 --- a/backend/docs/SSE_ARCHITECTURE.md +++ b/backend/docs/SSE_ARCHITECTURE.md @@ -199,3 +199,173 @@ Optimization: │ - Subscription filtering │ └─────────────────────────────────────────────────┘ ``` + +## Operational Runbook + +This section details operational procedures, health monitoring metrics, environment configuration, and recovery workflows for the indexer and real-time event streaming components. + +### Monitoring & Health Check + +The backend exposes two primary endpoints for observing indexer health and operational metrics. + +#### 1. Liveness & Readiness Health Endpoint (`GET /health`) + +Unauthenticated health endpoint suitable for load balancer probes, Kubernetes liveness/readiness checks, and monitoring alerts. + +* **URL**: `GET /health` +* **Response Schema**: + ```json + { + "status": "ok", + "db": "connected", + "indexerEnabled": true, + "indexerLag": 5, + "uptime": 3600 + } + ``` +* **Key Fields**: + * `status`: `"ok"` (healthy) or `"degraded"` (unhealthy / degraded). + * `db`: `"connected"` or `"disconnected"`. + * `indexerEnabled`: `true` if `STREAM_CONTRACT_ID` is set, `false` if unconfigured/disabled. + * `indexerLag`: Integer seconds since last indexer DB state update (`updatedAt`), or `null` on cold start (when no `IndexerState` row exists yet). + * `uptime`: Node process uptime in seconds. +* **Threshold & Status Logic**: + * **HTTP 200 OK** (`status: "ok"`): DB status is `"connected"` and `indexerDegraded` is `false`. + * **HTTP 503 Service Unavailable** (`status: "degraded"`): Triggered if DB status is `"disconnected"` OR `indexerDegraded` is `true`. + * **Degraded Rule**: `indexerDegraded = indexerEnabled && indexerLag > 60` (indexer enabled and lag strictly exceeds 60 seconds). + * **Cold Start Handling**: If no `IndexerState` record exists in the database (`indexerLag === -1` internally), `indexerLag` is returned as `null` and does **not** trigger degradation (returns HTTP 200 OK as long as DB is connected). + * **Disabled Handling**: If `STREAM_CONTRACT_ID` is unset (`indexerEnabled: false`), lag degradation checks are bypassed and `/health` returns HTTP 200 OK. + +##### On-Call Operational Actions by Health Status + +| `indexerLag` Value | HTTP Status | Operational Status | On-Call Action | +|-------------------|-------------|--------------------|----------------| +| `indexerLag <= 60s` | `200 OK` | **Healthy** | Normal operation. | +| `indexerLag > 60s` | `503 Service Unavailable` | **Degraded** | Indexer is stale or stalled. Check Soroban RPC connectivity (`SOROBAN_RPC_URL`), worker logs (`[SorobanWorker]`), and DB load. If RPC went down, execute the RPC Outage Recovery workflow below. | +| `indexerLag: null` | `200 OK` | **Cold Start** | Indexer state record has not yet been created. Wait for initial poll cycle to complete. | + +#### 2. Admin Health Metrics Endpoint (`GET /v1/admin/metrics`) + +Authenticated endpoint providing detailed protocol counters, cache status, active SSE connections, and indexer state. + +* **URL**: `GET /v1/admin/metrics` +* **Authentication**: Requires Admin JWT (`requireAdmin` middleware). +* **Caching**: Cached in Redis with key `admin:metrics` for 60 seconds (`X-Cache: HIT|MISS` header returned). +* **Indexer & Realtime Payload Fields**: + ```json + { + "indexer": { + "lastLedger": 1234567, + "lagSeconds": 5, + "lastUpdated": "2026-07-27T01:15:00.000Z" + }, + "sse": { + "activeConnections": 42 + }, + "events": { + "last24h": 1250 + } + } + ``` + * `indexer.lastLedger`: Last processed Stellar ledger sequence (`number`, defaults to 0). + * `indexer.lagSeconds`: Seconds elapsed since last indexer state update (`number | null`). + * `indexer.lastUpdated`: ISO timestamp of last indexer update (`string | null`). + * `sse.activeConnections`: Current number of active client SSE connections. + * `events.last24h`: Count of indexed stream events in the past 24 hours. + +--- + +### Indexer Environment Variables + +The indexer worker behavior is controlled by environment variables configured in `backend/src/workers/soroban-event-worker.ts`. + +| Environment Variable | Description | Default Value | Behavior when Disabled / Unset | +|----------------------|-------------|---------------|--------------------------------| +| `STREAM_CONTRACT_ID` | Contract address of the streaming smart contract on Stellar. | `""` (empty string) | **Disables indexer worker**. On startup, `sorobanEventWorker.start()` logs `[SorobanWorker] STREAM_CONTRACT_ID is not set — event indexing disabled.` and exits gracefully without starting the polling loop. `/health` reports `indexerEnabled: false` and skips 503 lag checks. | +| `SOROBAN_RPC_URL` | Endpoint URL for the Soroban RPC node. | `"https://soroban-testnet.stellar.org"` | Uses default public testnet RPC URL. | +| `INDEXER_POLL_INTERVAL_MS` | Polling interval in milliseconds between event fetch cycles. | `"5000"` (5 seconds) | Uses default 5000 ms interval. | +| `INDEXER_START_LEDGER` | Starting Stellar ledger sequence number for cold starts when no `IndexerState` record exists in the database. | `"0"` | Starts indexing from ledger 0 on initial setup. | + +--- + +### Admin Operations: Reset vs. Replay + +The backend provides two admin endpoints for managing indexer positioning and re-processing events. Both require Admin JWT authentication (`requireAdmin`). + +#### 1. Reset Indexer (`POST /v1/admin/indexer/reset`) + +* **Route**: `POST /v1/admin/indexer/reset` +* **Request Body**: + ```json + { + "ledger": 1234500 + } + ``` +* **Handler Behavior**: + * Calls `resetIndexer(toLedger)` in `backend/src/services/indexerService.ts:30`. + * Upserts the singleton row in `prisma.indexerState` (`id: "singleton"`), setting `lastLedger` to `toLedger` and `lastCursor` to `null`. + * **Does NOT immediately trigger a poll**. The worker will pick up from `toLedger` on its next scheduled interval (`INDEXER_POLL_INTERVAL_MS`). + +#### 2. Replay Indexer (`POST /v1/admin/indexer/replay`) + +* **Route**: `POST /v1/admin/indexer/replay?from_ledger=1234500` +* **Query Parameter**: `from_ledger` (non-negative integer). +* **Response**: `202 Accepted` (`{ "ok": true, "replayingFrom": 1234500 }`). +* **Handler Behavior**: + * Calls `replayFromLedger(fromLedger)` in `backend/src/services/indexerService.ts:49`. + * Calls `resetIndexer(fromLedger)` to update `lastLedger` and clear `lastCursor`. + * **Immediately triggers an out-of-band poll cycle** via `sorobanEventWorker.triggerPoll()` without waiting for the next polling interval timer. + +#### Operational Comparison + +| Feature / Scenario | Reset (`POST /v1/admin/indexer/reset`) | Replay (`POST /v1/admin/indexer/replay`) | +|--------------------|------------------------------------------|-------------------------------------------| +| **Execution** | Passive: Updates DB pointer only. | Active: Updates DB pointer **and** triggers immediate poll batch. | +| **Use Case** | Reconfiguring start ledger prior to scheduled maintenance or service restart. | Recovering from RPC outages or backfilling missed ledgers immediately. | +| **Parameters** | JSON body: `{ "ledger": }` | Query param: `?from_ledger=` | + +#### Event Deduplication & Idempotency Guarantee + +Replaying events is safe against duplicate event creation in the database: + +* **Database Constraint**: Grounded in `backend/prisma/schema.prisma:89`, the `StreamEvent` model enforces a unique compound index: + ```prisma + @@unique([transactionHash, eventType]) + ``` +* **Worker Execution**: In `backend/src/workers/soroban-event-worker.ts`, event handlers execute `prisma.streamEvent.upsert(...)` matching on `{ transactionHash_eventType: { transactionHash, eventType } }`. If an event for that transaction and type has already been recorded, the insertion is skipped. +* **Operational Note / Caveat**: As documented in `backend/src/services/indexerService.ts:44-47`, deduplication applies to `StreamEvent` log rows. Stream entity state mutations (such as `Stream.withdrawnAmount` in `handleTokensWithdrawn`) are currently updated unconditionally upon reprocessing events; full state mutation idempotency is tracked under issue #808. + +--- + +### RPC Outage Recovery Walkthrough + +When the Soroban RPC node experiences downtime or network disruption, the indexer will pause event ingestion, causing `indexerLag` on `/health` to increase beyond 60 seconds and report HTTP 503. Follow these steps to restore service: + +1. **Identify the Last Processed Ledger**: + Query the metrics endpoint to obtain the last processed ledger before the outage: + ```bash + curl -H "Authorization: Bearer " http://localhost:3001/v1/admin/metrics + ``` + Note `indexer.lastLedger` from the JSON response. + +2. **Verify RPC Connectivity**: + Ensure `SOROBAN_RPC_URL` is reachable and serving valid Stellar ledger data. + +3. **Trigger Event Replay**: + Call the replay endpoint passing the last known valid ledger sequence: + ```bash + curl -X POST -H "Authorization: Bearer " \ + "http://localhost:3001/v1/admin/indexer/replay?from_ledger=" + ``` + +4. **Monitor Catch-Up Progress**: + Continuously monitor progress via `GET /v1/admin/metrics`: + * Observe `indexer.lastLedger` incrementing toward the current network ledger. + * Observe `indexer.lagSeconds` decreasing. + +5. **Verify Health Restoration**: + Confirm `/health` returns HTTP `200 OK` with `status: "ok"` and `indexerLag <= 60`: + ```bash + curl http://localhost:3001/health + ``` + diff --git a/backend/package.json b/backend/package.json index 42c27d07..c94c733a 100644 --- a/backend/package.json +++ b/backend/package.json @@ -6,6 +6,7 @@ "main": "index.js", "scripts": { "prebuild": "prisma generate", + "pretest": "prisma generate", "test": "vitest run", "dev": "nodemon", "build": "tsc",