Feature/fee collected event

backend/src/controllers/stream.controller.ts

@@ -1,4 +1,4 @@
 import type { Request, Response } from 'express';
 import { prisma } from '../lib/prisma.js';
 import logger from '../logger.js';
 import { claimableAmountService } from '../services/claimable.service.js';
@@ -374,7 +374,7 @@
 /**
  * Get user-level stream summary used by dashboard/profile cards.
  */
-export const getUserStreamSummary = async (req: Request, res: Response) => {
+export const getUserStreamSummary = async (req: Request<{ address: string }>, res: Response) => {
   try {
     const address = Array.isArray(req.params.address) ? req.params.address[0] : (req.params.address ?? '').trim();
     if (!address) {

backend/src/services/sse.service.ts

@@ -20,7 +20,7 @@ interface SSECapacityCheckResult {
   message?: string;
 }
 
-class SSEService {
+export class SSEService {
   private clients: Map<string, SSEClient> = new Map();
   private readonly ipConnectionCounts: Map<string, number> = new Map();
   private shuttingDown = false;
@@ -196,6 +196,13 @@ class SSEService {
     }
   }
 
+  broadcastToAdmin(event: string, data: unknown): void {
+    const adminKey = process.env.ADMIN_PUBLIC_KEY;
+    if (adminKey) {
+      this.broadcastToUser(adminKey, event, data);
+    }
+  }
+
   private _localBroadcastToStream(streamId: string, event: string, data: unknown): void {
     this.broadcast(event, data, (client) =>
       client.subscriptions.has(streamId) || client.subscriptions.has('*')
@@ -225,5 +232,4 @@ class SSEService {
   }
 }
 
-export { SSEService };
 export const sseService = new SSEService();

backend/src/workers/soroban-event-worker.ts

@@ -193,7 +193,18 @@
     let lastCursor: string | null = state.lastCursor;
     let lastLedger: number = state.lastLedger;
 
-    for (const event of response.events) {
+    // Sort events so that 'stream_created' events are processed first in the batch.
+    // This ensures that subsequent events (like 'fee_collected') that depend on
+    // the stream existing in the DB can find it.
+    const sortedEvents = [...response.events].sort((a, b) => {
+      const aType = a.topic[0] ? decodeSymbol(a.topic[0]) : '';
+      const bType = b.topic[0] ? decodeSymbol(b.topic[0]) : '';
+      if (aType === 'stream_created' && bType !== 'stream_created') return -1;
+      if (bType === 'stream_created' && aType !== 'stream_created') return 1;
+      return 0;
+    });
+
+    for (const event of sortedEvents) {
       // Only process events from successful contract calls.
       if (!event.inSuccessfulContractCall) continue;
 
@@ -391,7 +402,7 @@
         select: { ratePerSecond: true, startTime: true, totalPausedDuration: true }
       });
 
       const durationSeconds = Number(BigInt(newDepositedAmount) / BigInt(stream.ratePerSecond));
       const newEndTime = stream.startTime + durationSeconds + stream.totalPausedDuration;
 
       await tx.stream.update({
@@ -607,15 +618,15 @@
     });
 
     // Broadcast to admin channel for treasury reporting
-    sseService.broadcast('stream.fee_collected', {
+    sseService.broadcastToAdmin('stream.fee_collected', {
       streamId,
       treasury,
       feeAmount,
       token,
       transactionHash: event.txHash,
       ledger: event.ledger,
       timestamp,
-    }, (client) => client.subscriptions.has('admin') || client.subscriptions.has('*'));
+    });
   }
 
   private async handleStreamPaused(
@@ -626,7 +637,7 @@
     const body = decodeMap(event.value);
 
     if (!body['sender'] || !body['paused_at']) {
       throw new Error(`StreamPaused #${streamId}: missing body fields`);
     }
 
     const sender = decodeAddress(body['sender']);