Merge branch 'main' into #1232

Author: ponmileleke54-dev

.github/workflows/ci.yml

@@ -52,10 +52,6 @@ jobs:
         run: npm run lint
         working-directory: frontend
 
-      - name: Install Rollup Native Binding
-        run: npm install @rollup/rollup-linux-x64-gnu --no-save
-        working-directory: frontend
-
       - name: Run Frontend Tests
         run: npm run test:coverage
         working-directory: frontend
@@ -64,6 +60,10 @@ jobs:
         run: npm run build
         working-directory: frontend
 
+      - name: Check frontend bundle size
+        run: bash ./scripts/check-bundle-size.sh
+        working-directory: frontend
+
   backend:
     name: Backend CI
     runs-on: ubuntu-latest
@@ -121,9 +121,6 @@ jobs:
           cd ..
           git diff --exit-code -- backend/swagger/flowfi.openapi.json frontend/src/lib/api-types.generated.ts
 
-      - name: Install Rollup Native Binding
-        run: npm install @rollup/rollup-linux-x64-gnu --no-save
-
       - name: Run Backend Tests
         run: |
           ls -la src/generated/prisma

.github/workflows/pr-test-gate.yml

@@ -63,11 +63,6 @@ jobs:
         env:
           DATABASE_URL: postgresql://postgres:password@127.0.0.1:5432/flowfi_test
 
-      - name: Install Native Bindings
-        run: |
-          npm install @rollup/rollup-linux-x64-gnu --no-save
-        working-directory: backend
-
       - name: Run backend tests
         run: |
           ls -la src/generated/prisma

backend/docs/SSE_ARCHITECTURE.md

@@ -294,6 +294,9 @@ The indexer worker behavior is controlled by environment variables configured in
 | `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. |
+| `INDEXER_DEAD_LETTER_MAX_RETRIES` | Max failed processing attempts before an event is abandoned to the `IndexerDeadLetterEvent` table and the cursor advances past it. | `"5"` | Uses default of 5 attempts. |
+
+Failed events never freeze the indexer: the cursor always advances past successfully processed events even when an earlier event in the batch failed, and each failing event is recorded (with its raw payload) in the `IndexerDeadLetterEvent` table for manual triage. After `INDEXER_DEAD_LETTER_MAX_RETRIES` attempts the event is abandoned and the cursor advances past it.
 
 ---
 

backend/prisma/migrations/20260830000000_add_indexer_dead_letter_event/migration.sql

@@ -0,0 +1,31 @@
+-- Dead-letter table for Soroban events that failed to process. A single
+-- malformed event must not freeze the indexer: after N failed attempts the
+-- worker abandons the event (recording it here with its raw payload for
+-- manual triage) and advances the cursor past it.
+
+-- CreateTable
+CREATE TABLE "IndexerDeadLetterEvent" (
+    "id" TEXT NOT NULL,
+    "eventId" TEXT NOT NULL,
+    "ledger" INTEGER NOT NULL,
+    "transactionHash" TEXT NOT NULL,
+    "rawPayload" TEXT NOT NULL,
+    "errorMessage" TEXT NOT NULL,
+    "attempts" INTEGER NOT NULL DEFAULT 1,
+    "lastAttemptAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "IndexerDeadLetterEvent_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "IndexerDeadLetterEvent_eventId_key" ON "IndexerDeadLetterEvent"("eventId");
+
+-- CreateIndex
+CREATE INDEX "IndexerDeadLetterEvent_ledger_idx" ON "IndexerDeadLetterEvent"("ledger");
+
+-- CreateIndex
+CREATE INDEX "IndexerDeadLetterEvent_transactionHash_idx" ON "IndexerDeadLetterEvent"("transactionHash");
+
+-- CreateIndex
+CREATE INDEX "IndexerDeadLetterEvent_createdAt_idx" ON "IndexerDeadLetterEvent"("createdAt");

backend/prisma/schema.prisma

@@ -64,6 +64,26 @@ model IndexerState {
   updatedAt   DateTime @updatedAt
 }
 
+// IndexerDeadLetterEvent model - Soroban events that failed to process, kept
+// with their raw payload for manual triage. The worker retries an event at
+// most INDEXER_DEAD_LETTER_MAX_RETRIES times, then abandons it and advances
+// the cursor so a single malformed event can never freeze the indexer.
+model IndexerDeadLetterEvent {
+  id              String   @id @default(uuid())
+  eventId         String   @unique // RPC paging-token id of the event
+  ledger          Int      // Ledger sequence the event was emitted in
+  transactionHash String   // Stellar transaction hash
+  rawPayload      String   // Full raw event JSON for manual triage/replay
+  errorMessage    String   // Last error thrown while processing
+  attempts        Int      @default(1) // Number of failed processing attempts
+  lastAttemptAt   DateTime @default(now())
+  createdAt       DateTime @default(now())
+
+  @@index([ledger])
+  @@index([transactionHash])
+  @@index([createdAt])
+}
+
 // StreamEvent model - indexer events for tracking all on-chain stream activities
 model StreamEvent {
   id              String   @id @default(uuid())

backend/src/controllers/stream.controller.ts

@@ -14,6 +14,7 @@ import {
 } from "../services/sorobanService.js";
 import type { AuthenticatedRequest } from "../types/auth.types.js";
 import { parseStreamId } from "../lib/stream-id.js";
+import { createStreamSchema } from "../validators/stream.validator.js";
 import {
   DEFAULT_EVENTS_PAGE_SIZE,
   MAX_EVENTS_PAGE_SIZE,
@@ -75,37 +76,6 @@ function sumStringI128(values: string[]): string {
   return total.toString();
 }
 
-/**
- * Thrown when a request body field fails presence/format validation. Kept
- * distinct from generic errors so createStream can reliably map it to a 400
- * response instead of falling through to the catch-all 500.
- */
-class StreamValidationError extends Error {
-  constructor(message: string) {
-    super(message);
-    this.name = "StreamValidationError";
-  }
-}
-
-/**
- * Validate presence and integer format of a required i128-style field, then
- * coerce it to a BigInt. Any missing value or conversion failure (SyntaxError
- * from a non-numeric string, TypeError from undefined/null/objects, etc.) is
- * normalized into a StreamValidationError so the caller can map it to 400.
- */
-function parseRequiredBigIntField(fieldName: string, value: unknown): bigint {
-  if (value === undefined || value === null || value === "") {
-    throw new StreamValidationError(`Missing required field: ${fieldName}`);
-  }
-  try {
-    return BigInt(value as bigint | number | string | boolean);
-  } catch {
-    throw new StreamValidationError(
-      `Invalid ${fieldName}: must be a valid integer`,
-    );
-  }
-}
-
 /**
  * Create a new stream (stub for on-chain indexing)
  */
@@ -129,6 +99,8 @@ export const createStream = async (req: Request, res: Response) => {
       return sendApiError(res, 400, 'INVALID_TOKEN_ADDRESS', 'Invalid tokenAddress: must be a non-empty string');
     }
 
+    const { streamId: parsedStreamId, sender, recipient, tokenAddress, ratePerSecond, depositedAmount, startTime: parsedStartTime } = parsed.data;
+
     // Issue #809: the authenticated wallet may only create/modify streams it owns.
     // Without this, any logged-in wallet could POST an arbitrary `sender` and have
     // it persisted, or flip another owner's cancelled stream back to active.

backend/src/controllers/stream/cancel.ts

@@ -88,7 +88,7 @@ export const cancelStreamHandler = async (req: AuthenticatedRequest, res: Respon
       return sendApiError(res, 500, 'INTERNAL_SERVER_ERROR', 'Backend not configured for on-chain calls');
     }
 
-    const txHash = await sorobanService.cancelStream(parsedStreamId, secretKey);
+    const txHash = await sorobanService.cancelStream(parsedStreamId, senderSecret);
 
     // 5. Update DB record status using repository helper
     await streamRepository.updateStatus(parsedStreamId, 'CANCELLED');

backend/src/controllers/user.controller.ts

@@ -7,9 +7,11 @@ import {
 } from "../validators/user.validator.js";
 import type { AuthenticatedRequest } from "../types/auth.types.js";
 import {
-  DEFAULT_EVENTS_PAGE_SIZE,
-  MAX_EVENTS_PAGE_SIZE,
-} from "../routes/v1/events.routes.js";
+  listEventsForWallet,
+  parseEventTypeFilter,
+  resolveEventsOffset,
+  resolveEventsPageSize,
+} from "../repositories/streamEvent.repository.js";
 import * as exportService from "../services/export.service.js";
 import { sendApiError } from "../types/api-error.js";
 
@@ -127,7 +129,17 @@ export const getUser = async (
 };
 
 /**
- * Get user events (history)
+ * Get user events (history) - paginated list of stream events where the
+ * given wallet was either the sender or recipient.
+ *
+ * Query params:
+ *   - type: optional comma-separated list of event types to filter by
+ *     (e.g. "PAUSED,RESUMED"); unknown values are ignored, and a filter
+ *     consisting entirely of unknown values is rejected with 400.
+ *   - limit, offset, page: pagination (see repositories/streamEvent.repository.ts)
+ *   - includeStream: set to "false" to omit the related `stream` object
+ *     from each event (included by default, matching this endpoint's
+ *     historical behavior).
  */
 export const getUserEvents = async (
   req: Request,
@@ -143,47 +155,38 @@ export const getUserEvents = async (
       return sendApiError(res, 400, "INVALID_PUBLIC_KEY", "Invalid Stellar public key format");
     }
 
-    const rawLimit = req.query["limit"];
-    const rawOffset = req.query["offset"];
-
-    const limit = Math.min(
-      rawLimit && typeof rawLimit === "string"
-        ? Number.parseInt(rawLimit, 10) || DEFAULT_EVENTS_PAGE_SIZE
-        : DEFAULT_EVENTS_PAGE_SIZE,
-      MAX_EVENTS_PAGE_SIZE,
-    );
-    const offset =
-      rawOffset && typeof rawOffset === "string"
-        ? Math.max(0, Number.parseInt(rawOffset, 10) || 0)
-        : 0;
-
-    const whereClause = {
-      stream: {
-        OR: [{ sender: publicKey }, { recipient: publicKey }],
-      },
-    };
+    const { requested, types } = parseEventTypeFilter(req.query["type"]);
+    if (requested.length > 0 && types.length === 0) {
+      return res
+        .status(400)
+        .json({ error: "No valid event types in `type` filter" });
+    }
 
-    const [events, total] = await Promise.all([
-      prisma.streamEvent.findMany({
-        where: whereClause,
-        orderBy: { timestamp: "desc" },
-        take: limit,
-        skip: offset,
-        include: {
-          stream: true,
-        },
-      }),
-      prisma.streamEvent.count({ where: whereClause }),
-    ]);
+    const limit = resolveEventsPageSize(req.query["limit"]);
+    const offset = resolveEventsOffset({
+      rawOffset: req.query["offset"],
+      rawPage: req.query["page"],
+      limit,
+    });
 
-    const hasMore = offset + events.length < total;
+    // Preserve this endpoint's historical behavior of always embedding the
+    // related stream, unless the caller opts out.
+    const includeStream = req.query["includeStream"] !== "false";
 
-    return res.status(200).json({
-      data: events,
-      total,
-      hasMore,
+    const result = await listEventsForWallet({
+      address: publicKey,
+      types,
       limit,
       offset,
+      includeStream,
+    });
+
+    return res.status(200).json({
+      data: result.events,
+      total: result.total,
+      hasMore: result.hasMore,
+      limit: result.limit,
+      offset: result.offset,
     });
   } catch (error) {
     return next(error);