Add GET /api/streams/:id/claimable

backend/src/index.ts

@@ -42,6 +42,8 @@ import {
   cancelStream,
   createStream,
   getStream,
+  getOnChainClaimableAmount,
+  getLatestLedgerTime,
   initSoroban,
   listStreams,
   listStreamsByRecipient,
@@ -201,6 +203,25 @@ const mutationLimiter = rateLimit({
   },
 });
 
+const CLAIMABLE_RATE_LIMIT = Number(process.env.CLAIMABLE_RATE_LIMIT ?? 30);
+
+const claimableLimiter = rateLimit({
+  windowMs: 60 * 1000,
+  max: CLAIMABLE_RATE_LIMIT,
+  standardHeaders: true,
+  legacyHeaders: false,
+  handler: (req: Request, res: Response) => {
+    const resetTime = (req as any).rateLimit?.resetTime;
+    const retryAfter = resetTime
+      ? Math.ceil((resetTime.getTime() - Date.now()) / 1000)
+      : 60;
+    res.set("Retry-After", String(Math.max(1, retryAfter)));
+    sendApiError(req, res, 429, "Too many requests. Please try again later.", {
+      code: "RATE_LIMIT_EXCEEDED",
+    });
+  },
+});
+
 app.use(cors());
 app.use(requestLogger);
 app.use(
@@ -501,13 +522,7 @@ app.get("/api/streams/:id", readLimiter, (req: Request, res: Response) => {
   });
 });
 
-app.get(
-  "/api/recipients/:accountId/streams",
-  readLimiter,
-  (req: Request, res: Response) => {
-    const parsedParams = recipientAccountIdSchema.safeParse({
-      accountId: req.params.accountId,
-    });
+
 
     if (!parsedParams.success) {
       sendValidationError(req, res, parsedParams.error.issues);

backend/src/integration.test.ts

@@ -1,11 +1,33 @@
-import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import request from "supertest";
 import { app } from "./index";
 import { initDb, getDb } from "./services/db";
 import { Keypair } from "@stellar/stellar-sdk";
 import path from "path";
 import fs from "fs";
 
+const mockSimulateTransaction = vi.fn();
+const mockGetLatestLedger = vi.fn();
+
+vi.mock("@stellar/stellar-sdk", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("@stellar/stellar-sdk")>();
+  return {
+    ...actual,
+    rpc: {
+      ...actual.rpc,
+      Server: vi.fn().mockImplementation(() => ({
+        getLatestLedger: mockGetLatestLedger,
+        simulateTransaction: mockSimulateTransaction,
+        prepareTransaction: vi.fn().mockImplementation((tx) => tx),
+      })),
+      Api: {
+        ...actual.rpc.Api,
+        isSimulationSuccess: (response: any) => response.kind === "success",
+      }
+    },
+  };
+});
+
 
 // Use a separate test database
 const TEST_DB_PATH = path.join(__dirname, "..", "data", "test-streams.db");
@@ -415,6 +437,118 @@ describe("Backend Integration Tests", () => {
       });
     });
 
+    describe("GET /api/streams/:id/claimable", () => {
+      beforeEach(() => {
+        vi.clearAllMocks();
+      });
+
+      it("should return 200 and the claimable amount from Soroban simulation", async () => {
+        mockGetLatestLedger.mockResolvedValue({
+          sequence: 12345,
+          closeTime: "1716812160",
+        });
+
+        mockSimulateTransaction.mockResolvedValue({
+          kind: "success",
+          result: {
+            retval: 450,
+          },
+        });
+
+        const response = await request(app).get(`/api/streams/${mockStream.id}/claimable`);
+
+        expect(response.status).toBe(200);
+        expect(response.body).toEqual({
+          streamId: mockStream.id,
+          claimableAmount: 450,
+          assetCode: mockStream.assetCode,
+          at: 1716812160,
+        });
+
+        expect(mockGetLatestLedger).toHaveBeenCalled();
+        expect(mockSimulateTransaction).toHaveBeenCalled();
+      });
+
+      it("should return 0 when stream is paused", async () => {
+        const db = getDb();
+        db.prepare("UPDATE streams SET paused_at = ? WHERE id = ?").run(Math.floor(Date.now() / 1000) - 1000, mockStream.id);
+
+        mockGetLatestLedger.mockResolvedValue({
+          sequence: 12345,
+          closeTime: "1716812160",
+        });
+
+        const response = await request(app).get(`/api/streams/${mockStream.id}/claimable`);
+
+        expect(response.status).toBe(200);
+        expect(response.body).toEqual({
+          streamId: mockStream.id,
+          claimableAmount: 0,
+          assetCode: mockStream.assetCode,
+          at: 1716812160,
+        });
+
+        expect(mockSimulateTransaction).not.toHaveBeenCalled();
+      });
+
+      it("should return 0 when stream is canceled", async () => {
+        const db = getDb();
+        db.prepare("UPDATE streams SET canceled_at = ? WHERE id = ?").run(Math.floor(Date.now() / 1000) - 1000, mockStream.id);
+
+        mockGetLatestLedger.mockResolvedValue({
+          sequence: 12345,
+          closeTime: "1716812160",
+        });
+
+        const response = await request(app).get(`/api/streams/${mockStream.id}/claimable`);
+
+        expect(response.status).toBe(200);
+        expect(response.body).toEqual({
+          streamId: mockStream.id,
+          claimableAmount: 0,
+          assetCode: mockStream.assetCode,
+          at: 1716812160,
+        });
+
+        expect(mockSimulateTransaction).not.toHaveBeenCalled();
+      });
+
+      it("should return 404 for non-existent stream", async () => {
+        const response = await request(app).get("/api/streams/999/claimable");
+
+        expect(response.status).toBe(404);
+        expect(response.body.error).toBe("Stream not found.");
+      });
+
+      it("should return 400 for invalid stream ID", async () => {
+        const response = await request(app).get("/api/streams/invalid-id/claimable");
+
+        expect(response.status).toBe(400);
+        expect(response.body.error).toContain("Stream ID must be");
+      });
+
+      it("should enforce rate limit of 30 requests per minute", async () => {
+        mockGetLatestLedger.mockResolvedValue({
+          sequence: 12345,
+          closeTime: "1716812160",
+        });
+        mockSimulateTransaction.mockResolvedValue({
+          kind: "success",
+          result: { retval: 10 },
+        });
+
+        for (let i = 0; i < 31; i++) {
+          const response = await request(app).get(`/api/streams/${mockStream.id}/claimable`);
+          if (i < 30) {
+            expect(response.status).toBe(200);
+          } else {
+            expect(response.status).toBe(429);
+            expect(response.body.code).toBe("RATE_LIMIT_EXCEEDED");
+          }
+        }
+      });
+    });
+
     describe("GET /api/recipients/:accountId/streams", () => {
       it("should get streams for a recipient", async () => {
         const response = await request(app)

backend/src/services/streamStore.ts

@@ -386,6 +386,7 @@ export function calculateProgress(
     stream.pausedAt !== undefined ? Math.min(at, stream.pausedAt) : at;
 
 
+
   const ratio = Math.min(1, elapsed / stream.durationSeconds);
   const vestedAmount = stream.totalAmount * ratio;
 
@@ -399,13 +400,46 @@ export function calculateProgress(
   };
 }
 
-/**
- * Syncs all on-chain streams from Soroban contract to local SQLite database.
- * Fetches streams in parallel (max 5 concurrent RPC calls) with fallback to sequential.
- * Updates existing streams and inserts new ones.
- * @async
- * @returns {Promise<void>}
- */
+export async function getOnChainClaimableAmount(
+  id: string,
+): Promise<{ claimableAmount: number; at: number }> {
+  const sorobanContext = getSorobanContext();
+  if (!sorobanContext || !rpcServer) {
+    throw new Error("Soroban RPC server is not initialized.");
+  }
+
+  const sourceAccount = await sorobanContext.sourceAccountPromise;
+  const latestLedger = await rpcServer.getLatestLedger();
+  const at = latestLedger.closeTime ? parseInt(latestLedger.closeTime, 10) : Math.floor(Date.now() / 1000);
+
+  const simRes = await simulateContractCall(
+    sorobanContext.contract,
+    sourceAccount,
+    "claimable",
+    nativeToScVal(parseInt(id), { type: "u64" }),
+    nativeToScVal(at, { type: "u64" }),
+  );
+
+  if (!rpc.Api.isSimulationSuccess(simRes) || !simRes.result) {
+    throw new Error("Simulation failed: " + JSON.stringify(simRes));
+  }
+
+  const claimableAmount = Number(scValToNative(simRes.result.retval));
+  return { claimableAmount, at };
+}
+
+export async function getLatestLedgerTime(): Promise<number> {
+  if (!rpcServer) {
+    return Math.floor(Date.now() / 1000);
+  }
+  try {
+    const latestLedger = await rpcServer.getLatestLedger();
+    return latestLedger.closeTime ? parseInt(latestLedger.closeTime, 10) : Math.floor(Date.now() / 1000);
+  } catch (e) {
+    return Math.floor(Date.now() / 1000);
+  }
+}
+
 export async function syncStreams() {
   const sorobanContext = getSorobanContext();
   if (!sorobanContext) return;
@@ -913,7 +947,6 @@ export async function cancelStream(
 }
 
 
-
 /**
  * Updates the start time of a scheduled stream.
  * Only scheduled streams (not yet started) can have their start time updated.

backend/src/swagger.ts

@@ -694,6 +694,61 @@ export const swaggerDocument = {
         },
       },
     },
+    "/api/streams/{id}/claimable": {
+      get: {
+        summary: "Get real-time claimable amount",
+        description: "Retrieves the current real-time claimable amount for a stream using Soroban contract simulation. Returns 0 if paused, canceled, or before the cliff.",
+        parameters: [
+          {
+            name: "id",
+            in: "path",
+            required: true,
+            description: "The unique ID of the stream.",
+            schema: {
+              type: "string",
+            },
+          },
+        ],
+        responses: {
+          "200": {
+            description: "Real-time claimable amount and query context.",
+            content: {
+              "application/json": {
+                schema: {
+                  type: "object",
+                  properties: {
+                    streamId: { type: "string", example: "1" },
+                    claimableAmount: { type: "number", example: 450.123456 },
+                    assetCode: { type: "string", example: "USDC" },
+                    at: { type: "integer", description: "Ledger timestamp at which query was simulated", example: 1716812160 },
+                  },
+                },
+              },
+            },
+          },
+          "404": {
+            description: "Stream not found.",
+            content: {
+              "application/json": {
+                schema: {
+                  $ref: "#/components/schemas/Error",
+                },
+              },
+            },
+          },
+          "500": {
+            description: "Failed to simulate claimable amount.",
+            content: {
+              "application/json": {
+                schema: {
+                  $ref: "#/components/schemas/Error",
+                },
+              },
+            },
+          },
+        },
+      },
+    },
     "/api/recipients/{accountId}/streams": {
       get: {
         summary: "Get recipient streams",