fix(backend): repair webhook delivery, analytics, and account balances

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<AccountOverview> {
   // 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,
   };
 }
 

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,
     })),

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<void> {
   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",

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,13 +29,17 @@ vi.mock("../src/logger.js", () => ({
 }));
 
 import { prisma } from "../src/lib/prisma.js";
+import analyticsRoutes from "../src/routes/v1/analytics.routes.js";
 import {
   getDefiLlamaAdapter,
   getHistoricalAnalytics,
   getProtocolTvl,
   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",
+      );
+    },
+  );
 });

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<typeof stream>, 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")