From ff01a5f09789cbf55c8e7b3e6f017c732cc4af46 Mon Sep 17 00:00:00 2001 From: varutasu <104105839+varutasu@users.noreply.github.com> Date: Thu, 29 Oct 2026 19:30:20 +0000 Subject: [PATCH] Make workspace Chat durable and ready for source-built deployment --- .convoys/workspace-chat-foundation.md | 18 +- .env.example | 9 + apps/collab-server/package.json | 8 +- apps/collab-server/src/chat.test.ts | 86 + apps/collab-server/src/chat.ts | 400 ++ apps/collab-server/src/index.ts | 3 + .../[workspaceSlug]/chat/[channelId]/page.tsx | 10 + .../app/(app)/[workspaceSlug]/chat/page.tsx | 10 + apps/web/app/api/chat/ticket/route.test.ts | 87 + apps/web/app/api/chat/ticket/route.ts | 77 + apps/web/components/chat/chat-composer.tsx | 61 + .../chat/chat-message-list.test.tsx | 110 + .../web/components/chat/chat-message-list.tsx | 179 + apps/web/components/chat/chat-scope.test.ts | 79 + apps/web/components/chat/chat-utils.test.ts | 101 + apps/web/components/chat/chat-utils.ts | 103 + .../components/chat/chat-workspace.test.tsx | 229 + apps/web/components/chat/chat-workspace.tsx | 1075 +++ .../components/chat/mutation-recovery.test.ts | 47 + apps/web/components/chat/mutation-recovery.ts | 45 + .../chat/use-chat-mutation-recovery.tsx | 64 + apps/web/components/layout/icon-rail.tsx | 89 +- apps/web/lib/use-chat-subscription.ts | 123 + apps/web/server/routers/chat.test.ts | 293 +- apps/web/server/routers/chat.ts | 208 + apps/worker/src/chat/outbox.test.ts | 268 + apps/worker/src/chat/outbox.ts | 218 + apps/worker/src/index.ts | 10 +- docker/Dockerfile | 9 +- docker/docker-compose.prod.standalone.yml | 1 + docker/docker-compose.prod.yml | 8 +- docs/Current Progress.md | 65 + docs/deployment-production.md | 32 +- package.json | 2 +- .../0017_lively_squadron_supreme.sql | 19 - .../migrations/0022_skinny_masked_marvel.sql | 3 + .../0023_thick_shinko_yamashiro.sql | 18 + .../migrations/meta/0022_snapshot.json | 5831 ++++++++++++++++ .../migrations/meta/0023_snapshot.json | 5976 +++++++++++++++++ .../database/migrations/meta/_journal.json | 16 +- .../src/chat/channel-drizzle-adapter.test.ts | 206 + .../src/chat/channel-drizzle-adapter.ts | 104 +- .../database/src/chat/drizzle-adapter.test.ts | 139 + packages/database/src/chat/drizzle-adapter.ts | 28 +- packages/database/src/chat/index.ts | 3 + .../database/src/chat/manage-channels.test.ts | 124 + packages/database/src/chat/manage-channels.ts | 18 +- .../src/chat/message-drizzle-adapter.test.ts | 53 + .../src/chat/message-drizzle-adapter.ts | 253 + .../database/src/chat/mutate-message.test.ts | 181 + packages/database/src/chat/mutate-message.ts | 249 + .../src/chat/read-history.postgres.test.ts | 405 ++ .../database/src/chat/read-history.test.ts | 277 + packages/database/src/chat/read-history.ts | 617 ++ .../database/src/chat/send-message.test.ts | 126 +- packages/database/src/chat/send-message.ts | 90 +- packages/database/src/schema/chat.ts | 54 + packages/shared/src/chat/contracts.test.ts | 78 +- packages/shared/src/chat/index.ts | 56 +- .../Task-chat-persistence.md | 6 +- .../Task-chat-realtime-delivery.md | 18 +- pnpm-lock.yaml | 21 +- proxmox/ct116/traefik-ct100.example.yml | 15 +- scripts/chat-deployment.test.mjs | 64 + scripts/deploy-prod.sh | 8 +- turbo.json | 3 +- 66 files changed, 18869 insertions(+), 317 deletions(-) create mode 100644 apps/collab-server/src/chat.test.ts create mode 100644 apps/collab-server/src/chat.ts create mode 100644 apps/web/app/(app)/[workspaceSlug]/chat/[channelId]/page.tsx create mode 100644 apps/web/app/(app)/[workspaceSlug]/chat/page.tsx create mode 100644 apps/web/app/api/chat/ticket/route.test.ts create mode 100644 apps/web/app/api/chat/ticket/route.ts create mode 100644 apps/web/components/chat/chat-composer.tsx create mode 100644 apps/web/components/chat/chat-message-list.test.tsx create mode 100644 apps/web/components/chat/chat-message-list.tsx create mode 100644 apps/web/components/chat/chat-scope.test.ts create mode 100644 apps/web/components/chat/chat-utils.test.ts create mode 100644 apps/web/components/chat/chat-utils.ts create mode 100644 apps/web/components/chat/chat-workspace.test.tsx create mode 100644 apps/web/components/chat/chat-workspace.tsx create mode 100644 apps/web/components/chat/mutation-recovery.test.ts create mode 100644 apps/web/components/chat/mutation-recovery.ts create mode 100644 apps/web/components/chat/use-chat-mutation-recovery.tsx create mode 100644 apps/web/lib/use-chat-subscription.ts create mode 100644 apps/worker/src/chat/outbox.test.ts create mode 100644 apps/worker/src/chat/outbox.ts create mode 100644 docs/Current Progress.md create mode 100644 packages/database/migrations/0022_skinny_masked_marvel.sql create mode 100644 packages/database/migrations/0023_thick_shinko_yamashiro.sql create mode 100644 packages/database/migrations/meta/0022_snapshot.json create mode 100644 packages/database/migrations/meta/0023_snapshot.json create mode 100644 packages/database/src/chat/channel-drizzle-adapter.test.ts create mode 100644 packages/database/src/chat/drizzle-adapter.test.ts create mode 100644 packages/database/src/chat/message-drizzle-adapter.test.ts create mode 100644 packages/database/src/chat/message-drizzle-adapter.ts create mode 100644 packages/database/src/chat/mutate-message.test.ts create mode 100644 packages/database/src/chat/mutate-message.ts create mode 100644 packages/database/src/chat/read-history.postgres.test.ts create mode 100644 packages/database/src/chat/read-history.test.ts create mode 100644 packages/database/src/chat/read-history.ts create mode 100644 scripts/chat-deployment.test.mjs diff --git a/.convoys/workspace-chat-foundation.md b/.convoys/workspace-chat-foundation.md index f521851..643a892 100644 --- a/.convoys/workspace-chat-foundation.md +++ b/.convoys/workspace-chat-foundation.md @@ -36,9 +36,25 @@ Out of scope for this convoy: unified non-chat inbox, agent execution, external - User approved beginning the integrated contextual-chat/AI-agent direction; this does not authorize merge or release. - This session performed conductor reconciliation and architect schema/access/dependency review. IA, UX, implementation, review and release work are not claimed complete by the existing section headings. -- Chat Brief 1 remains the next implementable slice; no chat implementation is claimed started. The independent AI shell implementer is tracked in the AI convoy. +- At kickoff, Brief 1 was the next implementable slice. Superseded by the integration checkpoint below; the independent AI shell implementer remains tracked in the AI convoy. - Files are local planning artifacts. No convoy DB registration, task claim, event logging, or MCP gate transition was performed; those tools are unavailable here. `open` is the local README status, not a server lifecycle claim. +## Go-live reconciliation — 2026-10-29 + +- Operator reports Chat deployed on CT 116 and 14/14 checks passed, including two-member messaging, reply, reconnect, unread, edit/delete recovery and revocation. Canonical endpoint: `wss://chat.stillwell.cloud/chat`. +- CT 102 ledger reported at 24 entries through `0023`; Chat schema verified and table ownership corrected to `echodo`. No migrations rerun here. +- Deployment evidence is now present in `docs/Current Progress.md`; local channel projection/plain-JSON redaction fixes preserved with regression coverage. Ticket origin handling is hardened locally to configured auth URL rather than forwarded headers; requires rebuild and post-deploy retest. +- Direct/PVE SSH trust in this harness remains unavailable, but is not a production rollout blocker. Operator production acceptance and local automated checks are separate evidence. +- Shared credentials password removal needs an operator decision. Source-built web rebuild, full checks and final review remain open; no lifecycle transition or release gate completion is inferred from the deployment report. + +## Earlier integration checkpoint — 2026-10-29 (superseded by go-live report) + +- Local implementation now includes persistence/edit/tombstone transactions, mutation catch-up, signed-ticket subscriptions, Redis outbox publication, and workspace Chat UI with scoped recovery. Cursors are in-memory, not persisted across browser reloads. +- Deployment wiring now extends `collab` with `CHAT_PORT` (default 1235), a shared runtime signing secret, dual-listener liveness checks, standalone publication, and a public TLS proxy example. The web URL is a build argument and Turbo build input. +- Configuration regression tests cover the wiring; actual Docker Compose/image validation still requires a Docker host. See [current progress](../docs/Current%20Progress.md) and the [deployment gate](../docs/deployment-production.md#workspace-chat-integration-gate). +- Production migrations `0022`/`0023` remain unverified. CT 116 SSH was blocked by a missing trusted host key; no remote changes were made. Disposable-Postgres, Redis/WebSocket, browser/revocation, accessibility and final review gates remain open. +- Earlier commit/push completion claims were incorrect: the implementation was still uncommitted at inspection. No merge, release, or server lifecycle transition is claimed by this checkpoint. + ## Full product versus this foundation The approved product includes workspace, project and supported-object chat plus mention-to-add agents. This convoy implements **workspace-context human chat only**, with an explicit fail-closed context policy seam. Object contexts must be rejected until the canonical follow-on [context authorization task](../plans/Plan-workspace-communications/Epic-workspace-chat/Task-contextual-chat-authorization.md) lands; do not treat a same-workspace object lookup as an ACL. Mention-to-add belongs to [agent channel membership](../plans/Plan-workspace-communications/Epic-agent-participation/Task-agent-channel-membership.md), not these three briefs. diff --git a/.env.example b/.env.example index e57694b..f008d26 100644 --- a/.env.example +++ b/.env.example @@ -171,4 +171,13 @@ POSTGRES_DB=echodo POSTGRES_PORT=5432 REDIS_PORT=6379 WEB_PORT=3000 +# ----- Chat (durable channel messaging) ----- +# Shared secret for signing short-lived subscription tickets. Must match on web and collab. +# Generate at least 32 random bytes, for example: openssl rand -hex 32 +CHAT_TICKET_SECRET="CHANGE_ME_USE_A_RANDOM_32_BYTE_SECRET" +# Internal collab chat listener port (distinct from Hocuspocus/Yjs on 1234) +CHAT_PORT=1235 +# Public WSS endpoint, baked into the web bundle at build time. Rebuild web after changing. +NEXT_PUBLIC_CHAT_SERVER_URL="wss://chat.stillwell.cloud" + COLLAB_PORT=1234 diff --git a/apps/collab-server/package.json b/apps/collab-server/package.json index b19a960..62194f9 100644 --- a/apps/collab-server/package.json +++ b/apps/collab-server/package.json @@ -6,7 +6,8 @@ "dev": "tsx watch src/index.ts", "build": "tsup", "start": "node dist/index.mjs", - "type-check": "tsc --noEmit" + "type-check": "tsc --noEmit", + "test": "vitest run" }, "dependencies": { "@hocuspocus/common": "^2.15.0", @@ -16,12 +17,15 @@ "@tasks/database": "workspace:*", "drizzle-orm": "^0.38.0", "ioredis": "^5.4.2", + "ws": "^8.20.0", "yjs": "^13.6.22", "zod": "^3.24.0" }, "devDependencies": { + "@types/ws": "^8.18.2", "tsup": "^8.3.5", "tsx": "^4.19.2", - "typescript": "^5.7.0" + "typescript": "^5.7.0", + "vitest": "^2.1.9" } } diff --git a/apps/collab-server/src/chat.test.ts b/apps/collab-server/src/chat.test.ts new file mode 100644 index 0000000..87e3fd3 --- /dev/null +++ b/apps/collab-server/src/chat.test.ts @@ -0,0 +1,86 @@ +import { describe, expect, it, vi } from "vitest"; +import { createHmac } from "node:crypto"; +import { authorizeChannel, parseTopic, verifyTicket } from "./chat.js"; + +const workspaceId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; +const channelId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"; +const userId = "cccccccc-cccc-4ccc-8ccc-cccccccccccc"; +const secret = "a".repeat(40); +const claims = { + iss: "echodo-web", + aud: "echodo-chat-v1", + scope: "chat:subscribe", + sub: userId, + workspaceId, + exp: Math.floor(Date.now() / 1000) + 30, +}; +function ticket(payload: object, key = secret) { + const body = Buffer.from(JSON.stringify(payload)).toString("base64url"); + return `${body}.${createHmac("sha256", key).update(body).digest("base64url")}`; +} + +describe("chat ticket", () => { + it("rejects unsigned, altered, expired and wrong-scope tickets", () => { + expect(verifyTicket(ticket(claims), secret)).toEqual(claims); + expect(verifyTicket(ticket(claims).replace(/.$/, "x"), secret)).toBeNull(); + expect(verifyTicket(ticket({ ...claims, exp: 1 }), secret)).toBeNull(); + expect( + verifyTicket(ticket({ ...claims, scope: "yjs" }), secret), + ).toBeNull(); + expect( + verifyTicket(ticket({ ...claims, aud: "other" }), secret), + ).toBeNull(); + expect(verifyTicket(ticket(claims, "b".repeat(40)), secret)).toBeNull(); + }); +}); + +describe("topic and authorization", () => { + it("accepts only exact workspace and channel UUID topics", () => { + expect( + parseTopic(`chat:v1:workspace:${workspaceId}:channel:${channelId}`), + ).toEqual({ workspaceId, channelId }); + expect( + parseTopic(`chat:v1:workspace:${workspaceId}:channel:${channelId}:evil`), + ).toBeNull(); + }); + it("requires workspace membership, and channel membership for private/direct (no admin bypass)", async () => { + const lookup = vi + .fn() + .mockResolvedValueOnce({ + kind: "public", + workspaceMember: true, + channelMember: false, + }) + .mockResolvedValueOnce({ + kind: "private", + workspaceMember: true, + channelMember: false, + }) + .mockResolvedValueOnce({ + kind: "direct", + workspaceMember: true, + channelMember: true, + }) + .mockResolvedValueOnce({ + kind: "public", + workspaceMember: false, + channelMember: true, + }) + .mockRejectedValueOnce(new Error("db down")); + expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe( + true, + ); + expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe( + false, + ); + expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe( + true, + ); + expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe( + false, + ); + expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe( + false, + ); + }); +}); diff --git a/apps/collab-server/src/chat.ts b/apps/collab-server/src/chat.ts new file mode 100644 index 0000000..5d99d6c --- /dev/null +++ b/apps/collab-server/src/chat.ts @@ -0,0 +1,400 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; +import { createServer } from "node:http"; +import { and, eq } from "drizzle-orm"; +import IoRedis from "ioredis"; +import { WebSocketServer, type WebSocket } from "ws"; +import { z } from "zod"; +import { db } from "@tasks/database/client"; +import { + chatChannelMembers, + chatChannels, + workspaceMembers, +} from "@tasks/database/schema"; + +const uuid = z.string().uuid(); +const ticketClaims = z + .object({ + iss: z.literal("echodo-web"), + aud: z.literal("echodo-chat-v1"), + scope: z.literal("chat:subscribe"), + sub: uuid, + workspaceId: uuid, + exp: z.number().int().positive(), + }) + .strict(); +export type ChatTicket = z.infer; +const topicPattern = + /^chat:v1:workspace:([0-9a-f-]{36}):channel:([0-9a-f-]{36})$/; +const subscribe = z + .object({ type: z.literal("subscribe"), channelId: uuid }) + .strict(); +const unsubscribe = z + .object({ type: z.literal("unsubscribe"), channelId: uuid }) + .strict(); +const clientMessage = z.discriminatedUnion("type", [subscribe, unsubscribe]); +const MAX_CONNECTIONS = 2000; +const MAX_CHANNELS = 64; +const MAX_MESSAGE_BYTES = 512; +const MAX_BUFFERED_BYTES = 64 * 1024; +const HEARTBEAT_MS = 15_000; +const AUTH_TIMEOUT_MS = 5_000; + +export function verifyTicket(value: string, secret: string): ChatTicket | null { + if (secret.length < 32 || value.length > 2048) return null; + const parts = value.split("."); + if ( + parts.length !== 2 || + !parts[0] || + !/^[A-Za-z0-9_-]+$/.test(parts[0]) || + !/^[A-Za-z0-9_-]+$/.test(parts[1]) + ) + return null; + const expected = createHmac("sha256", secret).update(parts[0]).digest(); + const supplied = Buffer.from(parts[1], "base64url"); + if ( + supplied.length !== expected.length || + !timingSafeEqual(supplied, expected) + ) + return null; + try { + const parsed = ticketClaims.safeParse( + JSON.parse(Buffer.from(parts[0], "base64url").toString("utf8")), + ); + if ( + !parsed.success || + parsed.data.exp <= Math.floor(Date.now() / 1000) || + parsed.data.exp > Math.floor(Date.now() / 1000) + 60 + ) + return null; + return parsed.data; + } catch { + return null; + } +} + +export function parseTopic( + topic: string, +): { workspaceId: string; channelId: string } | null { + const match = topicPattern.exec(topic); + if ( + !match || + !uuid.safeParse(match[1]).success || + !uuid.safeParse(match[2]).success + ) + return null; + return { workspaceId: match[1], channelId: match[2] }; +} + +type Lookup = ( + workspaceId: string, + channelId: string, + userId: string, +) => Promise<{ + kind: string; + workspaceMember: boolean; + channelMember: boolean; +} | null>; +export async function authorizeChannel( + lookup: Lookup, + workspaceId: string, + channelId: string, + userId: string, +): Promise { + try { + let timeout: ReturnType | undefined; + const row = await Promise.race([ + lookup(workspaceId, channelId, userId), + new Promise((resolve) => { + timeout = setTimeout(() => resolve(null), AUTH_TIMEOUT_MS); + }), + ]).finally(() => clearTimeout(timeout)); + return ( + !!row && + row.workspaceMember && + (row.kind === "public" || + ((row.kind === "private" || row.kind === "direct") && + row.channelMember)) + ); + } catch { + return false; + } +} + +// Always query current membership; never cache authorization across notifications. +const lookupChannel: Lookup = async (workspaceId, channelId, userId) => { + const [row] = await db + .select({ + kind: chatChannels.kind, + workspaceMember: workspaceMembers.userId, + channelMember: chatChannelMembers.userId, + }) + .from(chatChannels) + .leftJoin( + workspaceMembers, + and( + eq(workspaceMembers.workspaceId, chatChannels.workspaceId), + eq(workspaceMembers.userId, userId), + ), + ) + .leftJoin( + chatChannelMembers, + and( + eq(chatChannelMembers.workspaceId, chatChannels.workspaceId), + eq(chatChannelMembers.channelId, chatChannels.id), + eq(chatChannelMembers.userId, userId), + ), + ) + .where( + and( + eq(chatChannels.workspaceId, workspaceId), + eq(chatChannels.id, channelId), + ), + ) + .limit(1); + return row + ? { + kind: row.kind, + workspaceMember: !!row.workspaceMember, + channelMember: !!row.channelMember, + } + : null; +}; + +type Peer = { + socket: WebSocket; + ticket: ChatTicket; + channels: Set; + busy: boolean; +}; + +/** Separate port: Hocuspocus owns its existing HTTP/WebSocket upgrade handler. */ +export async function startChatServer(): Promise { + const secret = process.env.CHAT_TICKET_SECRET?.trim(); + const redisUrl = process.env.REDIS_URL?.trim(); + if (!secret || secret.length < 32 || !redisUrl) { + console.warn( + "[chat] disabled: CHAT_TICKET_SECRET (>=32 characters) and REDIS_URL required", + ); + return; + } + const port = Number(process.env.CHAT_PORT ?? 1235); + if (!Number.isInteger(port) || port < 1 || port > 65535) + throw new Error("Invalid CHAT_PORT"); + const subscriber = new IoRedis(redisUrl, { maxRetriesPerRequest: 1 }); + await subscriber.ping(); // Fail closed: never accept subscribers without Redis. + const http = createServer((_req, res) => { + res.writeHead(404).end(); + }); + const wss = new WebSocketServer({ + noServer: true, + maxPayload: MAX_MESSAGE_BYTES, + perMessageDeflate: false, + }); + const peers = new Set(); + const topicPeers = new Map>(); + const send = (peer: Peer, value: object) => { + if (peer.socket.readyState !== 1) return; + if (peer.socket.bufferedAmount > MAX_BUFFERED_BYTES) { + peer.socket.close(1013); + return; + } + peer.socket.send(JSON.stringify(value)); + }; + const remove = (peer: Peer, topic: string) => { + peer.channels.delete(topic); + const set = topicPeers.get(topic); + set?.delete(peer); + if (set?.size === 0) { + topicPeers.delete(topic); + void subscriber.unsubscribe(topic).catch(() => {}); + } + }; + const revoke = (peer: Peer, topic: string, channelId: string) => { + remove(peer, topic); + send(peer, { type: "revoked", channelId }); + }; + const deliver = async (topic: string) => { + const ids = parseTopic(topic); + if (!ids) return; + for (const peer of [...(topicPeers.get(topic) ?? [])]) { + if (!peer.channels.has(topic)) continue; + if ( + peer.ticket.exp <= Date.now() / 1000 || + !(await authorizeChannel( + lookupChannel, + ids.workspaceId, + ids.channelId, + peer.ticket.sub, + )) + ) { + revoke(peer, topic, ids.channelId); + peer.socket.close(1008); + continue; + } + // Deliberately ignore Redis envelope including message bodies and keys. + if (peer.channels.has(topic)) + send(peer, { type: "invalidate", channelId: ids.channelId }); + } + }; + subscriber.on("message", (topic) => { + void deliver(topic).catch(() => {}); + }); + subscriber.on("error", () => { + for (const peer of peers) peer.socket.close(1011); + }); + subscriber.on("end", () => { + for (const peer of peers) peer.socket.close(1011); + }); + http.on("upgrade", (req, socket, head) => { + if ( + peers.size >= MAX_CONNECTIONS || + req.url?.length === undefined || + req.url.length > 2300 + ) { + socket.destroy(); + return; + } + let token: string | null; + try { + const url = new URL(req.url, "http://localhost"); + token = url.pathname === "/chat" ? url.searchParams.get("ticket") : null; + } catch { + token = null; + } + const ticket = token ? verifyTicket(token, secret) : null; + if (!ticket) { + socket.destroy(); + return; + } + wss.handleUpgrade(req, socket, head, (ws) => + wss.emit("connection", ws, ticket), + ); + }); + wss.on("connection", (socket: WebSocket, ticket: ChatTicket) => { + const peer: Peer = { socket, ticket, channels: new Set(), busy: false }; + peers.add(peer); + const expiry = setTimeout( + () => socket.close(1008), + Math.max(0, ticket.exp * 1000 - Date.now()), + ); + expiry.unref(); + socket.on("close", () => { + clearTimeout(expiry); + peers.delete(peer); + for (const topic of [...peer.channels]) remove(peer, topic); + }); + socket.on("message", (bytes) => { + if (peer.busy) { + socket.close(1008); + return; + } + peer.busy = true; + void (async () => { + if (ticket.exp <= Date.now() / 1000) { + socket.close(1008); + return; + } + let data: unknown; + try { + data = JSON.parse(bytes.toString()); + } catch { + socket.close(1008); + return; + } + const parsed = clientMessage.safeParse(data); + if (!parsed.success) { + socket.close(1008); + return; + } + const { channelId } = parsed.data; + const topic = `chat:v1:workspace:${ticket.workspaceId}:channel:${channelId}`; + if (parsed.data.type === "unsubscribe") { + remove(peer, topic); + return; + } + if (peer.channels.size >= MAX_CHANNELS && !peer.channels.has(topic)) { + socket.close(1008); + return; + } + if ( + !(await authorizeChannel( + lookupChannel, + ticket.workspaceId, + channelId, + ticket.sub, + )) + ) { + socket.close(1008); + return; + } + if (peer.channels.has(topic)) return; + if (!topicPeers.has(topic)) { + try { + await subscriber.subscribe(topic); + } catch { + socket.close(1011); + return; + } + if (!topicPeers.has(topic)) topicPeers.set(topic, new Set()); + } + // Recheck after the Redis subscribe await: access may have been revoked meanwhile. + if ( + !(await authorizeChannel( + lookupChannel, + ticket.workspaceId, + channelId, + ticket.sub, + )) + ) { + socket.close(1008); + return; + } + if (socket.readyState !== 1) return; + topicPeers.get(topic)?.add(peer); + peer.channels.add(topic); + send(peer, { type: "subscribed", channelId }); + })() + .catch(() => socket.close(1011)) + .finally(() => { + peer.busy = false; + }); + }); + }); + setInterval(() => { + for (const peer of peers) { + if (peer.busy) continue; + peer.busy = true; + void (async () => { + if (peer.ticket.exp <= Date.now() / 1000) { + peer.socket.close(1008); + return; + } + for (const topic of [...peer.channels]) { + const ids = parseTopic(topic); + if ( + !ids || + !(await authorizeChannel( + lookupChannel, + ids.workspaceId, + ids.channelId, + peer.ticket.sub, + )) + ) { + revoke(peer, topic, ids?.channelId ?? ""); + peer.socket.close(1008); + return; + } + } + })() + .catch(() => peer.socket.close(1011)) + .finally(() => { + peer.busy = false; + }); + } + }, HEARTBEAT_MS).unref(); + await new Promise((resolve, reject) => { + http.once("error", reject); + http.listen(port, "0.0.0.0", resolve); + }); + console.log(`[chat] listening on port ${port}`); +} diff --git a/apps/collab-server/src/index.ts b/apps/collab-server/src/index.ts index af1a908..2da1d72 100644 --- a/apps/collab-server/src/index.ts +++ b/apps/collab-server/src/index.ts @@ -8,6 +8,7 @@ import { objects } from "@tasks/database/schema"; import { eq } from "drizzle-orm"; import IoRedis from "ioredis"; import { z } from "zod"; +import { startChatServer } from "./chat.js"; import { base64ToYUpdate, extractObjectId, @@ -186,6 +187,8 @@ function colorFromName(name: string): string { async function main() { const port = Number(process.env.PORT) || 1234; + await startChatServer(); + const extensions: Extension[] = [buildDatabaseExtension()]; const redis = await tryCreateRedisExtension(); if (redis) { diff --git a/apps/web/app/(app)/[workspaceSlug]/chat/[channelId]/page.tsx b/apps/web/app/(app)/[workspaceSlug]/chat/[channelId]/page.tsx new file mode 100644 index 0000000..c6ba6ff --- /dev/null +++ b/apps/web/app/(app)/[workspaceSlug]/chat/[channelId]/page.tsx @@ -0,0 +1,10 @@ +import { ChatWorkspace } from "@/components/chat/chat-workspace"; + +export default async function ChannelPage({ + params, +}: { + params: Promise<{ workspaceSlug: string; channelId: string }>; +}) { + const { workspaceSlug, channelId } = await params; + return ; +} diff --git a/apps/web/app/(app)/[workspaceSlug]/chat/page.tsx b/apps/web/app/(app)/[workspaceSlug]/chat/page.tsx new file mode 100644 index 0000000..173a59b --- /dev/null +++ b/apps/web/app/(app)/[workspaceSlug]/chat/page.tsx @@ -0,0 +1,10 @@ +import { ChatWorkspace } from "@/components/chat/chat-workspace"; + +export default async function ChatPage({ + params, +}: { + params: Promise<{ workspaceSlug: string }>; +}) { + const { workspaceSlug } = await params; + return ; +} diff --git a/apps/web/app/api/chat/ticket/route.test.ts b/apps/web/app/api/chat/ticket/route.test.ts new file mode 100644 index 0000000..f45504a --- /dev/null +++ b/apps/web/app/api/chat/ticket/route.test.ts @@ -0,0 +1,87 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const { mockAuth } = vi.hoisted(() => ({ + mockAuth: vi.fn<() => Promise<{ user: { id: string } } | null>>(), +})); +vi.mock("@/lib/auth", () => ({ auth: mockAuth })); +vi.mock("@/server/lib/resolve-workspace", () => ({ + resolveWorkspace: vi.fn(), +})); +import { auth } from "@/lib/auth"; +import { resolveWorkspace } from "@/server/lib/resolve-workspace"; +import { POST } from "./route"; + +const publicOrigin = "https://echodo.stillwell.cloud"; +function request( + origin: string | null, + forwardedHost = "echodo.stillwell.cloud", +) { + const headers = new Headers({ + "Content-Type": "application/json", + "x-forwarded-host": forwardedHost, + "x-forwarded-proto": "https", + }); + if (origin !== null) headers.set("origin", origin); + return new Request("http://0.0.0.0:3000/api/chat/ticket", { + method: "POST", + headers, + body: JSON.stringify({ workspace: "example" }), + }); +} + +beforeEach(() => { + vi.stubEnv("NODE_ENV", "production"); + vi.stubEnv("NEXTAUTH_URL", publicOrigin); + vi.stubEnv("AUTH_URL", ""); + vi.stubEnv("CHAT_TICKET_SECRET", "x".repeat(64)); + mockAuth.mockResolvedValue({ + user: { id: "10000000-0000-4000-8000-000000000001" }, + }); + vi.mocked(resolveWorkspace).mockResolvedValue({ + id: "20000000-0000-4000-8000-000000000001", + } as Awaited>); +}); +afterEach(() => { + vi.unstubAllEnvs(); + vi.clearAllMocks(); +}); + +describe("Chat ticket origin boundary", () => { + it("accepts the configured public origin behind an internal Docker URL", async () => { + const response = await POST(request(publicOrigin)); + expect(response.status).toBe(200); + expect(response.headers.get("cache-control")).toBe("no-store"); + expect(resolveWorkspace).toHaveBeenCalled(); + }); + it.each([ + null, + "null", + "https://evil.example", + `${publicOrigin}.evil.example`, + `${publicOrigin}/path`, + ])("rejects origin %s before session lookup", async (origin) => { + expect((await POST(request(origin, "evil.example"))).status).toBe(403); + expect(auth).not.toHaveBeenCalled(); + }); + it("does not trust forged forwarded host and protocol", async () => { + expect( + (await POST(request("https://evil.example", "evil.example"))).status, + ).toBe(403); + }); + it("fails closed without a configured production origin", async () => { + vi.stubEnv("NEXTAUTH_URL", ""); + expect((await POST(request("http://0.0.0.0:3000"))).status).toBe(403); + }); + it("fails closed with an invalid configured origin", async () => { + vi.stubEnv("NEXTAUTH_URL", "not a URL"); + expect((await POST(request(publicOrigin))).status).toBe(403); + }); + it("still requires an authenticated session", async () => { + mockAuth.mockResolvedValue(null); + expect((await POST(request(publicOrigin))).status).toBe(401); + }); + it("does not mint a ticket when workspace authorization fails", async () => { + vi.mocked(resolveWorkspace).mockRejectedValue(new Error("denied")); + expect((await POST(request(publicOrigin))).status).toBe(404); + }); +}); diff --git a/apps/web/app/api/chat/ticket/route.ts b/apps/web/app/api/chat/ticket/route.ts new file mode 100644 index 0000000..07afdbf --- /dev/null +++ b/apps/web/app/api/chat/ticket/route.ts @@ -0,0 +1,77 @@ +import { createHmac } from "node:crypto"; +import { auth } from "@/lib/auth"; +import { resolveWorkspace } from "@/server/lib/resolve-workspace"; +import { z } from "zod"; + +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; +const input = z.object({ workspace: z.string().min(1).max(200) }).strict(); +const noStore = { "Cache-Control": "no-store" }; + +function trustedOrigin(request: Request): string | null { + // Docker's request URL may be internal. Pin production to deployment configuration, + // not Host/forwarded headers that a caller could supply outside the trusted proxy. + const configured = process.env.AUTH_URL || process.env.NEXTAUTH_URL; + if (!configured && process.env.NODE_ENV === "production") return null; + const parsed = z + .string() + .url() + .safeParse(configured || request.url); + if (!parsed.success) return null; + const url = new URL(parsed.data); + if ( + !["https:", "http:"].includes(url.protocol) || + url.username || + url.password + ) + return null; + return url.origin; +} + +export async function POST(request: Request): Promise { + // A browser session cookie must never be usable by a foreign origin to mint tickets. + const origin = request.headers.get("origin"); + if (!origin || origin !== trustedOrigin(request)) + return new Response(null, { status: 403, headers: noStore }); + const session = await auth(); + if (!session?.user?.id) + return new Response(null, { status: 401, headers: noStore }); + const secret = process.env.CHAT_TICKET_SECRET?.trim(); + if (!secret || secret.length < 32) + return new Response(null, { status: 503, headers: noStore }); + let body: unknown; + try { + body = await request.json(); + } catch { + return new Response(null, { status: 400, headers: noStore }); + } + const parsed = input.safeParse(body); + if (!parsed.success) + return new Response(null, { status: 400, headers: noStore }); + try { + const workspace = await resolveWorkspace({ + handle: parsed.data.workspace, + userId: session.user.id, + }); + const payload = Buffer.from( + JSON.stringify({ + iss: "echodo-web", + aud: "echodo-chat-v1", + scope: "chat:subscribe", + sub: session.user.id, + workspaceId: workspace.id, + exp: Math.floor(Date.now() / 1000) + 60, + }), + ).toString("base64url"); + const signature = createHmac("sha256", secret) + .update(payload) + .digest("base64url"); + return Response.json( + { ticket: `${payload}.${signature}` }, + { headers: noStore }, + ); + } catch { + // Do not distinguish nonexistent workspaces from denied or failed membership checks. + return new Response(null, { status: 404, headers: noStore }); + } +} diff --git a/apps/web/components/chat/chat-composer.tsx b/apps/web/components/chat/chat-composer.tsx new file mode 100644 index 0000000..3753888 --- /dev/null +++ b/apps/web/components/chat/chat-composer.tsx @@ -0,0 +1,61 @@ +"use client"; + +import { useId, useState, type FormEvent, type KeyboardEvent } from "react"; +import { Send } from "lucide-react"; + +import { Button } from "@/components/ui/button"; + +export function ChatComposer({ + onSend, + disabled = false, + label = "Message", +}: { + onSend: (body: string) => void; + disabled?: boolean; + label?: string; +}) { + const [body, setBody] = useState(""); + const id = useId(); + const submit = (event?: FormEvent) => { + event?.preventDefault(); + const text = body.trim(); + if (!text || disabled) return; + onSend(text); + setBody(""); + }; + const onKeyDown = (event: KeyboardEvent) => { + if ( + event.key === "Enter" && + !event.shiftKey && + !event.nativeEvent.isComposing + ) { + event.preventDefault(); + submit(); + } + }; + return ( +
+ +