feat: health check breakdown, per-user SSE cap, claimable invalidation test, api-types sync doc

backend/src/controllers/sse.controller.ts

@@ -31,7 +31,8 @@ export const subscribe = async (req: Request, res: Response) => {
 
   try {
     const sourceIp = getClientIp(req);
-    const capacity = sseService.checkCapacity(sourceIp);
+    const authUserId = (req as AuthenticatedRequest).user?.publicKey;
+    const capacity = sseService.checkCapacity(sourceIp, authUserId);
     if (!capacity.allowed) {
       if (capacity.retryAfterSeconds) {
         res.setHeader('Retry-After', String(capacity.retryAfterSeconds));
@@ -87,7 +88,7 @@ export const subscribe = async (req: Request, res: Response) => {
     const requestId = requestContext.getStore()?.requestId;
     res.write(`data: ${JSON.stringify({ type: 'connected', clientId, requestId })}\n\n`);
 
-    sseService.addClient(clientId, res, subscriptions, sourceIp);
+    sseService.addClient(clientId, res, subscriptions, sourceIp, publicKey);
     return;
   } catch (error: unknown) {
     if (error instanceof z.ZodError) {

backend/src/routes/health.routes.ts

@@ -65,6 +65,39 @@ const router = Router();
  *                   type: number
  *                   description: Server uptime in seconds
  *                   example: 3600
+ *                 checks:
+ *                   type: object
+ *                   description: Per-subsystem status breakdown, so callers can tell "DB unreachable" apart from "indexer lagging" instead of inferring it from the top-level status alone.
+ *                   properties:
+ *                     database:
+ *                       type: object
+ *                       properties:
+ *                         status:
+ *                           type: string
+ *                           enum: [ok, down]
+ *                     indexer:
+ *                       type: object
+ *                       properties:
+ *                         status:
+ *                           type: string
+ *                           enum: [ok, degraded, disabled]
+ *                         enabled:
+ *                           type: boolean
+ *                         lagSeconds:
+ *                           type: integer
+ *                           nullable: true
+ *                     redis:
+ *                       type: object
+ *                       properties:
+ *                         status:
+ *                           type: string
+ *                           enum: [ok, unavailable, not_configured]
+ *                     sorobanRpc:
+ *                       type: object
+ *                       properties:
+ *                         status:
+ *                           type: string
+ *                           enum: [ok, down]
  *       503:
  *         description: Service is degraded or unhealthy
  */
@@ -104,6 +137,15 @@ router.get('/', async (_req: Request, res: Response) => {
     dbStatus === 'connected' && !indexerLagDegraded && !indexerFailureDegraded;
   const status = isHealthy ? 'ok' : 'degraded';
 
+  // Redis is optional (single-instance SSE mode falls back gracefully when it's
+  // absent), so its status never affects the top-level `isHealthy` verdict.
+  const redisConfigured = !!process.env.REDIS_URL;
+  const redisStatus = !redisConfigured ? 'not_configured' : isRedisAvailable() ? 'ok' : 'unavailable';
+
+  // Soroban RPC reachability is reported for observability only — it does not
+  // gate liveness, since a transient RPC blip shouldn't take the service down.
+  const sorobanRpcOk = await checkRpcHealth();
+
   res.status(isHealthy ? 200 : 503).json({
     status,
     db: dbStatus,
@@ -114,6 +156,22 @@ router.get('/', async (_req: Request, res: Response) => {
     lastErrorAt: eventCounters.lastErrorAt,
     indexerDegraded: eventCounters.degraded,
     uptime: process.uptime(),
+    checks: {
+      database: {
+        status: dbStatus === 'connected' ? 'ok' : 'down',
+      },
+      indexer: {
+        status: !indexerEnabled ? 'disabled' : indexerDegraded ? 'degraded' : 'ok',
+        enabled: indexerEnabled,
+        lagSeconds: indexerLag === -1 ? null : indexerLag,
+      },
+      redis: {
+        status: redisStatus,
+      },
+      sorobanRpc: {
+        status: sorobanRpcOk ? 'ok' : 'down',
+      },
+    },
   });
 });
 

backend/src/services/sorobanService.ts

@@ -126,6 +126,20 @@ function getServer(): rpc.Server {
   return _server;
 }
 
+/**
+ * Lightweight connectivity check used by the /health endpoint.
+ * Calls the RPC server's getHealth() with a bounded timeout so a slow or
+ * unreachable Soroban RPC endpoint can't hang the health check.
+ */
+export async function checkRpcHealth(timeoutMs = 3_000): Promise<boolean> {
+  try {
+    await withRpcTimeout('soroban rpc health check', () => getServer().getHealth(), timeoutMs);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
 export function setServer(server: rpc.Server): void {
   _server = server;
 }

backend/src/services/sse.service.ts

@@ -5,6 +5,7 @@ import { isRedisAvailable, getPublisher, getSubscriber } from '../lib/redis.js';
 const HEARTBEAT_INTERVAL_MS = 30_000;
 const MAX_WRITABLE_BUFFER = 64 * 1024;
 const MAX_CONNECTIONS_PER_IP = 5;
+const MAX_CONNECTIONS_PER_USER = 10;
 const RETRY_AFTER_SECONDS = 60;
 
 interface SSEClient {
@@ -13,6 +14,7 @@ interface SSEClient {
   subscriptions: Set<string>;
   paused: boolean;
   ip: string;
+  userId?: string;
 }
 
 interface SSECapacityCheckResult {
@@ -27,8 +29,10 @@ export class SSEService {
   private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
   private slowClientsDropped = 0;
   private readonly ipConnectionCounts: Map<string, number> = new Map();
+  private readonly userConnectionCounts: Map<string, number> = new Map();
   private shuttingDown = false;
   private perIpPeakConnections = 0;
+  private perUserPeakConnections = 0;
 
   private readonly maxConnections: number = (() => {
     const parsed = Number.parseInt(process.env.MAX_SSE_CONNECTIONS ?? '10000', 10);
@@ -61,7 +65,7 @@ export class SSEService {
     logger.info('[SSEService] Redis pub/sub subscription active.');
   }
 
-  checkCapacity(ip: string): SSECapacityCheckResult {
+  checkCapacity(ip: string, userId?: string): SSECapacityCheckResult {
     if (this.clients.size >= this.maxConnections) {
       return {
         allowed: false,
@@ -80,25 +84,53 @@ export class SSEService {
       };
     }
 
+    // Independent of the per-IP cap: bounds how many concurrent SSE
+    // subscriptions a single authenticated user can hold regardless of which
+    // IP(s) they connect from (e.g. multiple tabs/devices behind different NATs).
+    if (userId) {
+      const currentUserConnections = this.userConnectionCounts.get(userId) ?? 0;
+      if (currentUserConnections >= MAX_CONNECTIONS_PER_USER) {
+        return {
+          allowed: false,
+          status: 429,
+          retryAfterSeconds: RETRY_AFTER_SECONDS,
+          message: `Too many concurrent SSE connections for this user. Max ${MAX_CONNECTIONS_PER_USER}.`,
+        };
+      }
+    }
+
     return { allowed: true };
   }
 
-  addClient(clientId: string, res: Response, subscriptions: string[] = [], ip = 'unknown'): void {
+  addClient(
+    clientId: string,
+    res: Response,
+    subscriptions: string[] = [],
+    ip = 'unknown',
+    userId?: string,
+  ): void {
     const nextIpCount = (this.ipConnectionCounts.get(ip) ?? 0) + 1;
     this.ipConnectionCounts.set(ip, nextIpCount);
     this.perIpPeakConnections = Math.max(this.perIpPeakConnections, nextIpCount);
 
+    if (userId) {
+      const nextUserCount = (this.userConnectionCounts.get(userId) ?? 0) + 1;
+      this.userConnectionCounts.set(userId, nextUserCount);
+      this.perUserPeakConnections = Math.max(this.perUserPeakConnections, nextUserCount);
+    }
+
     const client: SSEClient = {
       id: clientId,
       res,
       subscriptions: new Set(subscriptions),
       paused: false,
       ip,
+      ...(userId !== undefined && { userId }),
     };
 
     this.clients.set(clientId, client);
     logger.info(
-      `[SSEService] Connection opened: ${clientId}, ip: ${ip}, subscriptions: ${subscriptions.join(', ')}`
+      `[SSEService] Connection opened: ${clientId}, ip: ${ip}, userId: ${userId ?? 'n/a'}, subscriptions: ${subscriptions.join(', ')}`
     );
 
     res.on('close', () => {
@@ -194,6 +226,18 @@ export class SSEService {
     return this.ipConnectionCounts.size;
   }
 
+  getPerUserPeakConnections(): number {
+    return this.perUserPeakConnections;
+  }
+
+  getActiveUserCount(): number {
+    return this.userConnectionCounts.size;
+  }
+
+  getUserConnectionCount(userId: string): number {
+    return this.userConnectionCounts.get(userId) ?? 0;
+  }
+
   stopHeartbeat(): void {
     if (this.heartbeatTimer) {
       clearInterval(this.heartbeatTimer);
@@ -235,6 +279,15 @@ export class SSEService {
       this.ipConnectionCounts.set(client.ip, currentIpCount - 1);
     }
 
+    if (client.userId) {
+      const currentUserCount = this.userConnectionCounts.get(client.userId) ?? 0;
+      if (currentUserCount <= 1) {
+        this.userConnectionCounts.delete(client.userId);
+      } else {
+        this.userConnectionCounts.set(client.userId, currentUserCount - 1);
+      }
+    }
+
     try {
       if (!client.res.writableEnded) {
         client.res.end();

backend/tests/claimable.service.test.ts

@@ -131,6 +131,57 @@ describe('ClaimableAmountService', () => {
     expect(third.cached).toBe(false);
   });
 
+  it('reflects an indexed withdrawal immediately, without waiting for the cache TTL', () => {
+    // The cache key is derived from getStateFingerprint(), which folds in
+    // withdrawnAmount and lastUpdateTime (see claimable.service.ts). When the
+    // indexer processes a TokensWithdrawn event it updates those fields on the
+    // Stream row, so the *next* read (which is always given the freshly
+    // reloaded stream from the DB, see stream.controller.ts / withdraw.ts)
+    // naturally lands on a different cache key and can never return the
+    // pre-withdrawal cached value — invalidation falls out of the key design
+    // rather than needing an explicit "on withdrawal, delete this key" hook.
+    vi.setSystemTime(50_000);
+    const service = new ClaimableAmountService({
+      cacheTtlMs: 60_000, // deliberately long TTL to prove this isn't just a TTL expiry
+    });
+
+    const preWithdrawalState = makeStreamState({
+      streamId: 7,
+      ratePerSecond: '10',
+      depositedAmount: '1000',
+      withdrawnAmount: '0',
+      lastUpdateTime: 0,
+    });
+
+    // Prime the cache with the pre-withdrawal state.
+    const primed = service.getClaimableAmount(preWithdrawalState, 40);
+    expect(primed.cached).toBe(false);
+    expect(primed.claimableAmount).toBe('400'); // 40s * 10/s
+
+    // A repeated read with the identical state still hits the cache.
+    const repeated = service.getClaimableAmount(preWithdrawalState, 40);
+    expect(repeated.cached).toBe(true);
+
+    // Simulate the indexer processing a TokensWithdrawn event: withdrawnAmount
+    // and lastUpdateTime are advanced on the stream row, exactly as
+    // handleTokensWithdrawn does in soroban-event-worker.ts.
+    const postWithdrawalState = makeStreamState({
+      streamId: 7,
+      ratePerSecond: '10',
+      depositedAmount: '1000',
+      withdrawnAmount: '400',
+      lastUpdateTime: 40,
+    });
+
+    // Well within the 60s TTL, so this only passes if the state change (not
+    // TTL expiry) is what causes the fresh calculation.
+    vi.advanceTimersByTime(1_000);
+    const afterWithdrawal = service.getClaimableAmount(postWithdrawalState, 40);
+
+    expect(afterWithdrawal.cached).toBe(false);
+    expect(afterWithdrawal.claimableAmount).toBe('0'); // fully withdrawn as of lastUpdateTime=40
+  });
+
   it('caps multiplication overflow at the remaining balance', () => {
     const i128Max = ((1n << 127n) - 1n).toString();
     vi.setSystemTime(1_000_000);

backend/tests/sse.service.test.ts

@@ -98,3 +98,55 @@ describe('SSEService backpressure', () => {
     expect(service.getSlowClientsDropped()).toBe(1);
   });
 });
+
+describe('SSEService per-user connection cap', () => {
+  let service: SSEService;
+
+  afterEach(() => {
+    service.stopHeartbeat();
+  });
+
+  it('allows connections from a user up to the per-user cap, independent of IP', () => {
+    service = new SSEService();
+    const userId = 'GUSER123';
+
+    // Spread the connections across different IPs so only the per-user cap
+    // (not the pre-existing per-IP cap) is exercised here.
+    for (let i = 0; i < 10; i += 1) {
+      const capacity = service.checkCapacity(`10.0.0.${i}`, userId);
+      expect(capacity.allowed).toBe(true);
+      service.addClient(`client-${i}`, createMockResponse(), [], `10.0.0.${i}`, userId);
+    }
+
+    expect(service.getUserConnectionCount(userId)).toBe(10);
+
+    // The 11th connection for the same user, from yet another IP, must be rejected.
+    const rejected = service.checkCapacity('10.0.0.99', userId);
+    expect(rejected.allowed).toBe(false);
+    expect(rejected.status).toBe(429);
+    expect(rejected.message).toMatch(/user/i);
+    expect(rejected.retryAfterSeconds).toBeGreaterThan(0);
+  });
+
+  it('does not cap unauthenticated/anonymous checks that omit a userId', () => {
+    service = new SSEService();
+
+    for (let i = 0; i < 10; i += 1) {
+      const capacity = service.checkCapacity(`10.1.0.${i}`);
+      expect(capacity.allowed).toBe(true);
+    }
+  });
+
+  it('releases the per-user slot once a client disconnects', () => {
+    service = new SSEService();
+    const userId = 'GUSER456';
+
+    const res = createMockResponse();
+    service.addClient('client-a', res, [], '10.2.0.1', userId);
+    expect(service.getUserConnectionCount(userId)).toBe(1);
+
+    res.emitter.emit('close');
+
+    expect(service.getUserConnectionCount(userId)).toBe(0);
+  });
+});

docs/api/flowfi.postman_collection.json

@@ -60,7 +60,7 @@
               "code": 200,
               "_postman_previewlanguage": "json",
               "header": [{ "key": "Content-Type", "value": "application/json; charset=utf-8" }],
-              "body": "{\n  \"status\": \"healthy\",\n  \"timestamp\": \"2024-02-21T14:30:00.000Z\",\n  \"uptime\": 3600,\n  \"version\": \"1.0.0\",\n  \"api\": {\n    \"supported\": [\"v1\"],\n    \"default\": \"v1\"\n  }\n}"
+              "body": "{\n  \"status\": \"ok\",\n  \"db\": \"connected\",\n  \"indexerEnabled\": true,\n  \"indexerLag\": 5,\n  \"uptime\": 3600,\n  \"checks\": {\n    \"database\": { \"status\": \"ok\" },\n    \"indexer\": { \"status\": \"ok\", \"enabled\": true, \"lagSeconds\": 5 },\n    \"redis\": { \"status\": \"ok\" },\n    \"sorobanRpc\": { \"status\": \"ok\" }\n  }\n}"
             }
           ]
         },

frontend/.gitignore

@@ -41,3 +41,6 @@ yarn-error.log*
 # typescript
 *.tsbuildinfo
 next-env.d.ts
+
+# manually-triggered API type codegen output (see src/lib/api-types.ts)
+src/lib/api-types.generated.ts

frontend/package.json

@@ -9,7 +9,8 @@
     "lint": "eslint",
     "test": "vitest run",
     "test:watch": "vitest",
-    "test:coverage": "vitest run --coverage"
+    "test:coverage": "vitest run --coverage",
+    "codegen:api-types": "openapi-typescript http://localhost:3001/api-docs.json -o src/lib/api-types.generated.ts"
   },
   "dependencies": {
     "@stellar/freighter-api": "^6.0.1",
@@ -39,6 +40,7 @@
     "eslint-config-next": "^16.2.9",
     "happy-dom": "^20.10.3",
     "jsdom": "^27.0.1",
+    "openapi-typescript": "^7.13.0",
     "tailwindcss": "^4",
     "typescript": "^5",
     "vitest": "^2.1.9"