feat(hub): add bounded WebSocket broadcasts

hub/docs/README.md

@@ -9,6 +9,7 @@ Hub is being rebuilt. The [Hub overview](../README.md) describes what the baseli
 - [Database](database.md): Prisma schema conventions, migrations, and the shared client.
 - [Logging](logging.md): application events, automatic request records, and payload safety.
 - [Background jobs](background-jobs.md): PostgreSQL queues, workers, schedules, setup, and test isolation.
+- [WebSocket updates](websockets.md): local broadcasts, subscriptions, access hooks, and connection limits.
 - [Testing and operations](testing-and-operations.md): baseline checks and their limits.
 - [Storybook](storybook.md): retained presentation, fixtures, and screenshot evidence.
 - [Prototype canvas](prototype-canvas.md): the separately retained extension.

hub/docs/architecture.md

@@ -22,6 +22,8 @@ The [Pino integration](logging.md) writes structured application and HTTP logs t
 
 The [background-job foundation](background-jobs.md) uses BullMQ with PostgreSQL and separate on-demand and scheduled worker processes. Setup migrates BullMQ's own schema independently of Prisma. Product processors and schedules are not registered yet.
 
+The [WebSocket foundation](websockets.md) attaches to the API's HTTP server and broadcasts to local channel subscribers without persistence. The application rejects upgrades until access checks are supplied. Product channels, client recovery, and communication between processes are not implemented.
+
 ## Frontend
 
 ```mermaid

hub/docs/development.md

@@ -38,6 +38,8 @@ npm run dev
 
 Open <http://127.0.0.1:5173>. Vite serves the frontend and forwards `/api` requests to Express at <http://127.0.0.1:3001>. `GET /api/health` returns JSON; other API routes have not been built. Both servers accept connections only from this machine. Stop each with Ctrl+C.
 
+Vite also forwards WebSocket upgrades under `/api`. The `/api/realtime` endpoint rejects connections until application access checks are connected; see [WebSocket updates](websockets.md). Stopping the API closes its WebSocket connections as well.
+
 The API reads an optional `PORT` environment variable, defaulting to 3001. If you change it, update the local Vite proxy target to match. The editor forwards ports 3001, 5173, and 6006 from the devcontainer. API startup reads `.env` when present; Compose's environment takes precedence.
 
 ## Background workers

hub/docs/websockets.md

@@ -0,0 +1,95 @@
+# WebSocket updates
+
+Hub can send live updates to the client app and Copilot plugin through the `ws` integration under `server/integrations/ws/`. The integration owns connections and channel subscriptions in one API process. It does not store messages, replay missed events, or deliver between processes.
+
+The application attaches `/api/realtime` to the same HTTP server as Express and passes its broadcast function to `createApp()`. Vite forwards WebSocket upgrades under `/api` during development. The endpoint rejects connections until application access checks are supplied. There are no comment channels or product events yet.
+
+## Server operations
+
+`attachRealtimeServer({ server, path, logger, limits?, authenticate?, authorizeSubscription? })` receives an existing Node HTTP server and an injected Pino logger. It does not start listening or open another port. It returns an instance with two operations:
+
+- `broadcast(channel, { type, payload })` sends a JSON event to that channel's connected subscribers in this process. It validates the event even when there are no subscribers. Invalid input throws; an oversized event throws `RangeError`. Individual connection failures are logged and do not stop delivery to other subscribers. Sending is not an acknowledgment that a client processed the event.
+- `close()` returns a promise, stops upgrades, cancels pending admission, clears timers and subscriptions, and closes sockets. It terminates sockets that do not close before the shutdown deadline. Repeated calls return the same promise. The caller still owns the HTTP server. Broadcasting after close throws.
+
+Application startup creates the shared server, attaches WebSockets, and installs Express as its HTTP request handler:
+
+```ts
+const server = createServer()
+const realtime = attachRealtimeServer({ server, path: "/api/realtime", logger })
+server.on("request", createApp({ broadcast: realtime.broadcast }))
+server.listen(port)
+```
+
+`createApp({ logger?, broadcast? })` and route registration accept an optional broadcast function. Features that receive it can call `broadcast?.(channel, event)` after a successful database commit. Without it, HTTP features work without live updates. Pass the broadcaster explicitly rather than using a global server. If a supplied broadcaster fails, report that failure rather than pretending it sent the update.
+
+Keep comment storage and channel selection in the feature's domain. A feature must not report a saved comment as unsaved merely because a later notification failed.
+
+## Client messages
+
+Browser clients use native `WebSocket`. Other clients use a WebSocket client supported by their runtime. Clients may subscribe and unsubscribe, but cannot publish application events.
+
+| Client message | Server reply |
+| --- | --- |
+| `{"type":"subscribe","channel":"example:123"}` | `{"type":"subscribed","channel":"example:123"}` |
+| `{"type":"unsubscribe","channel":"example:123"}` | `{"type":"unsubscribed","channel":"example:123"}` |
+
+Channel names are case-sensitive, contain 1 to 200 ASCII characters, and start with a letter or digit. Later characters may also contain `:`, `.`, `_`, `/`, or `-`. Repeating a subscription does not add another membership or consume another slot. Unsubscribing from a channel without membership is harmless.
+
+Broadcasts use this shape:
+
+```json
+{
+  "type": "event",
+  "channel": "example:123",
+  "event": { "type": "changed", "payload": { "id": "123" } }
+}
+```
+
+Event types contain 1 to 100 characters; payloads must be JSON values. Feature code owns the payload contract.
+
+Rejected client messages receive `{"type":"error","code":"INVALID_MESSAGE"}` or the same shape with code `FORBIDDEN` or `SUBSCRIPTION_LIMIT`. Invalid JSON, binary messages, unknown operations, extra fields, and invalid channel names are rejected. Logs contain static reasons and connection IDs, not payloads, channel names, credentials, or raw request URLs. The first rejected upgrade per status is logged immediately; repeats are summarized at each heartbeat and shutdown instead of producing an unrestricted stream of rejection logs.
+
+The server installs membership before confirming a subscription. Clients should subscribe, wait for confirmation, then fetch current state. Events received during that fetch must trigger another refresh rather than being overwritten by an older response. Clients must resubscribe and refetch after reconnecting; this integration does not implement client recovery.
+
+## Access checks
+
+The integration does not implement sign-in or choose sessions, tokens, or user records. Application startup may supply:
+
+- `authenticate(request, signal)`: returns the caller's identity, or `null` to reject admission.
+- `authorizeSubscription(identity, channel, signal)`: returns whether that caller may subscribe.
+
+These callbacks can be asynchronous. If either callback is missing, upgrades return HTTP 403. Authentication returning `null` produces 401, a failed check produces 500, and an admission timeout produces 408. A denied subscription returns `FORBIDDEN`; a failed subscription check closes the socket with code 1011.
+
+Callbacks must honor their abort signal and finish promptly. Admission timeout, disconnect, and shutdown cancel outstanding checks. Subscription checks also have a deadline. The integration bounds unresolved access checks even if a callback ignores cancellation; capacity remains occupied until those checks settle. Only one subscription check may be outstanding per socket; sending more messages while it is pending closes that socket with code 1008 instead of building a queue.
+
+When authentication is connected, its application policy must also check browser origins, credential expiry, and access revocation for existing subscriptions. Express middleware does not run automatically for WebSocket upgrades. Test-only callbacks that allow synthetic callers must not be used by application startup.
+
+## Resource limits
+
+The defaults in `server/integrations/ws/constants.ts` are initial safety limits, not measured production capacity. Supply positive integer overrides through `limits`. Tune connection limits against measured process memory with headroom before deployment.
+
+| Limit | Default |
+| --- | --- |
+| Pending and established connections per instance | 1,000 |
+| Connections per peer address | 50 |
+| Tracked peer addresses | 4,096 |
+| Upgrade attempts per peer address per minute | 60 |
+| Incoming message and serialized broadcast size | 16 KiB |
+| Subscriptions per connection | 32 |
+| Client messages, pings, and pongs per connection per second | 20 |
+| Queued outgoing bytes per connection | 256 KiB |
+| Heartbeat interval | 30 seconds |
+| Admission and subscription check deadline | 5 seconds |
+| Shutdown deadline | 5 seconds |
+
+Capacity is reserved before asynchronous admission and released on failure or disconnect. Full connection or address-tracking capacity returns HTTP 503; address connection and attempt limits return 429. Idle address counters expire after their attempt window. Compression is disabled. The transport also limits each message to 128 fragments and its receiver to 128 buffered chunks, so tiny fragments cannot bypass the payload byte limit through excessive bookkeeping.
+
+The integration uses the direct peer address and ignores forwarded headers. Behind a proxy, callers may therefore share one address. Operators must configure proxy-side caller limits and tune the peer cap for that topology; do not trust arbitrary forwarded headers. The proxy must support upgrades and limit handshakes and connections before they reach Node. Application limits do not protect against every form of network exhaustion.
+
+Oversized incoming messages close with code 1009. Excess message rates close with 1008. Dead or slow connections are terminated. Heartbeat checks, address tracking, subscriptions, and pending access checks all have bounded state.
+
+## Delivery limits
+
+Any route using the injected broadcaster reaches only this API process. A worker or another API process cannot reach these sockets without a shared signaling layer. Add that layer before porting a feature that publishes across processes.
+
+Notifications are best effort. A client can miss an event while disconnected, and a process can fail between saving a record and publishing. Feature clients should reconcile on reconnect and focus, with periodic refresh if a bounded delay is required. Durable records remain the source of truth.

hub/package.json

@@ -58,6 +58,7 @@
     "pg": "8.23.1",
     "react": "19.3.0",
     "react-dom": "19.3.0",
+    "ws": "8.22.0",
     "zod": "4.6.5"
   },
   "devDependencies": {
@@ -74,6 +75,7 @@
     "@types/pg": "8.23.1",
     "@types/react": "19.3.0",
     "@types/react-dom": "19.3.0",
+    "@types/ws": "8.18.2",
     "@typescript/native": "npm:typescript@7.0.2",
     "@vitejs/plugin-react": "6.1.2",
     "ajv": "8.20.0",

hub/server/create-app/index.test.ts

@@ -18,6 +18,28 @@ function parseRecord(line: string) {
 }
 
 describe("createApp()", () => {
+  it("accepts a broadcast dependency without changing HTTP routes", async () => {
+    const broadcast = vi.fn()
+    const server = createServer(createApp({ broadcast }))
+
+    try {
+      server.listen(0, "127.0.0.1")
+      await once(server, "listening")
+      const { port } = server.address() as AddressInfo
+      const health = await fetch(`http://127.0.0.1:${port}/api/health`)
+      const missing = await fetch(`http://127.0.0.1:${port}/api/realtime`)
+
+      expect(health.status).toBe(200)
+      expect(await health.json()).toEqual({ status: "ok" })
+      expect(missing.status).toBe(404)
+      expect(broadcast).not.toHaveBeenCalled()
+    } finally {
+      await new Promise<void>((resolve, reject) => {
+        server.close((error) => (error ? reject(error) : resolve()))
+      })
+    }
+  })
+
   it("automatically logs health and not-found responses with the injected logger", async () => {
     const records: Record<string, unknown>[] = []
     const logger = createLogger({

hub/server/create-app/index.ts

@@ -1,19 +1,20 @@
 import express from "express"
-import type { Logger } from "pino"
 
 import { handleError } from "../integrations/express/handle-error/index.ts"
 import { respondNotFound } from "../integrations/express/respond-not-found/index.ts"
 import { createRequestLogger } from "../integrations/pino/create-request-logger/index.ts"
 import { routes } from "../routes/index.ts"
 
+import type { CreateAppOptions } from "./types.ts"
+
 /** Give every endpoint the same request logging and safe error responses. */
-export function createApp({ logger }: { logger?: Logger } = {}) {
+export function createApp({ logger, broadcast }: CreateAppOptions = {}) {
   const app = express()
   app.disable("x-powered-by")
   app.use(createRequestLogger(logger))
   app.use(express.json())
 
-  routes(app)
+  routes(app, { broadcast })
   app.use(respondNotFound)
   app.use(handleError)
 

hub/server/create-app/types.ts

@@ -0,0 +1,8 @@
+import type { Logger } from "pino"
+
+import type { Broadcast } from "../integrations/ws/types.ts"
+
+export interface CreateAppOptions {
+  logger?: Logger
+  broadcast?: Broadcast
+}

hub/server/index.ts

@@ -1,13 +1,31 @@
 import "dotenv/config"
 
+import { createServer } from "node:http"
+
 import { createApp } from "./create-app/index.ts"
 import { log } from "./integrations/pino/log/index.ts"
+import { logger } from "./integrations/pino/index.ts"
+import { attachRealtimeServer } from "./integrations/ws/attach-realtime-server/index.ts"
+import { shutdownServer } from "./shutdown-server/index.ts"
 
 const port = Number(process.env.PORT ?? 3001)
 if (!Number.isInteger(port) || port < 1 || port > 65535) {
   throw new Error("PORT must be an integer between 1 and 65535.")
 }
 
-createApp().listen(port, "127.0.0.1", () => {
+const server = createServer()
+const realtime = attachRealtimeServer({ server, path: "/api/realtime", logger })
+server.on("request", createApp({ broadcast: realtime.broadcast }))
+server.listen(port, "127.0.0.1", () => {
   log("info", "Hub API listening", { host: "127.0.0.1", port })
 })
+
+let shuttingDown = false
+function shutdown() {
+  if (shuttingDown) return
+  shuttingDown = true
+  void shutdownServer(server, realtime)
+}
+
+process.once("SIGINT", shutdown)
+process.once("SIGTERM", shutdown)