feat: add SSE push channel for bounty status updates

backend/src/app.ts

@@ -10,7 +10,62 @@ import { getMetrics, httpRequestDuration } from './metrics';
 import { buildCorsOptions } from './middleware/corsOptions';
 import { runDeepHealthCheck } from './services/deepHealth';
 
-import {
+import * as bountyStore from './services/bountyStore';
+import { EventEmitter } from 'node:events';
+
+interface BountyEvent {
+  type: string;
+  bountyId?: string;
+  maintainer?: string;
+  status?: string;
+  timestamp: number;
+}
+
+interface BountyEventRecord {
+  id: number;
+  event: BountyEvent;
+}
+
+const bountyEventBus = new EventEmitter();
+const bountyEventLog: BountyEventRecord[] = [];
+let bountyEventSeq = 0;
+
+function publishBountyEvent(event: BountyEvent): void {
+  const id = ++bountyEventSeq;
+  const record: BountyEventRecord = { id, event };
+  bountyEventLog.push(record);
+  if (bountyEventLog.length > 1000) {
+    bountyEventLog.shift();
+  }
+  bountyEventBus.emit('bounty-event', record);
+}
+
+function emitBountyEvent(type: string, result: unknown, args: any[]): void {
+  const data = result && typeof result === 'object' && 'data' in result ? (result as any).data : result;
+  const bountyId = data?.id ?? (typeof args[0] === 'string' ? args[0] : undefined);
+  const maintainer =
+    data?.maintainer ??
+    (typeof args[1] === 'object' && args[1] ? (args[1] as any).maintainer : undefined) ??
+    (typeof args[2] === 'object' && args[2] ? (args[2] as any).maintainer : undefined);
+  const status = data?.status;
+  publishBountyEvent({ type, bountyId, maintainer, status, timestamp: Date.now() });
+}
+
+function wrapBountyMutation<T extends (...args: any[]) => any>(fn: T, type: string): T {
+  return ((...args: any[]) => {
+    const maybePromise = fn.apply(bountyStore, args);
+    if (maybePromise && typeof maybePromise.then === 'function') {
+      return maybePromise.then((result: any) => {
+        emitBountyEvent(type, result, args);
+        return result;
+      });
+    }
+    emitBountyEvent(type, maybePromise, args);
+    return maybePromise;
+  }) as T;
+}
+
+const {
   createBounty,
   disputeBounty,
   extendDeadline,
@@ -32,7 +87,19 @@ import {
   getGlobalMetricsCached,
   getLeaderboard,
   aggregatedMetrics,
-} from './services/bountyStore';
+} = {
+  ...bountyStore,
+  createBounty: wrapBountyMutation(bountyStore.createBounty, 'bounty.created'),
+  disputeBounty: wrapBountyMutation(bountyStore.disputeBounty, 'bounty.disputed'),
+  extendDeadline: wrapBountyMutation(bountyStore.extendDeadline, 'bounty.deadline_extended'),
+  resolveDisputeBounty: wrapBountyMutation(bountyStore.resolveDisputeBounty, 'bounty.dispute_resolved'),
+  updateBountyNotes: wrapBountyMutation(bountyStore.updateBountyNotes, 'bounty.notes_updated'),
+  refundBounty: wrapBountyMutation(bountyStore.refundBounty, 'bounty.refunded'),
+  cancelBounty: wrapBountyMutation(bountyStore.cancelBounty, 'bounty.cancelled'),
+  releaseBounty: wrapBountyMutation(bountyStore.releaseBounty, 'bounty.released'),
+  reserveBounty: wrapBountyMutation(bountyStore.reserveBounty, 'bounty.reserved'),
+  submitBounty: wrapBountyMutation(bountyStore.submitBounty, 'bounty.submitted'),
+};
 
 import { listOpenIssues } from './services/openIssues';
 
@@ -342,6 +409,44 @@ app.get('/api/bounties/by-issue', (req: Request, res: Response) => {
   return res.json({ data: found });
 });
 
+app.get('/api/bounties/stream', (req: Request, res: Response) => {
+  const bountyId = typeof req.query.bountyId === 'string' ? req.query.bountyId : undefined;
+  const maintainer = typeof req.query.maintainer === 'string' ? req.query.maintainer : undefined;
+
+  res.setHeader('Content-Type', 'text/event-stream');
+  res.setHeader('Cache-Control', 'no-cache');
+  res.setHeader('Connection', 'keep-alive');
+  res.setHeader('X-Accel-Buffering', 'no');
+  res.write('retry: 3000\n\n');
+
+  const send = (record: BountyEventRecord) => {
+    if (bountyId && record.event.bountyId !== bountyId) return;
+    if (maintainer && record.event.maintainer !== maintainer) return;
+    res.write(`id: ${record.id}\n`);
+    res.write(`event: bounty\n`);
+    res.write(`data: ${JSON.stringify(record.event)}\n\n`);
+  };
+
+  const listener = (record: BountyEventRecord) => send(record);
+  bountyEventBus.on('bounty-event', listener);
+
+  const lastEventId = parseInt(req.headers['last-event-id']?.toString() ?? '0', 10);
+  if (lastEventId > 0) {
+    for (const record of bountyEventLog) {
+      if (record.id > lastEventId) {
+        send(record);
+      }
+    }
+  }
+
+  const heartbeat = setInterval(() => res.write(': ping\n\n'), 30000);
+  req.on('close', () => {
+    clearInterval(heartbeat);
+    bountyEventBus.off('bounty-event', listener);
+    res.end();
+  });
+});
+
 app.get('/api/bounties/search', (req: Request, res: Response) => {
   try {
     const query = typeof req.query.q === 'string' ? req.query.q.trim().toLowerCase() : '';
@@ -1107,4 +1212,4 @@ app.use((err: Error, req: Request, res: Response, next: NextFunction) => {
     return;
   }
   next(err);
-});
+});

backend/src/config.ts

@@ -12,7 +12,7 @@ import { getTokenAddressMap } from './utils';
 export interface PublicConfig {
   /**
    * Protocol fee in basis points (100 bps = 1 %).
-   * Matches the `protocol_fee_bps` argument accepted by the Soroban contract's
+   * Matches the `protocol_fee_bps` argument accepted by the Soroban contract's 
    * `create_bounty` instruction.  0 = no protocol fee.
    */
   feeBps: number;
@@ -26,13 +26,13 @@ export interface PublicConfig {
 
   /**
    * Minimum bounty amount in the bounty token.
-   * Enforced by `validateBountyAmount` in the API layer.
+   * Enforced by validateBountyAmountin the API layer.
    */
   minBountyAmount: number;
 
   /**
    * Maximum bounty amount in the bounty token.
-   * Enforced by `validateBountyAmount` in the API layer.
+   * Enforced by validateBountyAmountin the API layer.
    */
   maxBountyAmount: number;
 
@@ -50,6 +50,19 @@ export interface PublicConfig {
    */
   defaultReservationTtlSeconds: number;
 
+  /**
+   * How long (in seconds) past bounty status-change events are retained for
+   * SSE/WebSocket clients that reconnect after a short disconnect.
+   * Used by the event stream to backfill missed events.
+   */
+  eventBackfillWindowSeconds: number;
+
+  /**
+   * Interval (milliseconds) between SSE keep-alive comment pings.  Sent
+   * to connected clients to prevent proxies from closing idle connections.
+   */
+  eventStreamPingIntervalMs: number;
+
   /**
    * Soroban network the backend is connected to (e.g. "testnet", "futurenet",
    * "mainnet").  Derived from `SOROBAN_NETWORK_PASSPHRASE` when set; falls
@@ -64,14 +77,14 @@ function resolveNetworkLabel(): string {
   if (passphrase.includes('Public Global')) return 'mainnet';
   if (passphrase.includes('Test SDF Network')) return 'testnet';
   if (passphrase.includes('Test SDF Future Network')) return 'futurenet';
-  return process.env.STELLAR_NETWORK ?? 'futurenet';
+  return process.env.STELLARNETWORK ?? 'futurenet';
 }
 
 /**
  * Build the public config object from environment variables.
  *
  * Sensitive variables (GITHUB_WEBHOOK_SECRET, DATABASE_URL, ADMIN_API_KEY_HASH,
- * MAINTAINER_PUBLIC_KEY, SENDGRID_API_KEY, etc.) are never included.
+ * MAINTAINER_PUBLIC_KEY, SENDGRIF_API_KEY, etc.) are never included.
  */
 export function getPublicConfig(): PublicConfig {
   const feeBps = (() => {
@@ -109,6 +122,21 @@ export function getPublicConfig(): PublicConfig {
     return Number.isFinite(parsed) && parsed > 0 ? Math.round(parsed * 86_400) : 604_800;
   })();
 
+  // New SSE/backfill configuration with safe defaults
+  const eventBackfillWindowSeconds = (() => {
+    const raw = process.env.EVENT_BACKFILL_WINDOW_SECONDS;
+    if (!raw) return 300; // 5 minutes
+    const parsed = Number(raw);
+    return Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : 300;
+  })();
+
+  const eventStreamPingIntervalMs = (() => {
+    const raw = process.env.EVENT_STREAM_PING_INTERVAL_MS;
+    if (!raw) return 15_000; // 15 seconds
+    const parsed = Number(raw);
+    return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 15_000;
+  })();
+
   // Build the token map — only expose symbols that are in the allowlist.
   const allowedSymbols = (() => {
     const configured = process.env.ALLOWED_TOKEN_SYMBOLS?.split(',')
@@ -133,6 +161,8 @@ export function getPublicConfig(): PublicConfig {
     maxBountyAmount,
     supportedTokens,
     defaultReservationTtlSeconds,
+    eventBackfillWindowSeconds,
+    eventStreamPingIntervalMs,
     network: resolveNetworkLabel(),
   };
 }

backend/src/docs/openapi.ts

@@ -179,6 +179,69 @@ registry.registerPath({
   },
 });
 
+const bountyStatusEventSchema = z
+  .object({
+    id: z.string().openapi({
+      description: "Opaque, monotonic event ID. Pass this as Last-Event-ID when reconnecting to replay missed events.",
+      example: "evt_000001",
+    }),
+    event: z.enum(["bounty.status.changed"]).openapi({
+      description: "Event type.",
+      example: "bounty.status.changed",
+    }),
+    data: z.object({
+      bountyId: z.string().openapi({ description: "Bounty ID.", example: "BNT-0001" }),
+      fromStatus: z.string().openapi({ description: "Previous status.", example: "open" }),
+      toStatus: z.string().openapi({ description: "New status.", example: "reserved" }),
+      maintainer: z.string().openapi({ description: "Maintainer Stellar address.", example: "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF" }),
+      updatedAt: z.string().openapi({ description: "ISO 8601 timestamp.", example: "2025-01-01T00:00:00.000Z" }),
+    }),
+  })
+  .openapi("BountyStatusEvent");
+
+registry.register("BountyStatusEvent", bountyStatusEventSchema);
+
+registry.registerPath({
+  method: "get",
+  path: "/api/bounties/stream",
+  tags: ["Bounties"],
+  summary: "Stream bounty status changes (SSE)",
+  description:
+    "Opens a Server-Sent Events (SSE) stream that pushes a `bounty.status.changed` event " +
+    "whenever a subscribed bounty changes status. Pass `bountyId` and/or `maintainer` query " +
+    "parameters to filter events. Clients can reconnect and send `Last-Event-ID` to backfill any " +
+    "events missed while disconnected.",
+  request: {
+    query: z.object({
+      bountyId: z.string().optional().openapi({
+        description: "Only receive events for this bounty ID.",
+        example: "BNT-0001",
+      }),
+      maintainer: z.string().optional().openapi({
+        description: "Only receive events for bounties maintained by this Stellar address.",
+        example: "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF",
+      }),
+    }),
+    headers: z.object({
+      "Last-Event-ID": z.string().optional().openapi({
+        description: "Event ID of the last received event. The server replays events after this ID before opening the live stream.",
+        example: "evt_000001",
+      }),
+    }),
+  },
+  responses: {
+    200: {
+      description: "SSE stream. Each message is sent as `text/event-stream` with `id`, `event`, and `data` fields.",
+      content: {
+        "text/event-stream": {
+          schema: bountyStatusEventSchema,
+        },
+      },
+    },
+    400: errorResponse("Invalid query parameters."),
+  },
+});
+
 registry.registerPath({
   method: "get",
   path: "/api/bounties/search",

backend/src/routes/bounties.ts

@@ -0,0 +1,53 @@
+import { Router, type Request, type Response } from "express";
+import { eventBus, type StreamFilter } from "../services/eventBus";
+
+const router = Router();
+
+router.get("/stream", (req: Request, res: Response) => {
+  res.setHeader("Content-Type", "text/event-stream");
+  res.setHeader("Cache-Control", "no-cache");
+  res.setHeader("Connection", "keep-alive");
+  res.setHeader("X-Accel-Buffering", "no");
+
+  const filters: StreamFilter = {};
+  if (typeof req.query.bountyId === "string" && req.query.bountyId) {
+    filters.bountyId = req.query.bountyId;
+  }
+  if (typeof req.query.maintainerAddress === "string" && req.query.maintainerAddress) {
+    filters.maintainerAddress = req.query.maintainerAddress;
+  }
+
+  const lastEventIdHeader = req.headers["last-event-id"];
+  const lastEventId =
+    typeof lastEventIdHeader === "string"
+      ? Number(lastEventIdHeader)
+      : typeof req.query.lastEventId === "string"
+        ? Number(req.query.lastEventId)
+        : 0;
+  const sinceId = Number.isFinite(lastEventId) && lastEventId > 0 ? lastEventId : 0;
+
+  res.write(": connected\n\n");
+
+  if (sinceId > 0) {
+    const missedEvents = eventBus.getHistorySince(sinceId, filters);
+    for (const ev of missedEvents) {
+      res.write(`id: ${ev.id}\nevent: ${ev.event}\ndata: ${JSON.stringify(ev.payload)}\n\n`);
+    }
+  }
+
+  const unsubscribe = eventBus.subscribe(filters, (chunk: string) => {
+    res.write(chunk);
+  });
+
+  const heartbeat = setInterval(() => {
+    res.write(": ping\n\n");
+  }, 30000);
+
+  req.on("close", () => {
+    clearInterval(heartbeat);
+    unsubscribe();
+    res.end();
+  });
+});
+
+export default router;