Merge branch 'main' into chore/1081-fix-nodemon-test-ignore

Author: ogazboiz

backend/prisma/migrations/20260630120000_add_stream_event_stream_id_timestamp_index/migration.sql

@@ -0,0 +1,8 @@
+-- Replace the unused (streamId, createdAt) composite with (streamId, timestamp),
+-- which matches streamId-scoped event listings that ORDER BY timestamp.
+
+-- CreateIndex
+CREATE INDEX IF NOT EXISTS "StreamEvent_streamId_timestamp_idx" ON "StreamEvent"("streamId", "timestamp");
+
+-- DropIndex
+DROP INDEX IF EXISTS "StreamEvent_streamId_createdAt_idx";

backend/prisma/schema.prisma

@@ -86,5 +86,6 @@ model StreamEvent {
   @@index([timestamp])
   @@index([transactionHash])
   @@index([createdAt])
-  @@index([streamId, createdAt])
+  @@index([streamId, timestamp])
+  @@unique([transactionHash, eventType])
 }

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

@@ -1,10 +1,12 @@
+import { randomUUID } from 'crypto';
 import type { Response } from 'express';
-import logger from '../logger.js';
+import logger, { requestContext } from '../logger.js';
 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 +15,7 @@ interface SSEClient {
   subscriptions: Set<string>;
   paused: boolean;
   ip: string;
+  userId?: string;
 }
 
 interface SSECapacityCheckResult {
@@ -27,8 +30,17 @@ 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;
+
+  /**
+   * Stable id attached to every log line emitted by the heartbeat
+   * setInterval callback, since it fires outside of any HTTP request and
+   * would otherwise have no requestContext (and thus no correlation id).
+   */
+  private readonly heartbeatWorkerId = `sse-heartbeat:${randomUUID()}`;
 
   private readonly maxConnections: number = (() => {
     const parsed = Number.parseInt(process.env.MAX_SSE_CONNECTIONS ?? '10000', 10);
@@ -61,7 +73,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 +92,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 +234,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);
@@ -207,7 +259,9 @@ export class SSEService {
     }
 
     this.heartbeatTimer = setInterval(() => {
-      this.sendHeartbeat();
+      requestContext.run({ requestId: this.heartbeatWorkerId }, () => {
+        this.sendHeartbeat();
+      });
     }, HEARTBEAT_INTERVAL_MS);
   }
 
@@ -235,6 +289,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/src/workers/soroban-event-worker.ts

@@ -1,8 +1,9 @@
+import { randomUUID } from "crypto";
 import { rpc, xdr, StrKey } from "@stellar/stellar-sdk";
 import { prisma } from "../lib/prisma.js";
 import { INDEXER_STATE_ID, ensureIndexerState } from "../lib/indexer-state.js";
 import { sseService } from "../services/sse.service.js";
-import logger from "../logger.js";
+import logger, { requestContext } from "../logger.js";
 import { Prisma } from "../generated/prisma/index.js";
 import "../lib/stream-id.js";
 
@@ -112,6 +113,13 @@ export class SorobanEventWorker {
   /** Recent attempt outcomes for sliding-window spike detection. */
   private recentOutcomes: { ok: boolean; at: number }[] = [];
 
+  /**
+   * Stable id attached to every log line emitted by the background poll
+   * loop, since these callbacks fire outside of any HTTP request and would
+   * otherwise have no requestContext (and thus no correlation id) at all.
+   */
+  private readonly workerId = `soroban-worker:${randomUUID()}`;
+
   constructor() {
     const rpcUrl =
       process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";

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/integration/streams.test.ts

@@ -231,6 +231,17 @@ describe('GET /v1/streams/:id/events — pagination and eventType filter', () =>
     expect(res.body).toHaveProperty('hasMore');
     expect(Array.isArray(res.body.data)).toBe(true);
     expect(res.body.hasMore).toBe(false);
+
+    const callArgs = mockPrisma.streamEvent.findMany.mock.calls[0]![0] as {
+      where: { streamId: number };
+      orderBy: { timestamp: string };
+      take: number;
+      skip: number;
+    };
+    expect(callArgs.where.streamId).toBe(1);
+    expect(callArgs.orderBy).toEqual({ timestamp: 'desc' });
+    expect(callArgs.take).toBe(10);
+    expect(callArgs.skip).toBe(0);
   });
 
   it('enforces default limit of 50', async () => {