From 10e82e5a435079298bdb5b1a636f43cba50a5b57 Mon Sep 17 00:00:00 2001 From: oshowunm Date: Sat, 29 Aug 2026 22:02:12 +0100 Subject: [PATCH 1/3] docs(backend): add ARCHITECTURE.md with indexer ownership, SSE flow, and keeper-key model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both indexerService.ts and soroban-indexer.service.ts reference "docs/ARCHITECTURE.md for the full indexer ownership model" but no such file existed in backend/docs/. This created a broken cross- reference for contributors investigating the dual-indexer race (#801) and the keeper-key authorization model. This adds backend/docs/ARCHITECTURE.md documenting: - Indexer ownership: which of the three similarly-named files is authoritative, the dual-indexer race, and the phase-out plan - SSE broadcast flow: end-to-end path from Soroban RPC through the worker, DB, SSE service, Redis fanout, to the frontend - Keeper-key authorization: the custodial vs non-custodial signing model, the KEEPER_SECRET_KEY role, and the security boundary Closes #1299 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- backend/docs/ARCHITECTURE.md | 191 +++++++++++++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 backend/docs/ARCHITECTURE.md diff --git a/backend/docs/ARCHITECTURE.md b/backend/docs/ARCHITECTURE.md new file mode 100644 index 00000000..325fba7e --- /dev/null +++ b/backend/docs/ARCHITECTURE.md @@ -0,0 +1,191 @@ +# Backend Architecture + +This document is the canonical reference for backend service architecture. It is +referenced by `backend/src/services/indexerService.ts` and +`backend/src/services/soroban-indexer.service.ts` as the authoritative source for +indexer ownership, SSE broadcast flow, and keeper-key authorization. + +For the full project-wide architecture (event type data flows, pause/resume +timing, environment variables, and operational runbook) see +[`docs/ARCHITECTURE.md`](../../docs/ARCHITECTURE.md). + +--- + +## Indexer Ownership Model + +Three files with overlapping names handle indexing and indexer management. Only +one of them is the source of truth for stream state. + +| File | Role | Status | +|------|------|--------| +| `src/workers/soroban-event-worker.ts` (`SorobanEventWorker`) | **Source-of-truth indexer.** Polls Soroban RPC, decodes XDR events, persists `Stream` / `StreamEvent` rows, advances the `IndexerState` cursor, and broadcasts SSE updates. | **Active / source of truth.** Started by `src/workers/index.ts`. | +| `src/services/soroban-indexer.service.ts` (`SorobanIndexerService`) | **Legacy indexer being phased out.** A simpler duplicate poller that writes to the same DB rows and races with the worker on the same `Stream` / `StreamEvent` records (issue #801). | **Legacy — do not extend.** Removal tracked with functional consolidation (issue #801). Started directly from `src/index.ts`. | +| `src/services/indexerService.ts` | **Not an indexer at all.** Admin control-plane helpers (`getIndexerStatus`, `resetIndexer`, `replayFromLedger`) that read/reset the shared `IndexerState` cursor row and trigger the worker's poll loop. | **Active.** Name is misleading; kept alongside the legacy indexer above. | + +### Key Rules + +1. **When debugging indexing, read `src/workers/soroban-event-worker.ts` + first.** It is the only file that persists canonical stream state. +2. **Do not add new behavior to `soroban-indexer.service.ts`.** It exists only + for backwards compatibility while the double-indexer race (issue #801) is + consolidated. Mirror any changes in `SorobanEventWorker` instead. +3. **`indexerService.ts` is control-plane only** — it never reads the chain; it + manages the shared cursor and triggers replays. + +### The Dual-Indexer Race + +Both `SorobanEventWorker` and `SorobanIndexerService` poll the same Soroban RPC +for the same contract events and write to the same `Stream` and `StreamEvent` +rows. Because they run on independent timers, they can race: + +- Both may process the same ledger simultaneously. +- Both write to the same `Stream` row (upsert), so the last writer wins — + usually harmless for immutable fields but problematic for additive mutations + like `withdrawnAmount` (issue #808). +- `StreamEvent` dedup via `@@unique([transactionHash, eventType])` prevents + duplicate event rows, but does **not** protect stream state mutations. + +**Mitigation:** Do not extend the legacy indexer. The consolidation (issue #801) +will remove `SorobanIndexerService` entirely. + +### Naming Convention Plan + +The team convention is kebab-case with a `.service.ts` suffix. Once functional +consolidation lands: + +| Current Name | Expected Future Name | +|---|---| +| `indexerService.ts` | `indexer.service.ts` | +| `soroban-indexer.service.ts` | *(removed)* | + +--- + +## SSE Broadcast Flow + +The SSE (Server-Sent Events) subsystem delivers real-time contract event +notifications to connected frontend clients. The full SSE architecture (scaling, +memory, security) is documented in +[`docs/SSE_ARCHITECTURE.md`](./SSE_ARCHITECTURE.md). + +### End-to-End Path + +``` +Soroban RPC + │ SorobanEventWorker polls for new contract events + ▼ +SorobanEventWorker (src/workers/soroban-event-worker.ts) + │ decode XDR → upsert Stream → insert StreamEvent + ▼ +PostgreSQL (via Prisma) + │ Stream + StreamEvent rows updated + ▼ +SSE broadcast (src/services/sse.service.ts) + │ sseService.broadcastToStream(streamId, event, data) + │ sseService.broadcastToUser(publicKey, event, data) + │ sseService.broadcastToAdmin(event, data) + │ + ├──► [Single instance] Direct write to in-memory client registry + │ + └──► [Multi-instance] Redis Pub/Sub + │ publish to sse:stream:, sse:user:
+ ▼ + All backend instances subscribe + │ rebroadcast to local connected clients + ▼ + Frontend (useStreamEvents hook) +``` + +### Broadcast Channels + +The worker uses three broadcast entry points depending on the event: + +| Method | When Used | Target Audience | +|--------|-----------|-----------------| +| `sseService.broadcastToStream(streamId, event, data)` | Stream lifecycle events (created, topped_up, withdrawn, cancelled, completed, paused, resumed) | Clients subscribed to that specific stream ID or `*` | +| `sseService.broadcastToUser(publicKey, event, data)` | (Reserved for user-scoped events) | Clients subscribed to `user:` or `*` | +| `sseService.broadcastToAdmin(event, data)` | Protocol-level events (fee_collected, fee_config_updated, admin_transferred) | The admin user identified by `ADMIN_PUBLIC_KEY` env var | + +### Multi-Instance Fanout + +When `REDIS_URL` is configured, broadcasts go through Redis Pub/Sub instead of +direct in-memory writes: + +1. The originating instance publishes `{ event, data }` to + `sse:stream:` or `sse:user:
`. +2. Every backend instance subscribes via `psubscribe('sse:stream:*', + 'sse:user:*')` and rebroadcasts to its own local clients. +3. This means events reach all connected clients regardless of which backend + instance they are connected to. + +### Client Connection Limits + +| Limit | Default | Env Var | +|-------|---------|---------| +| Max SSE connections per server | 10,000 | `MAX_SSE_CONNECTIONS` | +| Max connections per IP | 5 | Hardcoded | +| Max connections per authenticated user | 10 | Hardcoded | + +Slow clients (write buffer ≥ 64 KB) are automatically dropped to protect +throughput for healthy clients. + +--- + +## Keeper-Key Authorization Model + +FlowFi splits transaction signing into two categories: custodial (server-signed) +and non-custodial (wallet-signed). The signing key determines who is responsible +for the transaction. + +### Action Signing Matrix + +| Action | Signer | Mechanism | +|--------|--------|-----------| +| **Top-up** | Server (custodial) | Backend submits the transaction using `KEEPER_SECRET_KEY`. The frontend sends only the stream ID and amount. | +| **Withdraw** | Wallet (non-custodial) | Frontend builds and signs the transaction via the connected wallet (Freighter). The backend simulate endpoint exists for fee estimation only. | +| **Pause / Resume** | Wallet (non-custodial) | Same as withdraw — frontend-signed. Backend simulate endpoints exist for fee estimation but do not submit. | +| **Create stream** | Wallet (non-custodial) | Frontend signs via wallet and submits directly to the Soroban RPC. | + +### The `KEEPER_SECRET_KEY` + +- Stored as an environment variable on the backend. +- Loaded by `src/services/sorobanService.ts` via + `process.env.KEEPER_SECRET_KEY`. +- Used **exclusively** by the top-up flow. The `topUpStream` function builds + the transaction, signs it with the keeper keypair, and submits it to the + Soroban RPC. +- If `KEEPER_SECRET_KEY` is not configured, `topUpStream` throws + `'KEEPER_SECRET_KEY not configured'` and the request returns HTTP 500. +- The cancel endpoint (`src/controllers/stream/cancel.ts`) also reads + `KEEPER_SECRET_KEY` but only to check whether the server wallet is configured; + the actual cancel transaction is wallet-signed by the sender. + +### Security Boundary + +> **Do not wire pause/resume/withdraw to a server-side submit path.** Only +> `top-up` is intentionally custodial. All other mutating actions must be +> wallet-signed by the user to preserve the non-custodial security model. + +The keeper key is a server-side secret and is never exposed to the frontend. +It lives exclusively in the backend's environment configuration. + +--- + +## Database Models + +For a quick reference of the models involved in indexing: + +| Model | Key Fields | Purpose | +|-------|------------|---------| +| `User` | `publicKey` | Stellar wallet addresses | +| `Stream` | `streamId`, `sender`, `recipient`, `ratePerSecond`, `depositedAmount`, `withdrawnAmount`, `isActive` | Mirrors on-chain stream state | +| `StreamEvent` | `streamId`, `eventType`, `transactionHash`, `ledgerSequence`, `timestamp` | Indexed on-chain events; unique on `(transactionHash, eventType)` | +| `IndexerState` | `lastLedger`, `lastCursor` | Cursor for last successfully indexed ledger sequence | + +--- + +## Related Documentation + +- [Root Architecture](../../docs/ARCHITECTURE.md) — full project-wide architecture +- [SSE Architecture](./SSE_ARCHITECTURE.md) — SSE scaling, security, operational runbook +- [SSE Implementation](./SSE_IMPLEMENTATION.md) — client integration guide +- [Authentication](./AUTHENTICATION.md) — SEP-10 + JWT auth flow From 59cb70a4602bdbc7b755f7ce4d708caec7f22e74 Mon Sep 17 00:00:00 2001 From: oshowunm Date: Sat, 29 Aug 2026 22:29:50 +0100 Subject: [PATCH 2/3] test(frontend): add unit tests for connectFreighter error paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds tests for the three untested error branches in connectFreighter(): 1. FreighterNotInstalledError when isConnected returns false 2. getAddress failure with no address or explicit error 3. getNetworkDetails catch fallback Closes #1287 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- frontend/src/lib/wallet.test.ts | 211 ++++++++++++++++++++++++++++++++ 1 file changed, 211 insertions(+) create mode 100644 frontend/src/lib/wallet.test.ts diff --git a/frontend/src/lib/wallet.test.ts b/frontend/src/lib/wallet.test.ts new file mode 100644 index 00000000..7729126b --- /dev/null +++ b/frontend/src/lib/wallet.test.ts @@ -0,0 +1,211 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +// ── Mocks ──────────────────────────────────────────────────────────────────── + +const mockIsConnected = vi.fn(); +const mockSetAllowed = vi.fn(); +const mockGetAddress = vi.fn(); +const mockGetNetworkDetails = vi.fn(); + +vi.mock("@stellar/freighter-api", () => ({ + isConnected: (...args: unknown[]) => mockIsConnected(...args), + setAllowed: (...args: unknown[]) => mockSetAllowed(...args), + getAddress: (...args: unknown[]) => mockGetAddress(...args), + getNetworkDetails: (...args: unknown[]) => mockGetNetworkDetails(...args), +})); + +// ── Imports (after mocks) ──────────────────────────────────────────────────── + +import { + connectWallet, + FreighterNotInstalledError, + type WalletSession, +} from "./wallet"; + +// ── Helpers ────────────────────────────────────────────────────────────────── + +/** Default happy-path stubs so individual tests only override what they need. */ +function stubHappyPath(overrides?: { + connected?: boolean; + address?: string; + addressError?: string; + networkPassphrase?: string; + networkDetailsError?: string; +}) { + mockIsConnected.mockResolvedValue({ + isConnected: overrides?.connected ?? true, + }); + + mockSetAllowed.mockResolvedValue(undefined); + + mockGetAddress.mockResolvedValue({ + address: "address" in (overrides ?? {}) ? overrides!.address : "GABCDEF1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ", + error: overrides?.addressError ?? undefined, + }); + + mockGetNetworkDetails.mockResolvedValue({ + networkPassphrase: + overrides?.networkPassphrase ?? "Test SDF Network ; September 2015", + error: overrides?.networkDetailsError ?? undefined, + }); +} + +// ── Tests ──────────────────────────────────────────────────────────────────── + +describe("connectWallet → connectFreighter error paths", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + // ── 1. FreighterNotInstalledError ──────────────────────────────────────── + + describe("Freighter extension not installed", () => { + it("throws FreighterNotInstalledError when isConnected returns false", async () => { + stubHappyPath({ connected: false }); + + await expect(connectWallet("freighter")).rejects.toThrow( + FreighterNotInstalledError, + ); + }); + + it("FreighterNotInstalledError has the correct name and message", async () => { + stubHappyPath({ connected: false }); + + try { + await connectWallet("freighter"); + throw new Error("expected connectWallet to throw"); + } catch (error) { + expect(error).toBeInstanceOf(FreighterNotInstalledError); + expect((error as Error).name).toBe("FreighterNotInstalledError"); + expect((error as Error).message).toContain("not installed"); + } + }); + + it("does not call setAllowed or getAddress when not connected", async () => { + stubHappyPath({ connected: false }); + + await expect(connectWallet("freighter")).rejects.toThrow(); + + expect(mockSetAllowed).not.toHaveBeenCalled(); + expect(mockGetAddress).not.toHaveBeenCalled(); + }); + }); + + // ── 2. getAddress() failure ───────────────────────────────────────────── + + describe("getAddress() failure", () => { + it("throws when getAddress returns an error string", async () => { + stubHappyPath({ addressError: "User denied access" }); + + await expect(connectWallet("freighter")).rejects.toThrow( + "User denied access", + ); + }); + + it("throws a generic message when getAddress returns no address and no error", async () => { + stubHappyPath({ address: undefined, addressError: undefined }); + + await expect(connectWallet("freighter")).rejects.toThrow( + "Freighter did not return a valid public key.", + ); + }); + + it("throws when getAddress returns empty string address", async () => { + stubHappyPath({ address: "", addressError: undefined }); + + await expect(connectWallet("freighter")).rejects.toThrow( + "Freighter did not return a valid public key.", + ); + }); + + it("calls setAllowed before getAddress", async () => { + stubHappyPath({ addressError: "some error" }); + + await expect(connectWallet("freighter")).rejects.toThrow(); + + expect(mockSetAllowed).toHaveBeenCalledOnce(); + expect(mockGetAddress).toHaveBeenCalledOnce(); + + // setAllowed should be called before getAddress + const setAllowedOrder = mockSetAllowed.mock.invocationCallOrder[0]; + const getAddressOrder = mockGetAddress.mock.invocationCallOrder[0]; + expect(setAllowedOrder).toBeLessThan(getAddressOrder); + }); + }); + + // ── 3. getNetworkDetails() catch fallback ─────────────────────────────── + + describe("getNetworkDetails() catch fallback", () => { + it("returns a valid session when getNetworkDetails throws", async () => { + stubHappyPath(); + mockGetNetworkDetails.mockRejectedValue(new Error("network timeout")); + + const session: WalletSession = await connectWallet("freighter"); + + expect(session.walletId).toBe("freighter"); + expect(session.publicKey).toBe( + "GABCDEF1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ", + ); + // Falls back to env-based network (default is TESTNET → "Testnet") + expect(session.network).toBe("Testnet"); + expect(session.mocked).toBe(false); + }); + + it("returns Mainnet when getNetworkDetails throws and env is MAINNET", async () => { + // Temporarily override the env-based network ID used by wallet.ts. + // The module-level STELLAR_NETWORK_ID is computed at import time from + // NEXT_PUBLIC_STELLAR_NETWORK, so we mock the module to simulate MAINNET. + vi.resetModules(); + vi.doMock("@stellar/freighter-api", () => ({ + isConnected: mockIsConnected, + setAllowed: mockSetAllowed, + getAddress: mockGetAddress, + getNetworkDetails: mockGetNetworkDetails, + })); + + // Re-import with MAINNET env to get the correct STELLAR_NETWORK_ID + vi.stubEnv("NEXT_PUBLIC_STELLAR_NETWORK", "MAINNET"); + const { connectWallet: connectWalletMainnet } = await import("./wallet"); + + stubHappyPath(); + mockGetNetworkDetails.mockRejectedValue(new Error("boom")); + + const session: WalletSession = await connectWalletMainnet("freighter"); + + // The fallback uses STELLAR_NETWORK_ID which contains "Public" for mainnet + expect(session.network).toBe("Mainnet"); + + vi.unstubAllEnvs(); + vi.resetModules(); + }); + + it("returns a valid session when getNetworkDetails resolves with an error field", async () => { + stubHappyPath({ networkDetailsError: "something went wrong" }); + + const session: WalletSession = await connectWallet("freighter"); + + // When details.error is truthy, the passphrase branch is skipped, + // so the fallback STELLAR_NETWORK_ID is used (default TESTNET → "Testnet") + expect(session.network).toBe("Testnet"); + }); + }); + + // ── 4. Happy path (smoke) ─────────────────────────────────────────────── + + describe("happy path (smoke)", () => { + it("returns a valid WalletSession on successful connection", async () => { + stubHappyPath(); + + const session: WalletSession = await connectWallet("freighter"); + + expect(session).toMatchObject({ + walletId: "freighter", + walletName: "Freighter", + publicKey: "GABCDEF1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ", + network: "Testnet", + mocked: false, + }); + expect(session.connectedAt).toBeDefined(); + }); + }); +}); From 059552383458d46701ce4aecd1e6e0a187e8074a Mon Sep 17 00:00:00 2001 From: oshowunm Date: Sat, 29 Aug 2026 22:45:12 +0100 Subject: [PATCH 3/3] perf(backend): add composite index on StreamEvent(streamId, eventType) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit listStreams and findStreams in stream.repository.ts filter by events: { some: { eventType: X } } for cancelled/completed status, which is an EXISTS-style subquery on StreamEvent. The existing single-column indexes on streamId and eventType cannot efficiently serve this pattern. Adds @@index([streamId, eventType]) to the StreamEvent model and a corresponding migration. This allows Postgres to use an index-only scan for the status-filtered listing queries without scanning all events per stream. Closes #1248 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- .../migration.sql | 7 +++++++ backend/prisma/schema.prisma | 1 + 2 files changed, 8 insertions(+) create mode 100644 backend/prisma/migrations/20260829220000_add_stream_event_streamid_eventtype_index/migration.sql diff --git a/backend/prisma/migrations/20260829220000_add_stream_event_streamid_eventtype_index/migration.sql b/backend/prisma/migrations/20260829220000_add_stream_event_streamid_eventtype_index/migration.sql new file mode 100644 index 00000000..27e35838 --- /dev/null +++ b/backend/prisma/migrations/20260829220000_add_stream_event_streamid_eventtype_index/migration.sql @@ -0,0 +1,7 @@ +-- Add composite index for EXISTS-style subqueries that filter StreamEvent +-- by both streamId and eventType (e.g., cancelled/completed status filters). +-- The existing single-column indexes on streamId and eventType cannot efficiently +-- serve a query that needs both columns simultaneously. + +-- CreateIndex +CREATE INDEX IF NOT EXISTS "StreamEvent_streamId_eventType_idx" ON "StreamEvent"("streamId", "eventType"); diff --git a/backend/prisma/schema.prisma b/backend/prisma/schema.prisma index 320c1306..3ae1d0e9 100644 --- a/backend/prisma/schema.prisma +++ b/backend/prisma/schema.prisma @@ -83,6 +83,7 @@ model StreamEvent { @@unique([transactionHash, eventType]) @@index([streamId]) @@index([eventType]) + @@index([streamId, eventType]) @@index([timestamp]) @@index([transactionHash]) @@index([createdAt])