LabsCrypt · extolkom · Jul 27, 2026 · Jul 27, 2026
diff --git a/backend/SSE_README.md 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
@@ -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": <number> }` | Query param: `?from_ledger=<number>` |
+
+#### 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 <ADMIN_JWT>" 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 <ADMIN_JWT>" \
+     "http://localhost:3001/v1/admin/indexer/replay?from_ledger=<LAST_KNOWN_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
@@ -6,6 +6,7 @@
   "main": "index.js",
   "scripts": {
     "prebuild": "prisma generate",
+    "pretest": "prisma generate",
     "test": "vitest run",
     "dev": "nodemon",
     "build": "tsc",