Endpoint

src/repositories/streams.ts

@@ -60,10 +60,36 @@ export interface StreamFilter {
   sender?: string;
   recipient?: string;
   token?: string;
+  cancelled?: boolean;
+  cursor?: bigint;
   limit?: number;
   offset?: number;
 }
 
+export interface StreamListResult {
+  streams: Stream[];
+  nextCursor?: string;
+}
+
+export function encodeCursor(streamId: bigint): string {
+  const payload = JSON.stringify({ streamId: streamId.toString() });
+  return Buffer.from(payload, "utf8").toString("base64url");
+}
+
+export function decodeCursor(raw: string): bigint | null {
+  try {
+    const decoded = Buffer.from(raw, "base64url").toString("utf8");
+    const parsed = JSON.parse(decoded);
+    if (parsed && typeof parsed.streamId === "string") {
+      const value = BigInt(parsed.streamId);
+      if (value >= 0n) return value;
+    }
+  } catch {
+    // fall through to null
+  }
+  return null;
+}
+
 function decimal(value: bigint): Prisma.Decimal {
   return new Prisma.Decimal(value.toString());
 }
@@ -221,16 +247,31 @@ function whereFromFilter(filter: StreamFilter): Prisma.StreamWhereInput {
   if (filter.sender) where.sender = filter.sender;
   if (filter.recipient) where.recipient = filter.recipient;
   if (filter.token) where.token = filter.token;
+  if (filter.cancelled !== undefined) where.cancelled = filter.cancelled;
+  if (filter.cursor !== undefined) where.streamId = { lt: filter.cursor };
   return where;
 }
 
-export async function listStreams(filter: StreamFilter): Promise<Stream[]> {
-  return prisma.stream.findMany({
+export async function listStreams(filter: StreamFilter): Promise<StreamListResult> {
+  const limit = filter.limit ?? 50;
+  const useCursor = filter.cursor !== undefined;
+  const take = limit + 1;
+  const skip = useCursor ? 0 : filter.offset ?? 0;
+
+  const rows = await prisma.stream.findMany({
     where: whereFromFilter(filter),
     orderBy: { streamId: "desc" },
-    take: filter.limit ?? 50,
-    skip: filter.offset ?? 0,
+    take,
+    skip,
   });
+
+  if (rows.length <= limit) {
+    return { streams: rows };
+  }
+
+  const streams = rows.slice(0, limit);
+  const last = streams[streams.length - 1];
+  return { streams, nextCursor: encodeCursor(last.streamId) };
 }
 
 // Total number of streams matching the filter, ignoring limit and offset, so a

src/routes/streams.ts

@@ -4,6 +4,7 @@ import { StrKey } from "@stellar/stellar-sdk";
 import {
   aggregateStreams,
   countStreams,
+  decodeCursor,
   getStream,
   listStreams,
 } from "../repositories/streams.js";
@@ -86,6 +87,12 @@ function parseIncludeTotal(raw: string | undefined): boolean {
   return raw === "true";
 }
 
+function parseCancelled(raw: string | undefined): boolean | undefined {
+  if (raw === "true") return true;
+  if (raw === "false") return false;
+  return undefined;
+}
+
 // Stellar addresses are canonical uppercase base32 strkeys, but callers
 // sometimes send lowercase or whitespace-padded spellings. Normalize those
 // before matching so every rendering of one address filters identically, and
@@ -135,7 +142,9 @@ export async function streamRoutes(app: FastifyInstance): Promise<void> {
         summary: "List streams",
         description:
           "Returns a paginated list of token streams. Optionally filter by sender, recipient, or token address; " +
-          "address filters accept lowercase and whitespace-padded spellings and are normalized before matching.",
+          "address filters accept lowercase and whitespace-padded spellings and are normalized before matching. " +
+          "Use the opaque cursor returned by previous responses for stable pagination under concurrent inserts; " +
+          "when cursor is provided, offset is ignored and offset ceiling checks are skipped.",
         tags: ["streams"],
         querystring: {
           type: "object",
@@ -161,7 +170,7 @@ export async function streamRoutes(app: FastifyInstance): Promise<void> {
             },
             offset: {
               type: "string",
-              description: `Zero-based offset for pagination. Defaults to 0 and must not exceed ${MAX_OFFSET}.`,
+              description: `Zero-based offset for pagination. Defaults to 0 and must not exceed ${MAX_OFFSET}. Ignored when cursor is provided.`,
             },
             includeTotal: {
               type: "string",
@@ -170,6 +179,18 @@ export async function streamRoutes(app: FastifyInstance): Promise<void> {
                 "When true, the response includes the total number of streams matching the filters. " +
                 "Defaults to false, which skips the count query and omits total from the response.",
             },
+            cancelled: {
+              type: "string",
+              enum: ["true", "false"],
+              description:
+                "Filter by cancellation status. Omit to return both cancelled and active streams.",
+            },
+            cursor: {
+              type: "string",
+              description:
+                "Opaque cursor returned by a previous list response. Use this to fetch the next page " +
+                "with stable ordering under concurrent inserts. Takes precedence over offset when both are provided.",
+            },
           },
           additionalProperties: false,
         },
@@ -187,21 +208,44 @@ export async function streamRoutes(app: FastifyInstance): Promise<void> {
         limit?: string;
         offset?: string;
         includeTotal?: string;
+        cancelled?: string;
+        cursor?: string;
       };
 
       const limit = parseLimit(query.limit);
       const offset = parseOffset(query.offset);
-      if (offset > MAX_OFFSET) {
+
+      let cursor: bigint | undefined;
+      if (query.cursor !== undefined) {
+        const decoded = decodeCursor(query.cursor);
+        if (decoded === null) {
+          return reply.code(400).send({
+            code: "VALIDATION_ERROR",
+            error: "invalid cursor",
+            requestId: request.id,
+          });
+        }
+        cursor = decoded;
+      }
+
+      const usingCursor = cursor !== undefined;
+      if (!usingCursor && offset > MAX_OFFSET) {
         return reply.code(400).send({
           code: "VALIDATION_ERROR",
           error:
             `offset must not exceed ${MAX_OFFSET}. Page through results in order with limit and offset, ` +
-            "or narrow them with the sender, recipient, and token filters.",
+            "or narrow them with the sender, recipient, and token filters, or use the returned cursor for stable pagination.",
           requestId: request.id,
         });
       }
 
-      const filter: { sender?: string; recipient?: string; token?: string } = {};
+      const filter: {
+        sender?: string;
+        recipient?: string;
+        token?: string;
+        cancelled?: boolean;
+        cursor?: bigint;
+      } = {};
       for (const field of ["sender", "recipient", "token"] as const) {
         const raw = query[field];
         if (!raw) continue;
@@ -216,19 +260,23 @@ export async function streamRoutes(app: FastifyInstance): Promise<void> {
         filter[field] = normalized;
       }
 
+      filter.cancelled = parseCancelled(query.cancelled);
+      filter.cursor = cursor;
+
       const includeTotal = parseIncludeTotal(query.includeTotal);
 
-      const [streams, total] = await Promise.all([
+      const [listResult, total] = await Promise.all([
         listStreams({ ...filter, limit, offset }),
         includeTotal ? countStreams(filter) : Promise.resolve(undefined),
       ]);
 
       reply.header("Cache-Control", "public, max-age=30");
       return {
-        streams: streams.map(toView),
+        streams: listResult.streams.map(toView),
         ...(total === undefined ? {} : { total }),
+        ...(listResult.nextCursor === undefined ? {} : { nextCursor: listResult.nextCursor }),
         limit,
-        offset,
+        offset: usingCursor ? 0 : offset,
       };
     },
   );

src/schema.ts

@@ -140,7 +140,8 @@ export const streamListResponseSchema = {
   type: "object",
   description:
     "Paginated list of streams. The total is only computed and included when " +
-    "the request opted in with includeTotal=true.",
+    "the request opted in with includeTotal=true. Opaque cursor pagination is " +
+    "stable under concurrent inserts and takes precedence over offset when both are provided.",
   required: ["streams", "limit", "offset"],
   properties: {
     streams: {
@@ -162,9 +163,18 @@ export const streamListResponseSchema = {
     },
     offset: {
       type: "integer",
-      description: "Zero-based index of the first stream on this page.",
+      description:
+        "Zero-based index of the first stream on this page. Reflects the offset " +
+        "from the request and is 0 on cursor-driven pages.",
       examples: [0],
     },
+    nextCursor: {
+      type: "string",
+      description:
+        "Opaque cursor for the next page. Omitted when there are no further pages. " +
+        "Pass this as the cursor query parameter to fetch the next page with stable ordering.",
+      examples: ["eyJzdHJlYW1JZCI6IjQyIn0"],
+    },
   },
 } as const;
 

tests/routes/openapi.test.ts

@@ -146,6 +146,19 @@ describe("GET /docs/json (OpenAPI spec)", () => {
     expect(listSchema?.properties).toHaveProperty("streams");
   });
 
+  it("StreamListResponse exposes nextCursor for opaque cursor pagination", async () => {
+    const spec = await getSpec();
+    const schemas = (
+      spec.components as { schemas?: Record<string, unknown> }
+    )?.schemas ?? {};
+    const listSchema = schemas["StreamListResponse"] as {
+      properties?: Record<string, unknown>;
+      required?: string[];
+    };
+    expect(listSchema?.properties).toHaveProperty("nextCursor");
+    expect(listSchema?.required).not.toContain("nextCursor");
+  });
+
   it("StreamListResponse.total is optional so both response variants are valid", async () => {
     const spec = await getSpec();
     const schemas = (
@@ -165,6 +178,28 @@ describe("GET /docs/json (OpenAPI spec)", () => {
     expect(names).toContain("offset");
   });
 
+  it("documents the cancelled query parameter on GET /streams", async () => {
+    const spec = await getSpec();
+    const paths = spec.paths as Record<string, Record<string, unknown>>;
+    const params = (paths["/streams"]?.get as { parameters?: Array<Record<string, unknown>> })
+      ?.parameters ?? [];
+    const cancelledParam = params.find((p) => p.name === "cancelled");
+    expect(cancelledParam).toBeDefined();
+    expect((cancelledParam as { schema?: { enum?: string[] } })?.schema?.enum).toEqual(
+      expect.arrayContaining(["true", "false"]),
+    );
+  });
+
+  it("documents the cursor query parameter on GET /streams", async () => {
+    const spec = await getSpec();
+    const paths = spec.paths as Record<string, Record<string, unknown>>;
+    const params = (paths["/streams"]?.get as { parameters?: Array<Record<string, unknown>> })
+      ?.parameters ?? [];
+    const cursorParam = params.find((p) => p.name === "cursor");
+    expect(cursorParam).toBeDefined();
+    expect((cursorParam as { description?: string })?.description).toMatch(/cursor|opaque/i);
+  });
+
   it("exports StreamSummaryResponse as a reusable component schema", async () => {
     const spec = await getSpec();
     const schemas = (

tests/routes/streams-large-values.test.ts

@@ -137,10 +137,12 @@ describe("large uint128 values (#65)", () => {
 
   describe("GET /streams with large values", () => {
     it("returns large amounts as strings in the list endpoint", async () => {
-      streamsRepo.listStreams.mockResolvedValue([
-        makeLargeStream({ streamId: BigInt(1) }),
-        makeLargeStream({ streamId: BigInt(2), withdrawn: 1n }),
-      ]);
+      streamsRepo.listStreams.mockResolvedValue({
+        streams: [
+          makeLargeStream({ streamId: BigInt(1) }),
+          makeLargeStream({ streamId: BigInt(2), withdrawn: 1n }),
+        ],
+      });
       streamsRepo.countStreams.mockResolvedValue(2);
 
       const res = await listStreams();