Implement #1465-#1466 features: fee splits, GraphQL API, position rec…

backend/package.json

@@ -37,14 +37,16 @@
     "@opentelemetry/sdk-trace-node": "^2.11.0",
     "@opentelemetry/semantic-conventions": "^1.43.0",
     "@prisma/adapter-pg": "^7.8.0",
-    "@prisma/client": "^7.8.0",
+    "@prisma/client": "^7.10.1",
     "@prisma/extension-read-replicas": "^0.5.0",
     "@stellar/stellar-sdk": "^17.0.1",
     "cors": "^2.8.6",
+    "dataloader": "^2.2.3",
     "dotenv": "^17.4.2",
     "express": "^5.2.1",
     "express-rate-limit": "^8.5.2",
     "ioredis": "^5.11.1",
+    "nexus": "^1.3.0",
     "pg": "^8.21.0",
     "prom-client": "^15.1.3",
     "swagger-jsdoc": "^6.3.0",

backend/src/app.ts

@@ -20,6 +20,9 @@ import { bigIntSafeJsonMiddleware } from "./lib/serialize.js";
 import v1Routes from "./routes/v1/index.js";
 import healthRoutes from "./routes/health.routes.js";
 import metricsRoutes from "./routes/metrics.routes.js";
+import { graphqlRoutes } from "./graphql/routes.js";
+
+// Mount at POST /graphql and GET /graphql (GraphiQL playground in non-production).
 
 const app = express();
 
@@ -156,6 +159,10 @@ app.use("/health", healthRateLimiter, healthRoutes);
 // default. Access control lives in the router (see metrics.routes.ts).
 app.use("/metrics", metricsRoutes);
 
+// GraphQL API endpoint. GET serves the playground in non-production; POST
+// executes queries/mutations with a per-request DataLoader context.
+app.use("/graphql", graphqlRoutes);
+
 /**
  * @openapi
  * /:

backend/src/graphql/context.ts

@@ -0,0 +1,19 @@
+import type { PrismaClient } from "../generated/prisma/index.js";
+import { createDataLoaders, type DataLoaders } from "./dataloaders/index.js";
+
+/**
+ * Execution context shared by every GraphQL resolver. A fresh context (and a
+ * fresh set of DataLoaders) is created per request so batching never leaks
+ * across requests.
+ */
+export interface GraphQLContext {
+  prisma: PrismaClient;
+  loaders: DataLoaders;
+}
+
+export function createContext(prisma: PrismaClient): GraphQLContext {
+  return {
+    prisma,
+    loaders: createDataLoaders(prisma),
+  };
+}

backend/src/graphql/dataloaders/index.ts

@@ -0,0 +1,100 @@
+// DataLoader batching helpers for GraphQL queries.
+//
+// Each loader collapses the N+1 lookups produced by nested fields into a single
+// indexed SQL query per request (e.g. `WHERE streamId IN (...)`), then hands the
+// results back to the caller in the exact order the keys were requested.
+
+import DataLoader from "dataloader";
+import type { PrismaClient, Stream, StreamEvent, TokenRegistry } from "../../generated/prisma/index.js";
+
+export interface TokenMetadata {
+  address: string;
+  symbol: string;
+  name: string;
+  decimals: number;
+  iconUrl: string | null;
+}
+
+export interface DataLoaders {
+  stream: DataLoader<bigint, Stream | null>;
+  eventsByStream: DataLoader<bigint, StreamEvent[]>;
+  token: DataLoader<string, TokenMetadata | null>;
+}
+
+/**
+ * Batch stream lookups into a single SQL `WHERE streamId IN (...)` query.
+ */
+export async function batchStreamsByIds(
+  prisma: PrismaClient,
+  streamIds: readonly bigint[],
+): Promise<(Stream | null)[]> {
+  const rows = await prisma.stream.findMany({
+    where: { streamId: { in: [...streamIds] } },
+  });
+  const byId = new Map<bigint, Stream>(rows.map((row) => [row.streamId, row]));
+  return streamIds.map((id) => byId.get(id) ?? null);
+}
+
+/**
+ * Group events for multiple streams in a single indexed query, preserving the
+ * requested key order.
+ */
+export async function batchEventsByStreamIds(
+  prisma: PrismaClient,
+  streamIds: readonly bigint[],
+): Promise<StreamEvent[][]> {
+  const rows = await prisma.streamEvent.findMany({
+    where: { streamId: { in: [...streamIds] } },
+    orderBy: { timestamp: "desc" },
+  });
+  const byStream = new Map<bigint, StreamEvent[]>();
+  for (const row of rows) {
+    const bucket = byStream.get(row.streamId);
+    if (bucket) {
+      bucket.push(row);
+    } else {
+      byStream.set(row.streamId, [row]);
+    }
+  }
+  return streamIds.map((id) => byStream.get(id) ?? []);
+}
+
+/**
+ * Resolve token metadata for many contract addresses in one query.
+ */
+export async function batchTokensByAddresses(
+  prisma: PrismaClient,
+  addresses: readonly string[],
+): Promise<(TokenMetadata | null)[]> {
+  const rows = await prisma.tokenRegistry.findMany({
+    where: { contractAddress: { in: [...addresses] } },
+  });
+  const byAddress = new Map<string, TokenRegistry>(rows.map((row) => [row.contractAddress, row]));
+  return addresses.map((address) => {
+    const row = byAddress.get(address);
+    return row
+      ? {
+          address: row.contractAddress,
+          symbol: row.symbol,
+          name: row.name,
+          decimals: row.decimals,
+          iconUrl: row.iconUrl,
+        }
+      : null;
+  });
+}
+
+/**
+ * Create a per-request set of DataLoaders bound to the given Prisma client.
+ */
+export function createDataLoaders(prisma: PrismaClient): DataLoaders {
+  return {
+    stream: new DataLoader<bigint, Stream | null>((ids) => batchStreamsByIds(prisma, ids)),
+    eventsByStream: new DataLoader<bigint, StreamEvent[]>((ids) =>
+      batchEventsByStreamIds(prisma, ids),
+    ),
+    token: new DataLoader<string, TokenMetadata | null>((addresses) =>
+      batchTokensByAddresses(prisma, addresses),
+    ),
+  };
+}

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

@@ -0,0 +1,141 @@
+// Resolver implementations for the GraphQL Stream schema.
+//
+// These are deliberately framework-agnostic: `schema.ts` wires them into Nexus
+// fields, which keeps the data-access logic unit-testable on its own.
+
+import type { Prisma } from "../../generated/prisma/index.js";
+import type { GraphQLContext } from "../context.js";
+import type { TokenMetadata } from "../dataloaders/index.js";
+
+export interface StreamFilterInput {
+  status?: string | null;
+  tokenAddress?: string | null;
+  sender?: string | null;
+  recipient?: string | null;
+}
+
+export interface StreamsQueryArgs {
+  first?: number | null;
+  after?: string | null;
+  filter?: StreamFilterInput | null;
+}
+
+export interface StreamConnection {
+  edges: Array<{ node: unknown; cursor: string }>;
+  pageInfo: { hasNextPage: boolean; endCursor: string | null };
+}
+
+/** Normalise Prisma's `bigint` values to the string form exposed by GraphQL. */
+export function bigintToString(value: bigint | null | undefined): string | null {
+  return value === null || value === undefined ? null : value.toString();
+}
+
+function parseCursor(cursor: string | null | undefined): bigint | undefined {
+  if (cursor === null || cursor === undefined || cursor === "") return undefined;
+  try {
+    return BigInt(cursor);
+  } catch {
+    return undefined;
+  }
+}
+
+export async function resolveStream(
+  ctx: GraphQLContext,
+  streamId: number | string,
+): Promise<unknown> {
+  const id = BigInt(streamId);
+  return ctx.loaders.stream.load(id);
+}
+
+export async function resolveStreams(
+  ctx: GraphQLContext,
+  args: StreamsQueryArgs,
+): Promise<StreamConnection> {
+  const take = args.first ?? 20;
+  const after = parseCursor(args.after);
+  const filter = args.filter ?? {};
+
+  const where: Prisma.StreamWhereInput = {};
+  if (filter.status) where.isActive = filter.status.toLowerCase() === "active";
+  if (filter.tokenAddress) where.tokenAddress = filter.tokenAddress;
+  if (filter.sender) where.sender = filter.sender;
+  if (filter.recipient) where.recipient = filter.recipient;
+
+  const rows = await ctx.prisma.stream.findMany({
+    where,
+    take: take + 1,
+    orderBy: { streamId: "desc" },
+    ...(after !== undefined ? { cursor: { streamId: after }, skip: 1 } : {}),
+  });
+
+  const hasNextPage = rows.length > take;
+  const page = hasNextPage ? rows.slice(0, take) : rows;
+
+  return {
+    edges: page.map((node) => ({ node, cursor: node.streamId.toString() })),
+    pageInfo: {
+      hasNextPage,
+      endCursor: page.length > 0 ? page[page.length - 1]!.streamId.toString() : null,
+    },
+  };
+}
+
+export interface AccountOverview {
+  totalDeposited: string;
+  totalWithdrawn: string;
+  totalClaimable: string;
+  activeStreams: number;
+}
+
+export async function resolveAccountOverview(
+  ctx: GraphQLContext,
+  address: string,
+): 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;
+
+  return {
+    totalDeposited: deposited.toString(),
+    totalWithdrawn: withdrawn.toString(),
+    totalClaimable: claimable.toString(),
+    activeStreams: activeSent + activeReceived,
+  };
+}
+
+export async function resolveStreamEvents(
+  ctx: GraphQLContext,
+  streamId: bigint,
+): Promise<unknown[]> {
+  return ctx.loaders.eventsByStream.load(streamId);
+}
+
+export async function resolveStreamToken(
+  ctx: GraphQLContext,
+  tokenAddress: string,
+): Promise<TokenMetadata | null> {
+  return ctx.loaders.token.load(tokenAddress);
+}
+
+/**
+ * Position receipts are minted on-chain (soulbound position NFTs). The backend
+ * indexer does not persist them yet, so this resolver returns an empty list
+ * rather than issuing an unbatched per-stream query.
+ */
+export async function resolveStreamPositions(): Promise<never[]> {
+  return [];
+}

backend/src/graphql/routes.ts

@@ -0,0 +1,78 @@
+import express from "express";
+// Import the CommonJS build explicitly so it shares the single `graphql`
+// instance that `nexus` loads internally. Importing the ESM entry here creates
+// a second module realm and GraphQL rejects schemas built by the other one.
+import { graphql as executeGraphQL } from "graphql/index.js";
+import { prisma } from "../lib/prisma.js";
+import { createContext } from "./context.js";
+import { schema } from "./schema.js";
+
+export const graphqlRoutes = express.Router();
+
+const PLAYGROUND_HTML = `<!DOCTYPE html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8" />
+    <title>FlowFi GraphQL</title>
+    <style>
+      body { font-family: system-ui, sans-serif; margin: 2rem; }
+      textarea { width: 100%; height: 12rem; font-family: monospace; }
+      pre { background: #f5f5f5; padding: 1rem; overflow: auto; }
+    </style>
+  </head>
+  <body>
+    <h1>FlowFi GraphQL</h1>
+    <textarea id="query">{ __schema { types { name } } }</textarea>
+    <button id="run">Run</button>
+    <pre id="result"></pre>
+    <script>
+      document.getElementById("run").addEventListener("click", async () => {
+        const response = await fetch("/graphql", {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({ query: document.getElementById("query").value }),
+        });
+        document.getElementById("result").textContent = JSON.stringify(
+          await response.json(),
+          null,
+          2,
+        );
+      });
+    </script>
+  </body>
+</html>`;
+
+async function execute(request: express.Request, response: express.Response): Promise<void> {
+  const body = request.body ?? {};
+  try {
+    const result = await executeGraphQL({
+      schema,
+      source: typeof body.query === "string" ? body.query : "",
+      variableValues: body.variables ?? null,
+      operationName: body.operationName ?? undefined,
+      contextValue: createContext(prisma),
+    });
+    response.status(200).json(result);
+  } catch (error) {
+    response.status(200).json({
+      errors: [{ message: error instanceof Error ? error.message : "GraphQL execution failed" }],
+    });
+  }
+}
+
+graphqlRoutes.post("/", execute);
+
+graphqlRoutes.get("/", (request: express.Request, response: express.Response) => {
+  if (process.env.NODE_ENV === "production") {
+    response.status(200).json({
+      data: { message: "FlowFi GraphQL endpoint. Send queries via POST /graphql." },
+    });
+    return;
+  }
+  // The global CSP forbids inline scripts; relax it for the local playground only.
+  response.setHeader(
+    "Content-Security-Policy",
+    "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'",
+  );
+  response.type("html").send(PLAYGROUND_HTML);
+});