fix: resolve database write collisions on StreamEvent and fix schema/syntax errors

.github/workflows/ci.yml

@@ -96,7 +96,6 @@ jobs:
       - name: Run Backend Tests
         run: |
           ls -la src/generated/prisma
-          npm install @vitest/coverage-v8@2.1.9 --no-save
           npx vitest run --coverage --reporter=basic
         working-directory: backend
         env:

backend/package.json

@@ -45,7 +45,7 @@
     "@types/supertest": "^6.0.3",
     "@types/swagger-jsdoc": "^6.0.4",
     "@types/swagger-ui-express": "^4.1.6",
-    "@vitest/coverage-v8": "^2.1.8",
+    "@vitest/coverage-v8": "^3.2.4",
     "eventsource": "^2.0.2",
     "nodemon": "^3.1.11",
     "prisma": "^7.4.1",

backend/prisma/schema.prisma

@@ -87,12 +87,7 @@ model StreamEvent {
   @@index([transactionHash])
   @@index([createdAt])
   @@index([streamId, createdAt])
-  @@unique([transactionHash, eventType])
-}
 
-// IndexerState model - tracks indexer cursor for resumable event processing
-model IndexerState {
-  id         String   @id @default("singleton")
-  lastLedger Int      @default(0)
-  updatedAt  DateTime @updatedAt
 }
+
+

backend/src/controllers/stream.controller.ts

@@ -815,6 +815,36 @@ export const pauseStream = async (req: Request, res: Response) => {
         parsedStreamId,
       );
 
+      // Update the database to mark stream as paused
+      const now = Math.floor(Date.now() / 1000);
+      const updatedStream = await prisma.stream.update({
+        where: { streamId: parsedStreamId },
+        data: {
+          isPaused: true,
+          pausedAt: now,
+          lastUpdateTime: now,
+        },
+      });
+
+      // Create/upsert a PAUSED event
+      await prisma.streamEvent.upsert({
+        where: {
+          transactionHash_eventType: {
+            transactionHash: result.txHash,
+            eventType: 'PAUSED',
+          },
+        },
+        create: {
+          streamId: parsedStreamId,
+          eventType: 'PAUSED',
+          transactionHash: result.txHash,
+          ledgerSequence: 0,
+          timestamp: now,
+          metadata: JSON.stringify({ pausedBy: authReq.user.publicKey }),
+        },
+        update: {},
+      });
+
       logger.info(
         `Stream ${parsedStreamId} pause simulated by ${authReq.user.publicKey}`,
       );
@@ -823,7 +853,7 @@ export const pauseStream = async (req: Request, res: Response) => {
         success: true,
         streamId: parsedStreamId,
         txHash: result.txHash,
-        stream,
+        stream: updatedStream,
       });
     } catch (sorobanError) {
       logger.error(
@@ -899,6 +929,44 @@ export const resumeStream = async (req: Request, res: Response) => {
         parsedStreamId,
       );
 
+      // Calculate pause duration and update the database
+      const now = Math.floor(Date.now() / 1000);
+      const pausedAt = stream.pausedAt ?? now;
+      const pauseDuration = Math.max(0, now - pausedAt);
+      const totalPausedDuration = (stream.totalPausedDuration ?? 0) + pauseDuration;
+
+      const updatedStream = await prisma.stream.update({
+        where: { streamId: parsedStreamId },
+        data: {
+          isPaused: false,
+          pausedAt: null,
+          totalPausedDuration,
+          lastUpdateTime: now,
+        },
+      });
+
+      // Create/upsert a RESUMED event
+      await prisma.streamEvent.upsert({
+        where: {
+          transactionHash_eventType: {
+            transactionHash: result.txHash,
+            eventType: 'RESUMED',
+          },
+        },
+        create: {
+          streamId: parsedStreamId,
+          eventType: 'RESUMED',
+          transactionHash: result.txHash,
+          ledgerSequence: 0,
+          timestamp: now,
+          metadata: JSON.stringify({ 
+            resumedBy: authReq.user.publicKey,
+            pauseDuration,
+          }),
+        },
+        update: {},
+      });
+
       logger.info(
         `Stream ${parsedStreamId} resume simulated by ${authReq.user.publicKey}`,
       );
@@ -907,7 +975,7 @@ export const resumeStream = async (req: Request, res: Response) => {
         success: true,
         streamId: parsedStreamId,
         txHash: result.txHash,
-        stream,
+        stream: updatedStream,
       });
     } catch (sorobanError) {
       logger.error(

backend/src/routes/v1/streams/withdraw.ts

@@ -122,9 +122,15 @@ export const withdrawHandler = async (req: AuthenticatedRequest, res: Response)
         },
       });
 
-      // Create a WITHDRAWN event
-      await prisma.streamEvent.create({
-        data: {
+      // Create/upsert a WITHDRAWN event
+      await prisma.streamEvent.upsert({
+        where: {
+          transactionHash_eventType: {
+            transactionHash: result.txHash,
+            eventType: 'WITHDRAWN',
+          },
+        },
+        create: {
           streamId: parsedStreamId,
           eventType: 'WITHDRAWN',
           amount: claimable.claimableAmount,
@@ -133,6 +139,7 @@ export const withdrawHandler = async (req: AuthenticatedRequest, res: Response)
           timestamp: now,
           metadata: JSON.stringify({ withdrawnBy: req.user.publicKey }),
         },
+        update: {},
       });
 
       logger.info(`Stream ${parsedStreamId} withdrawn by ${req.user.publicKey}`);

backend/src/services/sse.service.ts

@@ -1,6 +1,6 @@
 import type { Response } from 'express';
 import logger from '../logger.js';
-import { isRedisAvailable, getPublisher, getSubscriber } from '../lib/redis.js';
+import { isRedisAvailable, getSubscriber } from '../lib/redis.js';
 
 const HEARTBEAT_INTERVAL_MS = 30_000;
 const MAX_WRITABLE_BUFFER = 64 * 1024;
@@ -16,7 +16,13 @@ export class SSEService {
   private clients: Map<string, SSEClient> = new Map();
   private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
   private slowClientsDropped = 0;
+  private shuttingDown: boolean = false;
+  private ipConnectionCounts: Map<string, number> = new Map();
+  private ipPeakConnections: Map<string, number> = new Map();
+  private maxConnectionsPerIp: number = 100;
+  private globalMaxConnections: number = 10000;
 
+  addClient(clientId: string, res: Response, subscriptions: string[] = [], ip = 'unknown'): void {
     const client: SSEClient = {
       id: clientId,
       res,
@@ -25,17 +31,113 @@ export class SSEService {
     };
 
     this.clients.set(clientId, client);
+
+    // Track per-IP connection counts
+    const ipCount = (this.ipConnectionCounts.get(ip) || 0) + 1;
+    this.ipConnectionCounts.set(ip, ipCount);
+    const peak = this.ipPeakConnections.get(ip) || 0;
+    if (ipCount > peak) {
+      this.ipPeakConnections.set(ip, ipCount);
+    }
+
     logger.info(
       `[SSEService] Connection opened: ${clientId}, ip: ${ip}, subscriptions: ${subscriptions.join(', ')}`
     );
 
     res.on('close', () => {
       this.removeClient(clientId);
+      // Decrement per-IP count
+      const currentCount = this.ipConnectionCounts.get(ip) || 1;
+      if (currentCount <= 1) {
+        this.ipConnectionCounts.delete(ip);
+      } else {
+        this.ipConnectionCounts.set(ip, currentCount - 1);
+      }
     });
 
     this.ensureHeartbeat();
   }
 
+  isShuttingDown(): boolean {
+    return this.shuttingDown;
+  }
+
+  setShuttingDown(value: boolean): void {
+    this.shuttingDown = value;
+  }
+
+  checkCapacity(ip: string): { allowed: boolean; status?: number; message?: string; retryAfterSeconds?: number } {
+    if (this.shuttingDown) {
+      return { allowed: false, status: 503, message: 'Server is shutting down' };
+    }
+
+    if (this.clients.size >= this.globalMaxConnections) {
+      return { allowed: false, status: 503, message: 'Server at capacity', retryAfterSeconds: 30 };
+    }
+
+    const ipCount = this.ipConnectionCounts.get(ip) || 0;
+    if (ipCount >= this.maxConnectionsPerIp) {
+      return { allowed: false, status: 429, message: 'Too many connections from this IP', retryAfterSeconds: 60 };
+    }
+
+    return { allowed: true };
+  }
+
+  initRedisSubscription(): void {
+    if (!isRedisAvailable()) {
+      return;
+    }
+
+    const subscriber = getSubscriber();
+    if (!subscriber) {
+      return;
+    }
+
+    subscriber.subscribe('sse-broadcast', (err) => {
+      if (err) {
+        logger.error('Failed to subscribe to Redis SSE channel', err);
+      }
+    });
+
+    subscriber.on('message', (_channel: string, message: string) => {
+      try {
+        const data = JSON.parse(message);
+        if (data.type === 'broadcast') {
+          this.broadcast(data.event, data.payload);
+        } else if (data.type === 'broadcastToStream') {
+          this.broadcastToStream(data.streamId, data.event, data.payload);
+        } else if (data.type === 'broadcastToUser') {
+          this.broadcastToUser(data.publicKey, data.event, data.payload);
+        }
+      } catch (error) {
+        logger.error('Failed to parse Redis SSE message', error);
+      }
+    });
+  }
+
+  sendReconnectToAll(): void {
+    this.shuttingDown = true;
+    this.broadcast('reconnect', { timestamp: Date.now() });
+  }
+
+  broadcastToAdmin(event: string, data: unknown): void {
+    this.broadcast(event, data, (client) =>
+      client.subscriptions.has('admin') || client.subscriptions.has('*')
+    );
+  }
+
+  getActiveIpCount(): number {
+    return this.ipConnectionCounts.size;
+  }
+
+  getPerIpPeakConnections(): Map<string, number> {
+    return this.ipPeakConnections;
+  }
+
+  getMaxConnections(): number {
+    return this.globalMaxConnections;
+  }
+
   sendHeartbeat(): void {
     const message = ': heartbeat\n\n';