Add GraphQL subscriptions for live transfer streams

[Just-Bamford]

Description

This PR implements GraphQL subscriptions for real-time streaming of TokenTransfer and HostFnLog events from the Wraith indexer, addressing the need for push updates without polling.

this pr Closes #99

Problem Solved

Clients previously had to poll the REST API repeatedly to get transfer updates. This approach was inefficient and didn't scale well. The new GraphQL subscriptions enable:

• Real-time push updates via WebSocket (event-driven for TokenTransfer, polling for HostFnLog)
• Per-client filtering by contract, sender, or recipient addresses
• Backpressure handling to protect the server from slow consumers
• Idiomatic GraphQL for better developer experience

What Was Implemented

Core Features

1. GraphQL Schema & Resolvers (src/api/graphql.ts)

   • Query resolvers for transfers, hostFnLogs, and status
   • Subscription resolvers with async generators
   • Full type definitions: TokenTransfer, HostFnLog, Status, BackpressureEvent
   • Union type for flexible subscription events

2. Subscription Logic (src/api/subscriptions.ts)

   • subscribeToTransfers() - Real-time event-driven subscriptions
   • subscribeToHostFnLogs() - Database polling subscriptions
   • Backpressure handling: bounded queue (1000 messages), drops oldest on overflow
   • Per-client filtering: contracts, senders, recipients

3. Database Queries (src/db.ts)

   • queryHostFnLogs() - Paginated host function log retrieval with cursor support

4. Server Integration (src/index.ts)

   • Apollo Server initialization
   • WebSocket endpoint at /graphql/ws
   • Graceful startup with GraphQL logging

5. API Updates (src/api.ts)

   • Fixed queryHostFnLogs integration
   • Updated /host-fn/:contractId endpoint

Acceptance Criteria Met

✅ Real-time subscription delivery

• TokenTransfer: ~100ms latency via event emitters
• HostFnLog: ~1 second latency via database polling

✅ Filtering works (contract/asset)

• Filter by contract addresses
• Filter by sender addresses
• Filter by recipient addresses
• Server-side filtering reduces bandwidth

✅ Backpressure prevents OOM

• Bounded queue of 1000 messages per subscription
• Oldest messages dropped on overflow
• BackpressureEvent notifies clients
• Server memory protected regardless of consumer speed

[drips-wave]

@Just-Bamford Great news! 🎉 Based on an automated assessment of this PR, the linked Wave issue(s) no longer count against your application limits.

You can now already apply to more issues while waiting for a review of this PR. Keep up the great work! 🚀

Learn more about application limits

[Miracle656]

The architecture here is solid — Apollo Server for queries + graphql-ws/ws for subscriptions is the right pattern, and you've implemented the W061 essentials (per-client filtering via SubscriptionFilters, server-side backpressure for slow consumers, and a queryHostFnLogs addition). Nice. A few things to address before merge:

1. Test coverage is the blocker. src/__tests__/graphql.test.ts is 14 lines and only asserts that createGraphQLServer() returns an ApolloServer instance — it doesn't exercise any of the actual feature: subscription streaming, the per-client filter logic, or the backpressure path (src/api/subscriptions.ts, 288 lines, has no tests at all). #99's value is the live stream working correctly under filtering/backpressure, so that's what needs covering — e.g. publish events and assert subscribers receive (and slow consumers get coalesced/dropped per the backpressure policy), and that filters scope the stream.

2. Remove IMPLEMENTATION_SUMMARY.md — that's a build/scratch summary, not something to commit to the repo root. For GRAPHQL_SUBSCRIPTIONS.md (403 lines), if it's user-facing docs please move it under docs/ and trim it to what a consumer needs; otherwise drop it too.

3. Drop the reformatting churn. src/api.ts (~126 lines) and the db.ts function signatures were reformatted, which inflates the diff and makes the real changes hard to review. Please keep db.ts/api.ts to the functional changes only (the queryHostFnLogs addition + the subscription wiring).

Once there are real subscription tests and the diff is cleaned up, I'll re-review and merge. Thanks! 🎯

[Miracle656]

Coordination note: #126 just merged and adds the canonical GraphQL server at src/graphql/server.ts (Apollo, mounted at /graphql, with persisted-query + cost/depth guards). To avoid two parallel GraphQL servers, please rebase this PR to add your subscription support on top of that server (extend its schema/transport with subscriptions) rather than standing up src/api/graphql.ts separately. That plus the real subscription tests from the earlier review should get this in. Thanks!

[Miracle656]

Thanks for moving the server toward the canonical location and merging main — but two blockers remain, and the latest main-merge actually broke the build:

1. package.json is now invalid JSON. The merge of main dumped #125's Jest config keys into the dependencies object (and dropped commas):

"dependencies": {
  "@apollo/server": "^4.11.0",
  "@graphql-tools/schema": "^10.0.0",
  "clearMocks": true          // ← Jest config, not a dependency; no comma
  "collectCoverage": true,    // ← these belong in the "jest" block
  "coverageThreshold": { ... }

This won't npm install or build. Please restore a valid package.json: keep clearMocks/collectCoverage/coverageThreshold inside the "jest" block (from #125 on main), not in dependencies.

2. Still on Apollo Server 4. This branch pins @apollo/server ^4.11.0 + @graphql-tools/schema, but main is on Apollo 5 (^5.5.1 + @as-integrations/express4) with persisted-query allowlisting and cost/depth limiting (from #126). This is the same downgrade that blocked #133. Please build the subscriptions on top of #126's existing Apollo 5 server rather than re-introducing a parallel Apollo 4 one — keep main's deps, add only graphql-ws/ws for the subscription transport.

The real subscription tests (subscriptions.test.ts) are a good addition. Once the package.json parses again and the server is built on Apollo 5, this should come together. 👍

[Miracle656]

Thanks for the rework — the two things I flagged before are addressed: you kept the repo on Apollo Server 5 (@apollo/server ^5.5.1) instead of downgrading, and you added real coverage (src/__tests__/subscriptions.test.ts, ~500 lines). That's exactly what I wanted to see.

Two remaining blockers, both from the branch having fallen behind main:

1. Lockfile out of sync — CI fails at npm ci (Missing: @emnapi/core@1.11.1 from lock file). package.json got new deps (@graphql-tools/schema, graphql-ws) but package-lock.json wasn't regenerated to match.

2. Conflicts with current main — #141 (OpenAPI-from-Zod) just merged and rewrote every handler in src/api.ts with runtime Zod validation. Your PR also rewrites those same handlers, so they now collide (8 conflict regions in api.ts, plus package.json).

To unblock:

git fetch origin
git rebase origin/main
# resolve src/api.ts by layering your host-fn / queryHostFnLogs changes
#   ON TOP of the new zod-validated handlers (keep both)
# package.json: keep all deps from both sides (union)
npm install            # regenerate package-lock.json in sync
git add package-lock.json
git rebase --continue
git push --force-with-lease

Once it's rebased and npm ci + tsc are green I'll re-review — the subscription design itself (bounded 1000-msg queue, per-client filtering, event-driven transfers + polled host-fn logs) looks good.

[Miracle656]

An honest update, because you got here first and deserve to hear it directly.

I have merged #168, which implements GraphQL subscriptions over the same two event types — TokenTransfer and HostFnLog. That means this PR's feature is now on main.

Why it went that way, plainly: this branch has been in a conflicting state since June and was asked to rebase at the time. #168 arrived current, at +619/-4 against your +1449/-174, with tests, and passed CI green. With the Wave running I did not think it was right to hold the feature for a branch that had been stalled for two months.

I have not closed this, and I am not going to. Two options, both fine:

• If you want to close it, that is the simplest ending.
• If there is anything in your implementation that feat: add GraphQL subscriptions for live transfer and host-fn log streams #168 does not cover — yours was a larger diff, so there may well be — open a focused PR for just that part and it will be reviewed on its own merits. Genuinely: rebasing 1,449 lines to re-land a feature that now exists is not a good use of your time, but a gap feat: add GraphQL subscriptions for live transfer and host-fn log streams #168 missed is worth having.

Sorry it landed this way. If you would rather I diff your branch against what merged and tell you what is actually missing, say so and I will do that.

[Miracle656]

Approved and merging — I ported the part of this that survives (a66fab0), since the wave has closed.

The honest summary: #133 landed GraphQL subscriptions while this was open, so src/api/subscriptions.ts became a second implementation of the same feature. That is a sequencing outcome, and I want to be clear about which parts of each were better before saying what I kept.

Where the merged one wins: it bounds memory using the socket's real ws.bufferedAmount rather than maintaining a synthetic 1000-message queue alongside it. A parallel queue can disagree with the kernel about how backed up a connection actually is — you can be well under 1000 queued messages and still have megabytes stuck in the socket buffer, or vice versa. Asking the socket is the measurement that cannot drift.

Where yours wins, and this is the part I kept: the merged implementation dropped silently. if (ws.bufferedAmount > MAX_BUFFERED_BYTES) return; — no counter, no notice, nothing. And a subscriber whose stream has lost events cannot distinguish a quiet chain from a hole in its own data. It will treat an incomplete history as complete, which is strictly worse than an error, because nothing downstream will ever question it. Your BackpressureEvent with a droppedCount was the right instinct and it is now what ships.

What I built from it:

• createBackpressureSender counts drops and emits one { type: "backpressure", payload: { droppedCount, message } }, pointing the client at the REST API to fill the gap.
• Debounced, not per-drop. A saturated socket drops in bursts; a notice per dropped message would add to the very congestion it is reporting.
• Only sent once bufferedAmount is back under the threshold. This is the detail that took a second pass to get right: sending the notice into a still-saturated socket means the notice gets dropped, and the client learns nothing. It re-arms and waits instead.
• Extracted over a minimal SendableSocket interface so it could be tested honestly. The interesting behaviour only happens above the buffer threshold, and that is not something a loopback connection can be made to do reliably — the existing "50 rapid transfers" test proves the server survives a burst, not that a drop is reported. Seven deterministic tests with a fake socket cover the count, the collapse of a 200-message burst into one notice, the re-arm while still saturated, the reset between gaps, and the closed socket.

Verified: tsc --noEmit clean, full suite 389 passed.

Sorry this went the way it did — the backpressure thinking was the best thing in either PR and it took a maintainer port to land it.

[Just-Bamford]

normal you no suppose merge this code
no wam sha