fix(api,export): resolve stream statuses, accept full-width IDs, inde…

backend/src/graphql/resolvers/stream.resolver.ts

@@ -40,11 +40,21 @@ function parseCursor(cursor: string | null | undefined): bigint | undefined {
   }
 }
 
+const MAX_U64 = 18446744073709551615n;
+
 export async function resolveStream(
   ctx: GraphQLContext,
   streamId: number | string,
 ): Promise<unknown> {
-  const id = BigInt(streamId);
+  let id: bigint;
+  try {
+    id = BigInt(streamId);
+  } catch {
+    throw new Error(`Invalid stream identifier: ${streamId}`);
+  }
+  if (id < 0n || id > MAX_U64) {
+    throw new Error(`Stream identifier out of supported range (0 to 2^64 - 1): ${streamId}`);
+  }
   return ctx.loaders.stream.load(id);
 }
 
@@ -57,7 +67,30 @@ export async function resolveStreams(
   const filter = args.filter ?? {};
 
   const where: Prisma.StreamWhereInput = {};
-  if (filter.status) where.isActive = filter.status.toLowerCase() === "active";
+  if (filter.status !== undefined && filter.status !== null) {
+    const normalized = filter.status.toLowerCase().trim();
+    switch (normalized) {
+      case "active":
+        where.isActive = true;
+        where.isPaused = false;
+        break;
+      case "paused":
+        where.isPaused = true;
+        break;
+      case "cancelled":
+        where.isActive = false;
+        where.events = { some: { eventType: "CANCELLED" } };
+        break;
+      case "completed":
+        where.isActive = false;
+        where.events = { some: { eventType: "COMPLETED" } };
+        break;
+      default:
+        throw new Error(
+          `Invalid status filter: '${filter.status}'. Supported statuses are 'active', 'paused', 'cancelled', 'completed'.`,
+        );
+    }
+  }
   if (filter.tokenAddress) where.tokenAddress = filter.tokenAddress;
   if (filter.sender) where.sender = filter.sender;
   if (filter.recipient) where.recipient = filter.recipient;

backend/src/graphql/schema.ts

@@ -144,7 +144,7 @@ const Query = queryType({
   definition(t) {
     t.field("stream", {
       type: "Stream",
-      args: { streamId: nonNull(intArg()) },
+      args: { streamId: nonNull(stringArg()) },
       resolve: (_source: any, args: any, ctx: any) =>
         resolveStream(asContext(ctx), args.streamId),
     });
@@ -167,7 +167,7 @@ const Subscription = subscriptionType({
   definition(t) {
     t.field("streamUpdated", {
       type: "Stream",
-      args: { streamId: nonNull(intArg()) },
+      args: { streamId: nonNull(stringArg()) },
       subscribe: () => {
         throw new Error("Subscriptions are delivered over the SSE gateway, not HTTP POST.");
       },

backend/src/services/export.service.ts

@@ -46,7 +46,7 @@ const CSV_HEADERS = [
  * Format amount from stroops (i128) to human-readable decimal
  * Assumes 7 decimals for Stellar tokens
  */
-function formatAmount(stroops: string, decimals = 7): string {
+export function formatAmount(stroops: string, decimals = 7): string {
   const amount = BigInt(stroops);
   const divisor = BigInt(10 ** decimals);
   const wholePart = amount / divisor;
@@ -56,22 +56,47 @@ function formatAmount(stroops: string, decimals = 7): string {
 }
 
 /**
- * Calculate protocol fee (simplified - adjust based on actual fee structure)
+ * Extract protocol fee for an event from indexed events or metadata.
+ * Fees are charged on create/top-up and indexed via FEE_COLLECTED events.
+ * No arbitrary withdrawal fees are fabricated.
  */
-function calculateProtocolFee(amount: string, eventType: string): string {
-  // Example: 0.5% fee on withdrawals
-  if (eventType === "WITHDRAWN") {
-    const amt = BigInt(amount);
-    const fee = (amt * 5n) / 1000n; // 0.5%
-    return fee.toString();
+export function getProtocolFeeForEvent(event: any, stream: any): string {
+  // If the event itself is FEE_COLLECTED
+  if (event.eventType === "FEE_COLLECTED") {
+    return event.amount || "0";
   }
+
+  // Check if there is an indexed FEE_COLLECTED event associated with this transaction
+  if (stream?.events && Array.isArray(stream.events)) {
+    const feeEvent = stream.events.find(
+      (e: any) =>
+        e.eventType === "FEE_COLLECTED" &&
+        e.transactionHash === event.transactionHash,
+    );
+    if (feeEvent && feeEvent.amount) {
+      return feeEvent.amount;
+    }
+  }
+
+  // Check event metadata for embedded fee information if present
+  if (event.metadata) {
+    try {
+      const meta = typeof event.metadata === "string" ? JSON.parse(event.metadata) : event.metadata;
+      if (meta?.feeAmount || meta?.fee_amount || meta?.protocolFee) {
+        return (meta.feeAmount || meta.fee_amount || meta.protocolFee).toString();
+      }
+    } catch {
+      // ignore metadata parse error
+    }
+  }
+
   return "0";
 }
 
 /**
  * Convert database records to transaction export format
  */
-function mapEventToTransaction(
+export function mapEventToTransaction(
   event: any,
   stream: any,
   userAddress: string,
@@ -81,8 +106,19 @@ function mapEventToTransaction(
   const counterparty = isOutgoing ? stream.recipient : stream.sender;
 
   const grossAmount = event.amount || "0";
-  const protocolFee = calculateProtocolFee(grossAmount, event.eventType);
-  const netAmount = (BigInt(grossAmount) - BigInt(protocolFee)).toString();
+  const protocolFee = getProtocolFeeForEvent(event, stream);
+
+  // Calculate net amount:
+  // For outgoing create/top-up or fee-bearing transactions, gross is input and fee is deducted.
+  // For incoming/withdrawals without fee, netAmount equals grossAmount.
+  let netAmountVal = BigInt(grossAmount);
+  const feeVal = BigInt(protocolFee);
+  if (netAmountVal >= feeVal) {
+    netAmountVal -= feeVal;
+  } else {
+    netAmountVal = 0n;
+  }
+  const netAmount = netAmountVal.toString();
 
   return {
     timestamp: new Date(Number(event.timestamp) * 1000).toISOString(),
@@ -102,7 +138,7 @@ function mapEventToTransaction(
 /**
  * Convert transaction record to CSV row
  */
-function transactionToCSVRow(record: TransactionRecord): string {
+export function transactionToCSVRow(record: TransactionRecord): string {
   const escapeCSV = (value: string): string => {
     if (value.includes(",") || value.includes('"') || value.includes("\n")) {
       return `"${value.replace(/"/g, '""')}"`;
@@ -222,6 +258,12 @@ export async function streamTransactionCSV(
     logger.error("[Export] CSV streaming error:", error);
     if (!res.headersSent) {
       res.status(500).json({ error: "Export failed" });
+    } else {
+      if (typeof res.destroy === "function") {
+        res.destroy(error as Error);
+      } else {
+        res.end();
+      }
     }
   }
 }
@@ -332,6 +374,12 @@ export async function streamTransactionJSON(
     logger.error("[Export] JSON streaming error:", error);
     if (!res.headersSent) {
       res.status(500).json({ error: "Export failed" });
+    } else {
+      if (typeof res.destroy === "function") {
+        res.destroy(error as Error);
+      } else {
+        res.end();
+      }
     }
   }
 }

backend/tests/export.service.test.ts

@@ -0,0 +1,191 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import {
+  formatAmount,
+  getProtocolFeeForEvent,
+  mapEventToTransaction,
+  streamTransactionCSV,
+  streamTransactionJSON,
+} from "../src/services/export.service.js";
+
+const { prismaMock } = vi.hoisted(() => ({
+  prismaMock: {
+    stream: {
+      findMany: vi.fn(),
+    },
+  },
+}));
+
+vi.mock("../src/lib/prisma.js", () => ({
+  prisma: prismaMock,
+  default: prismaMock,
+}));
+
+describe("Export Service - Issues #1700 & #1701", () => {
+  const userAddress = "GUSER1234567890ABCDEF1234567890ABCDEF1234567890";
+  const counterparty = "GPARTY1234567890ABCDEF1234567890ABCDEF1234567890";
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  describe("Issue #1700: Use indexed protocol fees in accounting exports", () => {
+    it("reports zero fee on withdrawals and preserves exact net amount (no fabricated 0.5% fee)", () => {
+      const withdrawalEvent = {
+        eventType: "WITHDRAWN",
+        amount: "10000000", // 1 token (7 decimals)
+        transactionHash: "tx_withdrawn_1",
+        timestamp: 1700000000n,
+      };
+
+      const stream = {
+        streamId: 42n,
+        sender: counterparty,
+        recipient: userAddress,
+        tokenAddress: "CDK...",
+        events: [withdrawalEvent],
+      };
+
+      const fee = getProtocolFeeForEvent(withdrawalEvent, stream);
+      expect(fee).toBe("0");
+
+      const tx = mapEventToTransaction(withdrawalEvent, stream, userAddress);
+      expect(tx.protocolFeeDeducted).toBe("0");
+      expect(tx.grossAmountStroops).toBe("10000000");
+      expect(tx.grossAmountFormatted).toBe("1.0000000");
+      expect(tx.netAmount).toBe("1.0000000");
+    });
+
+    it("derives nonzero fee on stream creation from indexed FEE_COLLECTED event", () => {
+      const createEvent = {
+        eventType: "CREATED",
+        amount: "10000000",
+        transactionHash: "tx_create_1",
+        timestamp: 1700000000n,
+      };
+
+      const feeEvent = {
+        eventType: "FEE_COLLECTED",
+        amount: "200000", // 0.02 token fee
+        transactionHash: "tx_create_1",
+        timestamp: 1700000000n,
+      };
+
+      const stream = {
+        streamId: 42n,
+        sender: userAddress,
+        recipient: counterparty,
+        tokenAddress: "CDK...",
+        events: [createEvent, feeEvent],
+      };
+
+      const fee = getProtocolFeeForEvent(createEvent, stream);
+      expect(fee).toBe("200000");
+
+      const tx = mapEventToTransaction(createEvent, stream, userAddress);
+      expect(tx.protocolFeeDeducted).toBe("200000");
+      expect(tx.grossAmountFormatted).toBe("1.0000000");
+      expect(tx.netAmount).toBe("0.9800000"); // 10000000 - 200000 = 9800000
+    });
+
+    it("derives fee from metadata if indexed FEE_COLLECTED event is not in stream events", () => {
+      const topUpEvent = {
+        eventType: "TOPPED_UP",
+        amount: "5000000",
+        transactionHash: "tx_topup_1",
+        timestamp: 1700000100n,
+        metadata: JSON.stringify({ feeAmount: "50000" }),
+      };
+
+      const stream = {
+        streamId: 42n,
+        sender: userAddress,
+        recipient: counterparty,
+        tokenAddress: "CDK...",
+        events: [topUpEvent],
+      };
+
+      const fee = getProtocolFeeForEvent(topUpEvent, stream);
+      expect(fee).toBe("50000");
+
+      const tx = mapEventToTransaction(topUpEvent, stream, userAddress);
+      expect(tx.protocolFeeDeducted).toBe("50000");
+      expect(tx.netAmount).toBe("0.4950000");
+    });
+  });
+
+  describe("Issue #1701: Terminate accounting export responses after streaming errors", () => {
+    function createMockResponse() {
+      const res: any = {
+        headersSent: false,
+        statusCode: 200,
+        headers: {},
+        writtenChunks: [] as string[],
+        ended: false,
+        destroyed: false,
+        destroyError: null as any,
+        setHeader: vi.fn((key: string, val: string) => {
+          res.headers[key] = val;
+        }),
+        status: vi.fn((code: number) => {
+          res.statusCode = code;
+          return res;
+        }),
+        json: vi.fn((data: any) => {
+          res.jsonData = data;
+          res.ended = true;
+          return res;
+        }),
+        write: vi.fn((chunk: string) => {
+          res.headersSent = true;
+          res.writtenChunks.push(chunk);
+          return true;
+        }),
+        end: vi.fn(() => {
+          res.ended = true;
+        }),
+        destroy: vi.fn((err?: any) => {
+          res.destroyed = true;
+          res.destroyError = err;
+        }),
+      };
+      return res;
+    }
+
+    it("terminates CSV export with res.destroy when database fails after headersSent", async () => {
+      const mockRes = createMockResponse();
+
+      // In the loop, findMany throws an error
+      prismaMock.stream.findMany.mockRejectedValueOnce(new Error("Database connection dropped mid-stream"));
+
+      await streamTransactionCSV(userAddress, { format: "csv", direction: "all" }, mockRes);
+
+      expect(mockRes.headersSent).toBe(true); // res.write(CSV_HEADERS) happened
+      expect(mockRes.destroyed).toBe(true);
+      expect(mockRes.destroyError?.message).toBe("Database connection dropped mid-stream");
+    });
+
+    it("terminates JSON export with res.destroy when database fails after headersSent", async () => {
+      const mockRes = createMockResponse();
+
+      prismaMock.stream.findMany.mockRejectedValueOnce(new Error("Database error mid-json"));
+
+      await streamTransactionJSON(userAddress, { format: "json", direction: "all" }, mockRes);
+
+      expect(mockRes.headersSent).toBe(true); // res.write('{"transactions":[') happened
+      expect(mockRes.destroyed).toBe(true);
+      expect(mockRes.destroyError?.message).toBe("Database error mid-json");
+    });
+
+    it("handles fallback to res.end() if res.destroy is unavailable", async () => {
+      const mockRes = createMockResponse();
+      delete mockRes.destroy;
+
+      prismaMock.stream.findMany.mockRejectedValueOnce(new Error("DB error"));
+
+      await streamTransactionCSV(userAddress, { format: "csv", direction: "all" }, mockRes);
+
+      expect(mockRes.headersSent).toBe(true);
+      expect(mockRes.ended).toBe(true);
+    });
+  });
+});