From 2429daaaccbea06d8b5ef0d489a4d1e449388b83 Mon Sep 17 00:00:00 2001 From: 0takuc0mrade Date: Wed, 21 Oct 2026 22:50:02 +0100 Subject: [PATCH 1/3] fix(webhooks): refresh signatures and count terminal failures --- backend/src/services/webhook.service.ts | 21 ++- backend/tests/webhook-delivery.test.ts | 187 ++++++++++++++++++++++++ 2 files changed, 202 insertions(+), 6 deletions(-) create mode 100644 backend/tests/webhook-delivery.test.ts diff --git a/backend/src/services/webhook.service.ts b/backend/src/services/webhook.service.ts index 78e2cf94..67e680b2 100644 --- a/backend/src/services/webhook.service.ts +++ b/backend/src/services/webhook.service.ts @@ -240,12 +240,6 @@ export async function deliverWebhook( const retryEnabled = options.retries !== false; const maxAttempts = retryEnabled ? MAX_RETRY_ATTEMPTS : 1; const payloadJson = JSON.stringify(payload); - const timestamp = Math.floor(Date.now() / 1000); - const signature = generateWebhookSignature( - payloadJson, - subscription.secretKey, - timestamp, - ); const deliveryId = crypto.randomUUID(); @@ -271,6 +265,14 @@ export async function deliverWebhook( ); } + // Sign at dispatch time so retries stay within the receiver's replay window. + const timestamp = Math.floor(Date.now() / 1000); + const signature = generateWebhookSignature( + payloadJson, + subscription.secretKey, + timestamp, + ); + const response = await fetch(subscription.targetUrl, { method: "POST", headers: { @@ -346,6 +348,13 @@ async function checkConsecutiveFailures(subscriptionId: string): Promise { const recentDeliveries = await prisma.webhookDelivery.findMany({ where: { subscriptionId, + eventType: { not: "webhook.test" }, + // Keep successes to reset the streak, but ignore rows still awaiting + // their final retry. Apply these filters before taking the latest 50. + OR: [ + { deliveredAt: { not: null } }, + { attempts: { gte: MAX_RETRY_ATTEMPTS }, error: { not: null } }, + ], }, orderBy: { createdAt: "desc", diff --git a/backend/tests/webhook-delivery.test.ts b/backend/tests/webhook-delivery.test.ts new file mode 100644 index 00000000..0afa8211 --- /dev/null +++ b/backend/tests/webhook-delivery.test.ts @@ -0,0 +1,187 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { prismaMock } = vi.hoisted(() => ({ + prismaMock: { + webhookSubscription: { update: vi.fn() }, + webhookDelivery: { create: vi.fn(), update: vi.fn(), findMany: vi.fn() }, + }, +})); + +vi.mock("../src/lib/prisma.js", () => ({ prisma: prismaMock })); +vi.mock("../src/logger.js", () => ({ + default: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import { + deliverWebhook, + verifyWebhookSignature, +} from "../src/services/webhook.service.js"; + +const SUBSCRIPTION = { + id: "wh_1", + userAddress: "G-OWNER", + targetUrl: "https://receiver.example/hook", + secretKey: "super-secret", + eventTypes: ["STREAM_CREATED"], + isActive: true, +}; +const PAYLOAD = { + eventType: "STREAM_CREATED", + data: { streamId: "1" }, + timestamp: "2026-10-21T12:00:00.000Z", +}; +const NOW = Date.parse(PAYLOAD.timestamp); + +interface Delivery { + id: string; + eventType: string; + attempts: number; + deliveredAt: Date | null; + error: string | null; +} + +function failedDeliveries(count: number, overrides: Partial = {}): Delivery[] { + return Array.from({ length: count }, (_, index) => ({ + id: `old_${index}`, + eventType: PAYLOAD.eventType, + attempts: 5, + deliveredAt: null, + error: "HTTP 503", + ...overrides, + })); +} + +describe("webhook delivery", () => { + const fetchMock = vi.fn(); + let history: Delivery[]; + + beforeEach(() => { + vi.resetAllMocks(); + vi.useFakeTimers(); + vi.setSystemTime(NOW); + vi.stubGlobal("fetch", fetchMock); + history = []; + + prismaMock.webhookDelivery.create.mockImplementation(async ({ data }) => { + const delivery = { id: "current", eventType: data.eventType, attempts: 0, deliveredAt: null, error: null }; + history.unshift(delivery); + return delivery; + }); + prismaMock.webhookDelivery.update.mockImplementation(async ({ where, data }) => { + const delivery = history.find((row) => row.id === where.id)!; + Object.assign(delivery, data); + return delivery; + }); + // Simulate Prisma filtering before LIMIT. Fixtures are newest first. + prismaMock.webhookDelivery.findMany.mockImplementation(async ({ where, take }) => + history.filter((row) => + (!where.eventType || row.eventType !== where.eventType.not) && + (!where.OR || where.OR.some((condition: { + deliveredAt?: { not: null }; + attempts?: { gte: number }; + error?: { not: null }; + }) => + (!condition.deliveredAt || row.deliveredAt !== null) && + (!condition.attempts || row.attempts >= condition.attempts.gte) && + (!condition.error || row.error !== null), + )), + ).slice(0, take), + ); + fetchMock.mockResolvedValue({ status: 503 }); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + async function failProductionDelivery() { + const result = deliverWebhook(SUBSCRIPTION, PAYLOAD); + await vi.runAllTimersAsync(); + return result; + } + + it("signs every attempt within the default replay window and keeps one delivery ID (#1694)", async () => { + const attempts: Array<{ timestamp: number; signature: string; id: string; body: string; valid: boolean }> = []; + fetchMock.mockImplementation(async (_url: string, init: RequestInit) => { + const headers = init.headers as Record; + const [, timestampText, signature] = /^t=(\d+),v1=([0-9a-f]+)$/.exec(headers["X-FlowFi-Signature"]!)!; + const timestamp = Number(timestampText); + const body = String(init.body); + attempts.push({ + timestamp, + signature: signature!, + id: headers["X-FlowFi-Delivery-ID"]!, + body, + valid: verifyWebhookSignature(body, signature!, SUBSCRIPTION.secretKey, timestamp), + }); + return { status: 503 }; + }); + + const result = await failProductionDelivery(); + + expect(result).toEqual({ success: false, status: 503, error: "HTTP 503" }); + expect(attempts.map((attempt) => attempt.timestamp - NOW / 1000)).toEqual([0, 60, 360, 1260, 4860]); + expect(attempts.map((attempt) => attempt.valid)).toEqual([true, true, true, true, true]); + expect(new Set(attempts.map((attempt) => attempt.signature)).size).toBe(5); + expect(new Set(attempts.map((attempt) => attempt.id)).size).toBe(1); + expect(attempts[0]!.id).toBeTruthy(); + expect(attempts.every((attempt) => attempt.body === JSON.stringify(PAYLOAD))).toBe(true); + expect(prismaMock.webhookDelivery.create).toHaveBeenCalledTimes(1); + }); + + it("does not disable a subscription because of 50 failed test pings (#1695)", async () => { + history = failedDeliveries(50, { eventType: "webhook.test", attempts: 1 }); + + await failProductionDelivery(); + + expect(prismaMock.webhookSubscription.update).not.toHaveBeenCalled(); + }); + + it("does not disable a subscription because of pending or retrying deliveries (#1695)", async () => { + history = [ + ...failedDeliveries(25, { attempts: 0, error: null }), + ...failedDeliveries(25, { attempts: 4 }), + ]; + + await failProductionDelivery(); + + expect(prismaMock.webhookSubscription.update).not.toHaveBeenCalled(); + }); + + it.each(["HTTP", "network"])("disables a subscription after 50 terminal %s production failures (#1695)", async (failure) => { + history = failedDeliveries(49); + if (failure === "network") fetchMock.mockRejectedValue(new Error("Network error")); + + await failProductionDelivery(); + + expect(prismaMock.webhookSubscription.update).toHaveBeenCalledExactlyOnceWith({ + where: { id: SUBSCRIPTION.id }, + data: { isActive: false }, + }); + }); + + it("ignores test pings and in-flight rows before limiting the failure history (#1695)", async () => { + history = [ + ...failedDeliveries(25, { eventType: "webhook.test", attempts: 1, deliveredAt: new Date(NOW), error: null }), + ...failedDeliveries(25, { attempts: 4 }), + ...failedDeliveries(49), + ]; + + await failProductionDelivery(); + + expect(prismaMock.webhookSubscription.update).toHaveBeenCalledTimes(1); + }); + + it("resets the failure streak at a successful production delivery (#1695)", async () => { + history = [ + ...failedDeliveries(48), + ...failedDeliveries(1, { deliveredAt: new Date(NOW), attempts: 1, error: null }), + ...failedDeliveries(50), + ]; + + await failProductionDelivery(); + + expect(prismaMock.webhookSubscription.update).not.toHaveBeenCalled(); + }); +}); From 4fb923cb145361eeff730270a40512c5f5797c72 Mon Sep 17 00:00:00 2001 From: 0takuc0mrade Date: Wed, 21 Oct 2026 22:50:02 +0100 Subject: [PATCH 2/3] fix(analytics): preserve fractional historical averages --- backend/src/services/analytics.service.ts | 3 +- backend/tests/analytics.test.ts | 41 +++++++++++++++++++++++ docs-site/docs/backend/analytics.md | 7 ++++ 3 files changed, 50 insertions(+), 1 deletion(-) diff --git a/backend/src/services/analytics.service.ts b/backend/src/services/analytics.service.ts index 3136793c..85418917 100644 --- a/backend/src/services/analytics.service.ts +++ b/backend/src/services/analytics.service.ts @@ -46,6 +46,7 @@ export interface TvlSnapshot { export interface HistoricalPoint { bucket: string; tokenAddress: string; + /** PostgreSQL NUMERIC text, preserving its full fractional scale without rounding. */ avgTvlStroops: string; velocityStroopsPerSecond: string; peakStreams: number; @@ -241,7 +242,7 @@ export async function getHistoricalAnalytics( points: seriesRows.map((row) => ({ bucket: new Date(row.bucket).toISOString(), tokenAddress: row.token_address, - avgTvlStroops: BigInt(row.avg_tvl).toString(), + avgTvlStroops: row.avg_tvl, velocityStroopsPerSecond: BigInt(row.aggregate_velocity).toString(), peakStreams: row.peak_streams, })), diff --git a/backend/tests/analytics.test.ts b/backend/tests/analytics.test.ts index d3701d02..f38e0a95 100644 --- a/backend/tests/analytics.test.ts +++ b/backend/tests/analytics.test.ts @@ -8,6 +8,8 @@ */ import { describe, it, expect, vi, beforeEach } from "vitest"; +import express from "express"; +import request from "supertest"; const queryMock = vi.fn(); @@ -27,6 +29,7 @@ vi.mock("../src/logger.js", () => ({ })); import { prisma } from "../src/lib/prisma.js"; +import analyticsRoutes from "../src/routes/v1/analytics.routes.js"; import { getDefiLlamaAdapter, getHistoricalAnalytics, @@ -34,6 +37,9 @@ import { resetTimescaleCacheForTests, } from "../src/services/analytics.service.js"; +const app = express(); +app.use("/v1/analytics", analyticsRoutes); + describe("analytics service (#1480)", () => { beforeEach(() => { vi.clearAllMocks(); @@ -150,4 +156,39 @@ describe("analytics service (#1480)", () => { expect(series.points[0]?.peakStreams).toBe(9); expect(queryMock.mock.calls.at(-1)![0]).toContain("daily_protocol_metrics"); }); + + it.each(["1h", "1d"] as const)( + "preserves PostgreSQL NUMERIC averages in %s historical responses (#1696)", + async (interval) => { + const averages = [ + "100.0000000000000000", + "1.5000000000000000", + "170141183460469231731687303715884105726.1234567890123456", + ]; + queryMock.mockResolvedValueOnce({ rows: [{ hypertable: "stream_flow_snapshots" }] }); + queryMock.mockResolvedValueOnce({ + rows: averages.map((avg_tvl, index) => ({ + bucket: new Date("2026-09-28T00:00:00Z"), + token_address: `TOKEN_${index}`, + avg_tvl, + aggregate_velocity: "4", + peak_streams: 9, + })), + }); + + const response = await request(app) + .get("/v1/analytics/historical") + .query({ period: "30d", interval }); + + expect(response.status).toBe(200); + expect(response.body.success).toBe(true); + expect(response.body.data.interval).toBe(interval); + expect(response.body.data.points.map((point: { avgTvlStroops: string }) => point.avgTvlStroops)) + .toEqual(averages); + expect(response.body.data.points[0].velocityStroopsPerSecond).toBe("4"); + expect(queryMock.mock.calls.at(-1)![0]).toContain( + interval === "1d" ? "daily_protocol_metrics" : "hourly_protocol_metrics", + ); + }, + ); }); diff --git a/docs-site/docs/backend/analytics.md b/docs-site/docs/backend/analytics.md index 19d5d1b6..8f5130a8 100644 --- a/docs-site/docs/backend/analytics.md +++ b/docs-site/docs/backend/analytics.md @@ -32,6 +32,13 @@ sequenceDiagram | `GET /api/v1/analytics/historical?period=30d&interval=1d` | `7d/30d/90d/1y` × `1h/1d` chart series | | `GET /api/v1/analytics/defillama` | DefiLlama adapter payload | +Historical `avgTvlStroops` values are decimal strings, preserving the full +precision and fractional scale returned by PostgreSQL `NUMERIC` for both +hourly and daily averages. No rounding or floating-point conversion is applied: +an average of 1.5 stroops may be returned as `"1.5000000000000000"`, and an +integer average as `"100.0000000000000000"`. Current TVL and velocity amounts +remain integer strings. + ## Degradation If the TimescaleDB extension or the hypertable is missing, `/tvl` falls back From f5ec5eef2f52755d697e77be2aad9eb4bf04f774 Mon Sep 17 00:00:00 2001 From: 0takuc0mrade Date: Wed, 21 Oct 2026 22:50:02 +0100 Subject: [PATCH 3/3] fix(graphql): aggregate unique accrued stream balances --- .../src/graphql/resolvers/stream.resolver.ts | 51 ++++++--- backend/tests/graphql.test.ts | 105 +++++++++++++++++- 2 files changed, 137 insertions(+), 19 deletions(-) diff --git a/backend/src/graphql/resolvers/stream.resolver.ts b/backend/src/graphql/resolvers/stream.resolver.ts index cc437ccf..3e400f91 100644 --- a/backend/src/graphql/resolvers/stream.resolver.ts +++ b/backend/src/graphql/resolvers/stream.resolver.ts @@ -6,6 +6,7 @@ import type { Prisma } from "../../generated/prisma/index.js"; import type { GraphQLContext } from "../context.js"; import type { TokenMetadata } from "../dataloaders/index.js"; +import { claimableAmountService } from "../../services/claimable.service.js"; export interface StreamFilterInput { status?: string | null; @@ -93,27 +94,45 @@ export async function resolveAccountOverview( ): Promise { // Amounts are persisted as decimal strings (i128 precision), so Prisma's // numeric `_sum` aggregate cannot be used; sum them in BigInt here instead. - const amountSelect = { depositedAmount: true, withdrawnAmount: true } as const; - const [sent, received, activeSent, activeReceived] = await Promise.all([ - ctx.prisma.stream.findMany({ where: { sender: address }, select: amountSelect }), - ctx.prisma.stream.findMany({ where: { recipient: address }, select: amountSelect }), - ctx.prisma.stream.count({ where: { sender: address, isActive: true } }), - ctx.prisma.stream.count({ where: { recipient: address, isActive: true } }), - ]); - - const sum = (rows: Array<{ depositedAmount: string; withdrawnAmount: string }>, key: "depositedAmount" | "withdrawnAmount"): bigint => - rows.reduce((total, row) => total + BigInt(row[key] || "0"), 0n); - - const all = [...sent, ...received]; - const deposited = sum(all, "depositedAmount"); - const withdrawn = sum(all, "withdrawnAmount"); - const claimable = deposited > withdrawn ? deposited - withdrawn : 0n; + // A single OR query returns self-directed streams once, even when both + // the sender and recipient match the wallet. + const streams = await ctx.prisma.stream.findMany({ + where: { OR: [{ sender: address }, { recipient: address }] }, + select: { + streamId: true, + recipient: true, + tokenAddress: true, + depositedAmount: true, + withdrawnAmount: true, + ratePerSecond: true, + startTime: true, + lastUpdateTime: true, + isActive: true, + isPaused: true, + pausedAt: true, + totalPausedDuration: true, + }, + }); + + let deposited = 0n; + let withdrawn = 0n; + let claimable = 0n; + let activeStreams = 0; + const now = Math.floor(Date.now() / 1000); + for (const stream of streams) { + deposited += BigInt(stream.depositedAmount); + withdrawn += BigInt(stream.withdrawnAmount); + if (stream.isActive) activeStreams++; + if (stream.recipient === address) { + claimable += BigInt(claimableAmountService.getClaimableAmount(stream, now).claimableAmount); + } + } return { totalDeposited: deposited.toString(), totalWithdrawn: withdrawn.toString(), totalClaimable: claimable.toString(), - activeStreams: activeSent + activeReceived, + activeStreams, }; } diff --git a/backend/tests/graphql.test.ts b/backend/tests/graphql.test.ts index 57cab3c6..1bc218c8 100644 --- a/backend/tests/graphql.test.ts +++ b/backend/tests/graphql.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; import request from "supertest"; const { prismaMock } = vi.hoisted(() => ({ @@ -47,18 +47,44 @@ vi.mock("../src/services/sorobanService.js", async (importOriginal) => ({ import app from "../src/app.js"; +const ADDRESS = "GABCDEF1234567890ABCDEF1234567890ABCDEF12345678"; +const NOW = 1_700_000_000; + +function stream(overrides = {}) { + return { + streamId: 1n, + sender: "G-OTHER", + recipient: ADDRESS, + tokenAddress: "TOKEN_A", + depositedAmount: "1000", + withdrawnAmount: "0", + ratePerSecond: "10", + startTime: BigInt(NOW - 10), + lastUpdateTime: BigInt(NOW - 10), + isActive: true, + isPaused: false, + pausedAt: null, + totalPausedDuration: 0, + ...overrides, + }; +} + describe("GraphQL API", () => { beforeEach(() => { vi.clearAllMocks(); // Account-overview queries select the amount columns; list queries do not. prismaMock.stream.findMany.mockImplementation(async (args: { select?: unknown }) => - args && args.select ? [{ depositedAmount: "100", withdrawnAmount: "25" }] : [], + args && args.select ? [stream({ depositedAmount: "100", withdrawnAmount: "25", isActive: false })] : [], ); prismaMock.streamEvent.findMany.mockResolvedValue([]); prismaMock.tokenRegistry.findMany.mockResolvedValue([]); prismaMock.stream.count.mockResolvedValue(0); }); + afterEach(() => { + vi.restoreAllMocks(); + }); + it("passes introspection queries and validates schema against TypeScript types", async () => { const response = await request(app) .post("/graphql") @@ -131,10 +157,83 @@ describe("GraphQL API", () => { expect(response.status).toBe(200); expect(response.body.errors).toBeUndefined(); - expect(response.body.data.accountOverview.totalDeposited).toBe("200"); + expect(response.body.data.accountOverview.totalDeposited).toBe("100"); expect(response.body.data.accountOverview.activeStreams).toBe(0); }); + it.each([ + { + name: "outgoing-only wallet", + rows: [stream({ sender: ADDRESS, recipient: "G-OTHER", withdrawnAmount: "25" })], + expected: { totalDeposited: "1000", totalWithdrawn: "25", totalClaimable: "0", activeStreams: 1 }, + }, + { + name: "unvested incoming stream", + rows: [stream({ startTime: BigInt(NOW + 60), lastUpdateTime: BigInt(NOW + 60) })], + expected: { totalDeposited: "1000", totalWithdrawn: "0", totalClaimable: "0", activeStreams: 1 }, + }, + { + name: "partially accrued incoming stream", + rows: [stream({ lastUpdateTime: BigInt(NOW - 30), withdrawnAmount: "100" })], + expected: { totalDeposited: "1000", totalWithdrawn: "100", totalClaimable: "300", activeStreams: 1 }, + }, + { + name: "self-directed stream counted once", + rows: [stream({ sender: ADDRESS, withdrawnAmount: "20" })], + expected: { totalDeposited: "1000", totalWithdrawn: "20", totalClaimable: "100", activeStreams: 1 }, + }, + { + name: "paused incoming stream frozen at pause time", + rows: [stream({ isPaused: true, pausedAt: BigInt(NOW - 6) })], + expected: { totalDeposited: "1000", totalWithdrawn: "0", totalClaimable: "40", activeStreams: 1 }, + }, + { + name: "inactive incoming stream", + rows: [stream({ isActive: false, withdrawnAmount: "20" })], + expected: { totalDeposited: "1000", totalWithdrawn: "20", totalClaimable: "0", activeStreams: 0 }, + }, + { + name: "claimable capped by the remaining deposit", + rows: [stream({ withdrawnAmount: "975" })], + expected: { totalDeposited: "1000", totalWithdrawn: "975", totalClaimable: "25", activeStreams: 1 }, + }, + { + name: "mixed directions and states", + rows: [ + stream({ sender: ADDRESS, withdrawnAmount: "20" }), + stream({ streamId: 2n, sender: ADDRESS, recipient: "G-OTHER", depositedAmount: "2000", withdrawnAmount: "100" }), + stream({ streamId: 3n, isPaused: true, pausedAt: BigInt(NOW - 6) }), + stream({ streamId: 4n, isActive: false, withdrawnAmount: "20" }), + ], + expected: { totalDeposited: "5000", totalWithdrawn: "140", totalClaimable: "140", activeStreams: 3 }, + }, + { + name: "empty wallet", + rows: [], + expected: { totalDeposited: "0", totalWithdrawn: "0", totalClaimable: "0", activeStreams: 0 }, + }, + ])("returns correct accountOverview totals for $name (#1697)", async ({ rows, expected }) => { + vi.spyOn(Date, "now").mockReturnValue(NOW * 1000); + type AddressFilter = { sender?: string; recipient?: string }; + const matches = (row: ReturnType, filter: AddressFilter) => + (!filter.sender || row.sender === filter.sender) && + (!filter.recipient || row.recipient === filter.recipient); + prismaMock.stream.findMany.mockImplementation(async ({ where }: { where: AddressFilter & { OR?: AddressFilter[] } }) => + rows.filter((row) => where.OR ? where.OR.some((filter) => matches(row, filter)) : matches(row, where)), + ); + prismaMock.stream.count.mockImplementation(async ({ where }: { where: AddressFilter }) => + rows.filter((row) => row.isActive && matches(row, where)).length, + ); + + const response = await request(app).post("/graphql").send({ + query: `query { accountOverview(address: "${ADDRESS}") { totalDeposited totalWithdrawn totalClaimable activeStreams } }`, + }); + + expect(response.status).toBe(200); + expect(response.body.errors).toBeUndefined(); + expect(response.body.data.accountOverview).toEqual(expected); + }); + it("rejects subscription operations executed over HTTP POST", async () => { const response = await request(app) .post("/graphql")