feat: add topic-based filtering to the intents WebSocket gateway

PR_DESCRIPTION_66.md

@@ -0,0 +1,57 @@
+# PR Description: Guard minDstAmount BigInt Parsing in fill()
+
+## Issue Reference
+
+- Closes #66
+
+## PR Links
+
+- PR #160: https://github.com/stellar-vortex-protocol/vortex-backend/pull/160
+- Branch: `fix/min-dst-amount-defensive-parsing`
+
+## Summary
+
+`IntentsController.fill()` calls `BigInt(intent.minDstAmount)` directly. While `minDstAmount` is validated at creation time via `CreateIntentDto`, any future direct-DB-write path (e.g., seed data or an admin tool) could introduce a malformed value that causes an unhandled `SyntaxError` from `BigInt()`, surfacing as an ungraceful 500 instead of a clean 400.
+
+## Changes
+
+### Modified File: `src/intents/intents.controller.ts`
+
+Wrapped `BigInt(intent.minDstAmount)` in a defensive try/catch that throws a `BadRequestException` with a clear data-integrity error message:
+
+```typescript
+let minAmount: bigint;
+try {
+  minAmount = BigInt(intent.minDstAmount);
+} catch {
+  throw new BadRequestException({
+    error: "Data integrity error: intent minDstAmount is not a valid integer",
+    intentId: id,
+    minDstAmount: intent.minDstAmount,
+  });
+}
+```
+
+This ensures that even if a malformed `minDstAmount` exists in the database, the fill endpoint returns a structured 400 response with a clear error message rather than crashing with an unhandled exception.
+
+### Modified File: `test/intents.e2e-spec.ts`
+
+Added an e2e test that:
+1. Creates an intent and accepts it via the API
+2. Uses `IntentsService.update()` directly to corrupt the intent's `minDstAmount` to `"not-a-number"`
+3. Attempts to fill the intent via the API
+4. Asserts HTTP 400 with `error: "Data integrity error: intent minDstAmount is not a valid integer"`
+
+## Verification
+
+- `npm run lint` passes
+- `npm run typecheck` passes
+- `npm test` passes
+- `npm run test:e2e` passes (new e2e test for malformed minDstAmount)
+
+## Acceptance Criteria
+
+- [x] Defensive try/catch around `BigInt(intent.minDstAmount)` in `fill()`
+- [x] Clean `BadRequestException` with data-integrity error message instead of unhandled 500
+- [x] E2E test added and passing
+- [x] PR description includes "Closes #66"

PR_DESCRIPTION_79.md

@@ -0,0 +1,49 @@
+# PR Description: Add Topic-Based Subscriptions to the WebSocket Gateway
+
+## Issue Reference
+
+- Closes #79
+
+## PR Links
+
+- PR #161: https://github.com/stellar-vortex-protocol/vortex-backend/pull/161
+- Branch: `feature/ws-topic-subscriptions`
+
+## Summary
+
+`IntentsGateway.broadcast()` sends every event to every connected subscriber unconditionally. A solver bot only interested in stellar intents still receives the full firehose across all seven `SupportedChain` values. This PR adds topic-based filtering so subscribers can receive only the events they care about.
+
+## Changes
+
+### Modified File: `src/intents/intents.gateway.ts`
+
+- Changed `subscribers` from `Set<WebSocket>` to `Map<WebSocket, SubscriberFilter>` to track per-subscriber chain filters
+- Added `handleMessage()` private method that processes incoming WebSocket messages
+- Clients can send `{ type: "subscribe", chains: ["stellar"] }` after connecting to filter events
+- `broadcast()` now checks each subscriber's filter before sending; events whose chain doesn't match the subscriber's filter are skipped
+- Added `getEventChain()` helper that resolves the chain from events:
+  - For `intent_created` events, reads `srcChain` directly from the intent object
+  - For `intent_filled`, `intent_accepted`, `intent_cancelled`, and `intent_expired` events, looks up the intent via `IntentsService.get()` to resolve `srcChain`
+  - Returns `null` for events without a chain, which are always delivered to all subscribers
+
+### Modified File: `scripts/solver-bot.ts`
+
+- Added `SOLVER_CHAINS` environment variable (defaults to all supported chains)
+- Bot subscribes to configured chains after connecting via `{ type: "subscribe", chains: [...] }`
+- Bot handles the `subscribed` confirmation event
+- Bot skips intents on non-subscribed chains in `tryFillOpenIntent()` with a log message
+
+## Verification
+
+- `npm run lint` passes
+- `npm run typecheck` passes
+- `npm test` passes
+- `npm run test:e2e` passes
+
+## Acceptance Criteria
+
+- [x] Subscribe/filter message implemented for WebSocket clients
+- [x] Per-subscriber chain filters tracked in the subscribers map
+- [x] `broadcast()` filters events by subscriber's chain filter
+- [x] `solver-bot.ts` updated to demonstrate topic-based filtering
+- [x] PR description includes "Closes #79"

scripts/solver-bot.ts

@@ -4,12 +4,14 @@ import WebSocket from "ws";
 const API_BASE = process.env.API_BASE ?? "http://localhost:4000";
 const WS_URL = process.env.WS_URL ?? "ws://localhost:4000/ws";
 const SOLVER_ADDRESS = process.env.SOLVER_ADDRESS ?? "SOLVER_ALPHA";
+const SOLVER_CHAINS = (process.env.SOLVER_CHAINS ?? "stellar,ethereum,base,polygon,arbitrum,optimism,avalanche").split(",");
 
 interface Intent {
   intentId: string;
   state: string;
   minDstAmount: string;
   deadline: number;
+  srcChain: string;
 }
 
 async function acceptIntent(intentId: string): Promise<boolean> {
@@ -46,6 +48,10 @@ async function fillIntent(intentId: string, minDstAmount: string): Promise<void>
 async function tryFillOpenIntent(intent: Intent): Promise<void> {
   const now = Math.floor(Date.now() / 1000);
   if (intent.state !== "open" || intent.deadline <= now) return;
+  if (!SOLVER_CHAINS.includes(intent.srcChain)) {
+    console.log(`[solver-bot] skipping ${intent.intentId} on ${intent.srcChain} (not subscribed)`);
+    return;
+  }
 
   const accepted = await acceptIntent(intent.intentId);
   if (!accepted) return;
@@ -55,6 +61,7 @@ async function tryFillOpenIntent(intent: Intent): Promise<void> {
 
 function main() {
   console.log(`[solver-bot] connecting as ${SOLVER_ADDRESS} to ${WS_URL}`);
+  console.log(`[solver-bot] subscribed chains: ${SOLVER_CHAINS.join(", ")}`);
   const ws = new WebSocket(WS_URL);
 
   ws.on("message", (raw) => {
@@ -63,6 +70,10 @@ function main() {
     switch (event.type) {
       case "connected":
         console.log(`[solver-bot] ${event.message}`);
+        ws.send(JSON.stringify({ type: "subscribe", chains: SOLVER_CHAINS }));
+        break;
+      case "subscribed":
+        console.log(`[solver-bot] subscribed with filter: ${JSON.stringify(event.filter)}`);
         break;
       case "snapshot":
         console.log(`[solver-bot] snapshot: ${event.intents.length} open intent(s)`);
@@ -71,7 +82,7 @@ function main() {
         }
         break;
       case "intent_created":
-        console.log(`[solver-bot] new intent ${event.intent.intentId}`);
+        console.log(`[solver-bot] new intent ${event.intent.intentId} on ${event.intent.srcChain}`);
         void tryFillOpenIntent(event.intent as Intent);
         break;
       case "intent_accepted":

src/intents/intents.gateway.ts

@@ -2,15 +2,20 @@ import { OnGatewayConnection, OnGatewayDisconnect, WebSocketGateway } from "@nes
 import { WebSocket } from "ws";
 import { IntentsService } from "./intents.service";
 
+interface SubscriberFilter {
+  chains?: string[];
+}
+
 @WebSocketGateway({ path: "/ws" })
 export class IntentsGateway implements OnGatewayConnection, OnGatewayDisconnect {
-  private readonly subscribers = new Set<WebSocket>();
+  private readonly subscribers = new Map<WebSocket, SubscriberFilter>();
 
   constructor(private readonly intentsService: IntentsService) {}
 
   handleConnection(client: WebSocket) {
-    this.subscribers.add(client);
+    this.subscribers.set(client, {});
     client.on("error", () => this.subscribers.delete(client));
+    client.on("message", (raw) => this.handleMessage(client, raw));
 
     client.send(JSON.stringify({ type: "connected", message: "Vortex intent stream" }));
 
@@ -22,12 +27,50 @@ export class IntentsGateway implements OnGatewayConnection, OnGatewayDisconnect
     this.subscribers.delete(client);
   }
 
+  private handleMessage(client: WebSocket, raw: Buffer) {
+    let message: { type?: string; chains?: string[] };
+    try {
+      message = JSON.parse(raw.toString());
+    } catch {
+      return;
+    }
+
+    if (message.type === "subscribe") {
+      const filter: SubscriberFilter = {};
+      if (message.chains && message.chains.length > 0) {
+        filter.chains = message.chains;
+      }
+      this.subscribers.set(client, filter);
+      client.send(JSON.stringify({ type: "subscribed", filter }));
+    }
+  }
+
   broadcast(event: { type: string; [key: string]: unknown }) {
     const payload = JSON.stringify(event);
-    for (const client of this.subscribers) {
-      if (client.readyState === WebSocket.OPEN) {
-        client.send(payload);
+    for (const [client, filter] of this.subscribers) {
+      if (client.readyState !== WebSocket.OPEN) continue;
+      if (filter.chains && filter.chains.length > 0) {
+        const eventChain = this.getEventChain(event);
+        if (eventChain && !filter.chains.includes(eventChain)) continue;
       }
+      client.send(payload);
     }
   }
-}
+
+  private getEventChain(event: { type: string; [key: string]: unknown }): string | null {
+    if (event.type === "intent_created" && event.intent) {
+      return (event.intent as { srcChain?: string }).srcChain ?? null;
+    }
+    if (
+      (event.type === "intent_filled" ||
+        event.type === "intent_accepted" ||
+        event.type === "intent_cancelled" ||
+        event.type === "intent_expired") &&
+      event.intentId
+    ) {
+      const intent = this.intentsService.get(event.intentId as string);
+      return intent?.srcChain ?? null;
+    }
+    return null;
+  }
+}