fix: resolve failing backend tests and dynamic import issues

backend/migrations/005_add_cliff_seconds.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE streams DROP COLUMN cliff_seconds;
+ALTER TABLE stream_archive DROP COLUMN cliff_seconds;

backend/migrations/005_add_cliff_seconds.sql

@@ -0,0 +1,6 @@
+-- Add cliff_seconds (vesting cliff in seconds) to stream tables.
+-- streamStore.ts has written this column since the Soroban create-stream wiring,
+-- but no migration ever added it — upserts failed with
+-- "table streams has no column named cliff_seconds".
+ALTER TABLE streams ADD COLUMN cliff_seconds INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE stream_archive ADD COLUMN cliff_seconds INTEGER NOT NULL DEFAULT 0;

backend/src/config/validateEnv.test.ts

@@ -598,6 +598,12 @@ describe("ADMIN_API_KEY validation", () => {
   });
 
   describe("Acceptance Criteria: Startup validation for SOROBAN_RPC_URL, STELLAR_CONTRACT_ID, and STELLAR_NETWORK", () => {
+    beforeEach(() => {
+      // test-setup.ts sets SOROBAN_DISABLED=true globally for the rest of the
+      // suite; these scenarios exercise real Soroban validation, so clear it.
+      delete process.env.SOROBAN_DISABLED;
+    });
+
     describe("in production mode", () => {
       beforeEach(() => {
         process.env.NODE_ENV = "production";

backend/src/index.test.ts

@@ -20,6 +20,7 @@ const streamStoreMocks = vi.hoisted(() => ({
   listStreamsBySender: vi.fn(),
   syncStreams: vi.fn(),
   updateStreamStartAt: vi.fn(),
+  nowInSeconds: vi.fn(() => Math.floor(Date.now() / 1000)),
 }));
 
 const eventHistoryMocks = vi.hoisted(() => ({
@@ -34,8 +35,37 @@ const eventHistoryMocks = vi.hoisted(() => ({
 
 vi.mock("./services/streamStore", () => streamStoreMocks);
 vi.mock("./services/eventHistory", () => eventHistoryMocks);
+
+// The real db module requires an initialised SQLite connection. The routes
+// under test only read the asset allowlist, so stub it out.
+vi.mock("./services/db", () => ({
+  getAllowedAssets: vi.fn(() => ["USDC", "XLM"]),
+  addAllowedAsset: vi.fn(),
+  removeAllowedAsset: vi.fn(),
+  searchStreamsFts: vi.fn(() => []),
+  syncFtsIndex: vi.fn(),
+  initDb: vi.fn(),
+  getDb: vi.fn(),
+}));
+
+// Handlers look up the in-memory cache before responding. The direct-invocation
+// helpers call handlers without an Express pipeline, so make the cache fail
+// fast (matching the catch-and-proceed fallback in the route handlers).
+vi.mock("./services/cache", () => ({
+  initCache: vi.fn(),
+  getCache: vi.fn(() => ({
+    get: vi.fn(() => {
+      throw new Error("cache unavailable in unit tests");
+    }),
+    set: vi.fn(() => {
+      throw new Error("cache unavailable in unit tests");
+    }),
+  })),
+  shutdownCache: vi.fn(),
+}));
 vi.mock("./services/auth", () => ({
   authMiddleware: vi.fn((req: any, res: any, next: any) => next()),
+  adminJwtAuth: vi.fn((req: any, res: any, next: any) => next()),
   generateChallenge: vi.fn(),
   refreshToken: vi.fn(),
   verifyChallengeAndIssueToken: vi.fn(),
@@ -163,7 +193,12 @@ function invokeListStreamsRoute(
     throw new Error("GET /api/streams route not found");
   }
 
-  const handler = layer.route.stack[0].handle as (req: any, res: any) => void;
+  // Rate-limit middleware is registered ahead of the handler; the tests invoke
+  // the handler directly with a minimal req/res, so pick the final handler.
+  const handler = layer.route.stack[layer.route.stack.length - 1].handle as (
+    req: any,
+    res: any,
+  ) => void;
 
   let statusCode = 200;
   let jsonBody: any;
@@ -178,6 +213,9 @@ function invokeListStreamsRoute(
       jsonBody = payload;
       return this;
     },
+    set() {
+      return this;
+    },
   };
 
   handler(req, res);
@@ -197,7 +235,10 @@ function invokeSenderStreamsRoute(
     throw new Error("GET /api/senders/:accountId/streams route not found");
   }
 
-  const handler = layer.route.stack[0].handle as (req: any, res: any) => void;
+  const handler = layer.route.stack[layer.route.stack.length - 1].handle as (
+    req: any,
+    res: any,
+  ) => void;
 
   let statusCode = 200;
   let jsonBody: any;
@@ -212,6 +253,9 @@ function invokeSenderStreamsRoute(
       jsonBody = payload;
       return this;
     },
+    set() {
+      return this;
+    },
   };
 
   handler(req, res);
@@ -645,15 +689,25 @@ function invokeGlobalEventsRoute(
     throw new Error("GET /api/events route not found");
   }
 
-  const handler = layer.route.stack[0].handle as (req: any, res: any) => void;
-
-  let statusCode = 200;
+  const handler = layer.route.stack[layer.route.stack.length - 1].handle as (
+    req: any,
+    res: any,
+  ) => void;  let statusCode = 200;
   let jsonBody: any;
 
   const req = { query, requestId: "test-request-id" };
   const res = {
-    status(code: number) { statusCode = code; return this; },
-    json(payload: any) { jsonBody = payload; return this; },
+    status(code: number) {
+      statusCode = code;
+      return this;
+    },
+    json(payload: any) {
+      jsonBody = payload;
+      return this;
+    },
+    set() {
+      return this;
+    },
   };
 
   handler(req, res);
@@ -686,7 +740,7 @@ describe("GET /api/events", () => {
 
     expect(status).toBe(200);
     expect(body.total).toBe(2);
-    expect(eventHistoryMocks.countAllEvents).toHaveBeenCalledWith("created");
+    expect(eventHistoryMocks.countAllEvents).toHaveBeenCalledWith("created", undefined, undefined);
 
   });
 

backend/src/index.ts

@@ -8,7 +8,7 @@ import rateLimit from "express-rate-limit";
 import swaggerUi from "swagger-ui-express";
 import { z } from "zod";
 import { createServer } from "http";
-import { searchStreamsFts, getAllowedAssets, addAllowedAsset, removeAllowedAsset } from "./services/db";
+import { searchStreamsFts, getAllowedAssets, addAllowedAsset, removeAllowedAsset, getDb } from "./services/db";
 import { initWebSocket } from "./services/websocket";
 import {
   normalizeUnknownApiError,
@@ -24,6 +24,7 @@ import {
   getStreamHistory,
   countStreamEvents,
   getStreamEventSummary,
+  recordEventWithDb,
   StreamEventType,
 } from "./services/eventHistory";
 import { fetchOpenIssues } from "./services/openIssues";
@@ -171,12 +172,12 @@ const listStreamsQuerySchema = z.object({
       `limit must be less than or equal to ${PAGINATION_MAX_LIMIT}`,
     )
     .optional(),
-  sort: z
-    .enum(SORT_FIELDS)
-    .optional(),
-  order: z
-    .enum(SORT_ORDERS)
-    .optional(),
+  sort: z.enum(SORT_FIELDS, {
+    message: `sort must be one of: ${SORT_FIELDS.join(", ")}`,
+  }).optional(),
+  order: z.enum(SORT_ORDERS, {
+    message: `order must be one of: ${SORT_ORDERS.join(", ")}`,
+  }).optional(),
 });
 
 const AUTH_CHALLENGE_RATE_LIMIT = Number(
@@ -545,10 +546,21 @@ app.get("/api/streams", readLimiter, async (req: Request, res: Response) => {
   const hasLimit = req.query.limit !== undefined;
 
   const now = nowInSeconds();
-  let data = listStreams(query.include_archived, query.sort ?? "createdAt", query.order ?? "desc").map((stream) => ({
-    ...stream,
-    progress: calculateProgress(stream, now),
-  }));
+  let data;
+  try {
+    data = listStreams(query.include_archived, query.sort ?? "createdAt", query.order ?? "desc").map((stream) => ({
+      ...stream,
+      progress: calculateProgress(stream, now),
+    }));
+  } catch (error: any) {
+    // DB errors (e.g. the connection was closed) must surface as a 500, not
+    // an unhandled rejection that hangs the request.
+    logger.error({ err: error }, "failed to list streams");
+    sendApiError(req, res, 500, "Failed to list streams.", {
+      code: "INTERNAL_ERROR",
+    });
+    return;
+  }
 
   if (query.status) {
     data = data.filter((stream) => stream.progress.status === query.status);
@@ -663,8 +675,13 @@ app.get("/api/events", readLimiter, (req: Request, res: Response) => {
 
   const total = countAllEvents(eventType, streamId, since);
 
+  const hasPage = req.query.page !== undefined;
+  const hasLimit =
+    req.query.limit !== undefined || req.query.pageSize !== undefined;
+
   const page = query.page ?? PAGINATION_DEFAULT_PAGE;
   const pageSize = query.pageSize ?? query.limit ?? PAGINATION_DEFAULT_LIMIT;
+  const limit = !hasPage && !hasLimit ? total : pageSize;
 
   const offset = (page - 1) * pageSize;
   const data = getGlobalEvents(
@@ -676,7 +693,7 @@ app.get("/api/events", readLimiter, (req: Request, res: Response) => {
     since,
   );
 
-  res.json({ data, total, page, pageSize, limit: pageSize });
+  res.json({ data, total, page, pageSize, limit });
 });
 
 app.get(
@@ -767,6 +784,61 @@ const claimableBatchBodySchema = z.object({
     .max(50, "Maximum 50 stream IDs per batch"),
 });
 
+// GET /api/streams/:id/claimable — real-time claimable amount for a single stream
+app.get(
+  "/api/streams/:id/claimable",
+  claimableLimiter,
+  async (req: Request, res: Response) => {
+    const parsedId = parseStreamId(req.params.id);
+    if (!parsedId.ok) {
+      sendValidationError(req, res, parsedId.issues);
+      return;
+    }
+
+    const stream = getStream(parsedId.value);
+    if (!stream) {
+      sendApiError(req, res, 404, "Stream not found.", { code: "NOT_FOUND" });
+      return;
+    }
+
+    try {
+      if (stream.pausedAt !== undefined || stream.canceledAt !== undefined) {
+        const at = await getLatestLedgerTime();
+        res.json({
+          streamId: stream.id,
+          claimableAmount: 0,
+          assetCode: stream.assetCode,
+          at,
+        });
+        return;
+      }
+
+      const { claimableAmount, at } = await getOnChainClaimableAmount(stream.id);
+      res.json({
+        streamId: stream.id,
+        claimableAmount: Number(claimableAmount),
+        assetCode: stream.assetCode,
+        at,
+      });
+    } catch (error: any) {
+      logger.error({ err: error, streamId: parsedId.value }, "failed to query claimable amount");
+      const normalizedError = normalizeUnknownApiError(
+        error,
+        "Failed to query claimable amount.",
+      );
+      sendApiError(
+        req,
+        res,
+        normalizedError.statusCode,
+        normalizedError.message,
+        {
+          code: normalizedError.code ?? "INTERNAL_ERROR",
+        },
+      );
+    }
+  },
+);
+
 app.post(
   "/api/streams/claimable/batch",
   claimableLimiter,
@@ -1408,6 +1480,59 @@ app.post(
 );
 
 // POST /api/streams/:id/pause — sender pauses an active stream
+// POST /api/streams/:id/mark-complete — sender marks a fully-vested stream as complete
+app.post(
+  "/api/streams/:id/mark-complete",
+  mutationLimiter,
+  authMiddleware,
+  async (req: Request, res: Response) => {
+    const parsedId = parseStreamId(req.params.id);
+    if (!parsedId.ok) {
+      sendValidationError(req, res, parsedId.issues);
+      return;
+    }
+
+    const stream = getStream(parsedId.value);
+    if (!stream) {
+      sendApiError(req, res, 404, "Stream not found.", { code: "NOT_FOUND" });
+      return;
+    }
+
+    const user = (req as any).user;
+    if (stream.sender !== user.accountId) {
+      sendApiError(req, res, 403, "Only the sender can complete this stream.", {
+        code: "FORBIDDEN",
+      });
+      return;
+    }
+
+    try {
+      const updated = markStreamComplete(parsedId.value);
+      res.json({
+        data: {
+          ...updated,
+          progress: calculateProgress(updated),
+        },
+      });
+    } catch (error: any) {
+      logger.error({ err: error, streamId: parsedId.value }, "failed to mark stream complete");
+      const normalizedError = normalizeUnknownApiError(
+        error,
+        "Failed to mark stream complete.",
+      );
+      sendApiError(
+        req,
+        res,
+        normalizedError.statusCode,
+        normalizedError.message,
+        {
+          code: normalizedError.code ?? "INTERNAL_ERROR",
+        },
+      );
+    }
+  },
+);
+
 app.post(
   "/api/streams/:id/pause",
   mutationLimiter,
@@ -1538,8 +1663,7 @@ app.post(
     try {
       // Record the claim event in the local DB.
       // In a full on-chain implementation this would submit a `claim` Soroban tx.
-      const db = (await import("./services/db")).getDb();
-      const { recordEventWithDb } = await import("./services/eventHistory");
+      const db = getDb();
       const now = Math.floor(Date.now() / 1000);
 
       // Guard against double-spend: check and write inside one atomic transaction.
@@ -1572,9 +1696,7 @@ app.post(
         return;
       }
 
-      const history = await import("./services/eventHistory").then((m) =>
-        m.getStreamHistory(stream.id),
-      );
+      const history = getStreamHistory(stream.id);
 
       res.json({
         result: {