diff --git a/backend/src/graphql/resolvers/stream.resolver.ts b/backend/src/graphql/resolvers/stream.resolver.ts index 3e400f91..bfbe293c 100644 --- a/backend/src/graphql/resolvers/stream.resolver.ts +++ b/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 { - 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; diff --git a/backend/src/graphql/schema.ts b/backend/src/graphql/schema.ts index b90ab067..b57a4979 100644 --- a/backend/src/graphql/schema.ts +++ b/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."); }, diff --git a/backend/src/services/export.service.ts b/backend/src/services/export.service.ts index 7199f682..05f6f6b8 100644 --- a/backend/src/services/export.service.ts +++ b/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(); + } } } } diff --git a/backend/tests/export.service.test.ts b/backend/tests/export.service.test.ts new file mode 100644 index 00000000..49682ce6 --- /dev/null +++ b/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); + }); + }); +}); diff --git a/backend/tests/graphql.test.ts b/backend/tests/graphql.test.ts index 1bc218c8..cbb47d3d 100644 --- a/backend/tests/graphql.test.ts +++ b/backend/tests/graphql.test.ts @@ -267,4 +267,89 @@ describe("GraphQL API", () => { expect(response.status).toBe(200); expect(Array.isArray(response.body.errors)).toBe(true); }); + + describe("Issue #1698: resolveStreams status filtering without collapsing inactive states", () => { + it.each([ + { + status: "active", + expectedWhere: { isActive: true, isPaused: false }, + }, + { + status: "paused", + expectedWhere: { isPaused: true }, + }, + { + status: "cancelled", + expectedWhere: { isActive: false, events: { some: { eventType: "CANCELLED" } } }, + }, + { + status: "completed", + expectedWhere: { isActive: false, events: { some: { eventType: "COMPLETED" } } }, + }, + ])("maps status '$status' to distinct where condition", async ({ status, expectedWhere }) => { + let capturedWhere: any = null; + prismaMock.stream.findMany.mockImplementation(async ({ where }: { where: any }) => { + capturedWhere = where; + return []; + }); + + const response = await request(app).post("/graphql").send({ + query: `query { streams(filter: { status: "${status}" }) { edges { node { id } } } }`, + }); + + expect(response.status).toBe(200); + expect(response.body.errors).toBeUndefined(); + expect(capturedWhere).toMatchObject(expectedWhere); + }); + + it("rejects unknown status values", async () => { + const response = await request(app).post("/graphql").send({ + query: 'query { streams(filter: { status: "unknown_status" }) { edges { node { id } } } }', + }); + + expect(response.status).toBe(200); + expect(Array.isArray(response.body.errors)).toBe(true); + expect(response.body.errors[0].message).toContain("Invalid status filter"); + }); + }); + + describe("Issue #1699: Accept full-width stream identifiers in GraphQL schema", () => { + it("queries stream with full-width u64 ID (> 2^31 - 1)", async () => { + const largeStreamId = "9223372036854775807"; // 2^63 - 1 + prismaMock.stream.findUnique.mockResolvedValueOnce({ + id: "stream-large", + streamId: BigInt(largeStreamId), + sender: ADDRESS, + recipient: "G-RECIPIENT", + tokenAddress: "TOKEN_A", + ratePerSecond: "10", + depositedAmount: "1000", + withdrawnAmount: "0", + startTime: BigInt(NOW), + endTime: null, + isActive: true, + isPaused: false, + }); + + const response = await request(app).post("/graphql").send({ + query: `query { stream(streamId: "${largeStreamId}") { id streamId sender recipient } }`, + }); + + expect(response.status).toBe(200); + expect(response.body.errors).toBeUndefined(); + expect(response.body.data.stream).not.toBeNull(); + expect(response.body.data.stream.streamId).toBe(largeStreamId); + }); + + it("rejects out-of-range stream IDs", async () => { + const negativeStreamId = "-1"; + const response = await request(app).post("/graphql").send({ + query: `query { stream(streamId: "${negativeStreamId}") { id streamId } }`, + }); + + expect(response.status).toBe(200); + expect(Array.isArray(response.body.errors)).toBe(true); + expect(response.body.errors[0].message).toContain("out of supported range"); + }); + }); });