| 1 | /** |
| 2 | * One channel's live room, as a Durable Object named by the channel id. |
| 3 | * |
| 4 | * Why an object per channel: everyone looking at a channel must see the |
| 5 | * same messages in the same order the moment they are written, and one |
| 6 | * object per channel sees every socket open on it. It holds the sockets |
| 7 | * with the WebSocket Hibernation API, so a quiet channel with people |
| 8 | * looking at it costs nothing between messages: the object is evicted |
| 9 | * from memory and the sockets stay open at the edge. Each socket carries |
| 10 | * who is behind it (`serializeAttachment`), which survives hibernation. |
| 11 | * |
| 12 | * The room authorizes nothing. The Worker checks a viewer may read the |
| 13 | * channel before forwarding their socket here (src/index.ts, `live`), and |
| 14 | * tells the room to drop someone's sockets when they leave a private one. |
| 15 | */ |
| 16 | import { DurableObject } from "cloudflare:workers"; |
| 17 | |
| 18 | import { |
| 19 | G1T_KEY, |
| 20 | integrationsClient, |
| 21 | principalKey, |
| 22 | type AskMessage, |
| 23 | type AskPage, |
| 24 | type ChatAnswer, |
| 25 | type ChatLiveEvent, |
| 26 | type ChatMessage, |
| 27 | type MemberProfile, |
| 28 | type ServiceBinding, |
| 29 | type User, |
| 30 | } from "@g1t/contracts"; |
| 31 | |
| 32 | import { runAnswer } from "./answer.ts"; |
| 33 | import { G1T_PROFILE, answerAllowed, metaColumn } from "./g1t.ts"; |
| 34 | import { PERSON_TYPING_MS, isTypingFrame } from "./messages.ts"; |
| 35 | import { notifyMessage } from "./notify.ts"; |
| 36 | |
| 37 | /** Header the Worker sets on a socket it forwards: who it is, as JSON (`RoomMember`). */ |
| 38 | export const ROOM_MEMBER_HEADER = "x-g1t-chat-member"; |
| 39 | |
| 40 | export type RoomMember = { channel_id: string; member: MemberProfile }; |
| 41 | |
| 42 | /** The shortest time between two "is typing" events from one socket. */ |
| 43 | const TYPING_EVERY_MS = 2_000; |
| 44 | /** How long g1t shows as typing before its first words, at most. */ |
| 45 | const G1T_TYPING_MS = 30_000; |
| 46 | |
| 47 | type RoomEnv = { |
| 48 | DB: D1Database; |
| 49 | /** Ask (services/integrations): where @g1t's answers come from. */ |
| 50 | INTEGRATIONS?: ServiceBinding; |
| 51 | NOTIFY?: ServiceBinding; |
| 52 | }; |
| 53 | |
| 54 | /** |
| 55 | * A question for @g1t to answer, as src/index.ts hands it over: who asked |
| 56 | * (the person, as their session had them), the conversation, the page, and |
| 57 | * the message the answer is written into. |
| 58 | */ |
| 59 | export type AnswerJob = { |
| 60 | slug: string; |
| 61 | channel: { id: string; kind: "channel" | "dm"; name: string | null }; |
| 62 | actor: User; |
| 63 | messages: AskMessage[]; |
| 64 | page: AskPage; |
| 65 | message: ChatMessage; |
| 66 | /** |
| 67 | * The person's newest message the answer read. Gone when it is looked |
| 68 | * for: the conversation was cleared after it was read, and the answer |
| 69 | * is thrown away (`clear`). Absent on a job handed over before it was. |
| 70 | */ |
| 71 | question_id?: string; |
| 72 | }; |
| 73 | |
| 74 | export class ChannelRoom extends DurableObject<RoomEnv> { |
| 75 | /** When each socket last said it was typing; lost on hibernation, which only lets one more through. */ |
| 76 | private typedAt = new WeakMap<WebSocket, number>(); |
| 77 | /** The answer being written now, to stop; one at a time per room. */ |
| 78 | private running: { id: string; stop: AbortController; superseded: boolean; cleared: boolean } | null = null; |
| 79 | |
| 80 | constructor(ctx: DurableObjectState, env: RoomEnv) { |
| 81 | super(ctx, env); |
| 82 | // Keepalives are answered without waking the object. |
| 83 | ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong")); |
| 84 | } |
| 85 | |
| 86 | override async fetch(request: Request): Promise<Response> { |
| 87 | if (await this.ctx.storage.get("erased")) return new Response("Conversation erased", { status: 410 }); |
| 88 | if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") { |
| 89 | return new Response("Expected a WebSocket upgrade\n", { status: 426 }); |
| 90 | } |
| 91 | let who: RoomMember; |
| 92 | try { |
| 93 | who = JSON.parse(request.headers.get(ROOM_MEMBER_HEADER) ?? "") as RoomMember; |
| 94 | } catch { |
| 95 | return new Response("Missing member\n", { status: 400 }); |
| 96 | } |
| 97 | if (await this.ctx.storage.get(`erased:${principalKey(who.member)}`)) return new Response("Account erased", { status: 410 }); |
| 98 | const pair = new WebSocketPair(); |
| 99 | const [client, server] = Object.values(pair) as [WebSocket, WebSocket]; |
| 100 | // Tagged with the member's key, so their sockets can be found to drop. |
| 101 | this.ctx.acceptWebSocket(server, [principalKey(who.member)]); |
| 102 | server.serializeAttachment(who); |
| 103 | return new Response(null, { status: 101, webSocket: client }); |
| 104 | } |
| 105 | |
| 106 | /** Sends `event` to every socket in the room, except those of `except` (a principal key). */ |
| 107 | broadcast(event: ChatLiveEvent, except: string | null = null): void { |
| 108 | const text = JSON.stringify(event); |
| 109 | for (const socket of this.ctx.getWebSockets()) { |
| 110 | if (except && this.ctx.getTags(socket).includes(except)) continue; |
| 111 | try { |
| 112 | socket.send(text); |
| 113 | } catch { |
| 114 | // Closing already; webSocketClose tidies up. |
| 115 | } |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | /** Closes a member's sockets: they left a private channel and may no longer read it. */ |
| 120 | drop(principal: string): void { |
| 121 | for (const socket of this.ctx.getWebSockets(principal)) { |
| 122 | try { |
| 123 | socket.close(4403, "No longer a member"); |
| 124 | } catch { |
| 125 | // Already closed. |
| 126 | } |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | // ── @g1t's answers (src/answer.ts) ────────────────────────────────────── |
| 131 | |
| 132 | /** |
| 133 | * Answers a question in this direct message with g1t. One answer at a |
| 134 | * time: a newer question takes over, stopping the one being written |
| 135 | * (what came of it stays; one that had not started is taken away), and |
| 136 | * the new answer reads both questions. Run from an alarm, so a long |
| 137 | * answer is not cut short when the request that asked ends. |
| 138 | */ |
| 139 | async startAnswer(job: AnswerJob): Promise<void> { |
| 140 | // Read before a clear that has deleted its question since: thrown away, never asked. |
| 141 | if (await this.questionGone(job)) { |
| 142 | await this.discard(job.channel.id, job.message.id); |
| 143 | return; |
| 144 | } |
| 145 | const times = answerAllowed((await this.ctx.storage.get<number[]>("answered")) ?? [], Date.now()); |
| 146 | await this.ctx.storage.put("answered", times.times); |
| 147 | if (!times.ok) { |
| 148 | const answer: ChatAnswer = { state: "error", error: "Too many questions in a minute." }; |
| 149 | await this.end(job, "You've asked a lot in the last minute. Wait a moment, then ask again.", answer, false); |
| 150 | return; |
| 151 | } |
| 152 | const pending = await this.ctx.storage.get<AnswerJob>("job"); |
| 153 | if (pending) await this.dropAnswer(pending); |
| 154 | if (this.running) { |
| 155 | this.running.superseded = true; |
| 156 | this.running.stop.abort(); |
| 157 | } |
| 158 | await this.ctx.storage.put("job", job); |
| 159 | await this.ctx.storage.setAlarm(Date.now()); |
| 160 | } |
| 161 | |
| 162 | /** Stops the answer being written, or about to be; what came stays. */ |
| 163 | async stopAnswer(): Promise<void> { |
| 164 | const pending = await this.ctx.storage.get<AnswerJob>("job"); |
| 165 | if (pending) { |
| 166 | await this.ctx.storage.delete("job"); |
| 167 | await this.end(pending, "_Stopped before answering._", { ...pending.message.answer, state: "stopped" }, false); |
| 168 | } |
| 169 | this.running?.stop.abort(); |
| 170 | } |
| 171 | |
| 172 | override async alarm(): Promise<void> { |
| 173 | const job = await this.ctx.storage.get<AnswerJob>("job"); |
| 174 | if (!job) return; |
| 175 | await this.ctx.storage.delete("job"); |
| 176 | const run = { id: job.message.id, stop: new AbortController(), superseded: false, cleared: false }; |
| 177 | // Running before it looks again, so a clear from now on finds it and stops it. |
| 178 | this.running = run; |
| 179 | if (await this.questionGone(job)) { |
| 180 | if (this.running === run) this.running = null; |
| 181 | await this.discard(job.channel.id, job.message.id); |
| 182 | return; |
| 183 | } |
| 184 | const ask = integrationsClient(this.env.INTEGRATIONS as ServiceBinding); |
| 185 | try { |
| 186 | await runAnswer( |
| 187 | { |
| 188 | slug: job.slug, |
| 189 | setup: async () => { |
| 190 | if (!this.env.INTEGRATIONS) return null; |
| 191 | const found = await ask.askSettings(job.slug, job.actor); |
| 192 | return found.ok ? found.value : { connection_id: "unknown", connection_name: null, provider: null, model: null, problem: found.error.message }; |
| 193 | }, |
| 194 | ask: (signal) => ask.ask(job.actor, job.slug, job.messages, job.page, signal), |
| 195 | typing: () => |
| 196 | this.broadcast({ type: "typing", channel_id: job.channel.id, member: G1T_PROFILE, until: new Date(Date.now() + G1T_TYPING_MS).toISOString() }), |
| 197 | show: (body, answer) => { |
| 198 | if (!run.cleared) this.broadcast({ type: "message.streaming", channel_id: job.channel.id, id: job.message.id, body, answer }); |
| 199 | }, |
| 200 | save: (body, answer) => (run.cleared ? Promise.resolve() : this.store(job.message.id, body, answer)), |
| 201 | finish: (body, answer) => (run.cleared ? this.discard(job.channel.id, job.message.id) : this.end(job, body, answer, run.superseded)), |
| 202 | now: () => Date.now(), |
| 203 | }, |
| 204 | run.stop.signal, |
| 205 | ); |
| 206 | } catch (error) { |
| 207 | console.error("g1t could not answer in", job.channel.id, error); |
| 208 | } finally { |
| 209 | if (this.running === run) this.running = null; |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | /** |
| 214 | * The conversation was just cleared (src/index.ts `g1tClear`, after its |
| 215 | * rows were deleted): an answer waiting to start is dropped, one being |
| 216 | * written is stopped and thrown away, not saved, so nothing it read |
| 217 | * comes back; and every tab looking empties. What it had already sent |
| 218 | * the workspace's AI provider has gone to that provider. |
| 219 | */ |
| 220 | async clear(channelId: string): Promise<void> { |
| 221 | const pending = await this.ctx.storage.get<AnswerJob>("job"); |
| 222 | await this.ctx.storage.delete("job"); |
| 223 | const ids = new Set<string>(); |
| 224 | if (pending) ids.add(pending.message.id); |
| 225 | if (this.running) { |
| 226 | this.running.cleared = true; |
| 227 | this.running.stop.abort(); |
| 228 | ids.add(this.running.id); |
| 229 | } |
| 230 | // Written after the rows went, an answer's own message would outlive the clear. |
| 231 | for (const id of ids) await this.env.DB.prepare("DELETE FROM messages WHERE id = ?").bind(id).run(); |
| 232 | this.broadcast({ type: "channel.cleared", channel_id: channelId }); |
| 233 | } |
| 234 | |
| 235 | /** |
| 236 | * An erasure (src/erasure.ts). With a member, only that person is shut |
| 237 | * out: everyone else's view and answers carry on, and nothing is cleared |
| 238 | * for them. Without one, the whole conversation is going: its answer |
| 239 | * job stops and every socket closes. |
| 240 | */ |
| 241 | async erase(channelId: string, principal: string | null): Promise<void> { |
| 242 | if (principal) { |
| 243 | await this.ctx.storage.put(`erased:${principal}`, true); |
| 244 | this.drop(principal); |
| 245 | return; |
| 246 | } |
| 247 | await this.ctx.storage.put("erased", true); |
| 248 | await this.clear(channelId); |
| 249 | await this.ctx.storage.deleteAlarm(); |
| 250 | for (const socket of this.ctx.getWebSockets()) { try { socket.close(4403, "Conversation erased"); } catch {} } |
| 251 | } |
| 252 | |
| 253 | /** Whether the question a job answers has been deleted by a clear. */ |
| 254 | private async questionGone(job: AnswerJob): Promise<boolean> { |
| 255 | if (await this.ctx.storage.get("erased") || await this.ctx.storage.get(`erased:user:${job.actor.id}`)) return true; |
| 256 | if (!job.question_id) return false; |
| 257 | const row = await this.env.DB.prepare("SELECT id FROM messages WHERE id = ?").bind(job.question_id).first<{ id: string }>(); |
| 258 | return !row; |
| 259 | } |
| 260 | |
| 261 | /** An answer to a cleared conversation: its message goes for good, and tabs drop it. */ |
| 262 | private async discard(channelId: string, id: string): Promise<void> { |
| 263 | await this.env.DB.prepare("DELETE FROM messages WHERE id = ?").bind(id).run(); |
| 264 | this.broadcast({ type: "message.deleted", channel_id: channelId, id }); |
| 265 | } |
| 266 | |
| 267 | /** Saves an answer as it is now. */ |
| 268 | private async store(id: string, body: string, answer: ChatAnswer): Promise<void> { |
| 269 | await this.env.DB.prepare("UPDATE messages SET body = ?, meta = ? WHERE id = ? AND deleted_at IS NULL") |
| 270 | .bind(body, metaColumn({ answer }), id) |
| 271 | .run(); |
| 272 | } |
| 273 | |
| 274 | /** A question taken over before its answer started: its empty message goes. */ |
| 275 | private async dropAnswer(job: AnswerJob): Promise<void> { |
| 276 | await this.ctx.storage.delete("job"); |
| 277 | await this.env.DB.prepare("UPDATE messages SET deleted_at = ?, body = '' WHERE id = ?").bind(new Date().toISOString(), job.message.id).run(); |
| 278 | this.broadcast({ type: "message.deleted", channel_id: job.channel.id, id: job.message.id }); |
| 279 | } |
| 280 | |
| 281 | /** |
| 282 | * Ends an answer: stored as a normal message, sent to the room, and the |
| 283 | * person notified (their direct message notifies them, as any would). |
| 284 | * One taken over by a newer question before a word of it came is taken |
| 285 | * away instead. |
| 286 | */ |
| 287 | private async end(job: AnswerJob, body: string, answer: ChatAnswer, superseded: boolean): Promise<void> { |
| 288 | const { message, channel } = job; |
| 289 | if (superseded && answer.state === "stopped" && !message.body && body.startsWith("_Stopped")) { |
| 290 | await this.dropAnswer(job); |
| 291 | return; |
| 292 | } |
| 293 | const at = new Date().toISOString(); |
| 294 | const [saved] = await this.env.DB.batch([ |
| 295 | this.env.DB.prepare("UPDATE messages SET body = ?, meta = ? WHERE id = ?").bind(body, metaColumn({ answer }), message.id), |
| 296 | this.env.DB.prepare("UPDATE channels SET last_message_at = ? WHERE id = ? AND EXISTS (SELECT 1 FROM messages WHERE id = ?)").bind(at, channel.id, message.id), |
| 297 | ]); |
| 298 | // Its message is gone: the conversation was cleared meanwhile. Nobody is shown or told it. |
| 299 | if (saved && saved.meta && saved.meta.changes === 0) return; |
| 300 | this.broadcast({ type: "message.updated", message: { ...message, body, answer } }); |
| 301 | if (answer.state === "stopped") return; |
| 302 | await notifyMessage( |
| 303 | this.env.DB, |
| 304 | this.env.NOTIFY, |
| 305 | // The person's name only matters for mentions, which g1t's answers do not notify by. |
| 306 | async (keys) => new Map(keys.map((key) => [key, key === G1T_KEY ? G1T_PROFILE : ({ kind: "user", id: key.slice(5), name: "", display_name: "", avatar: null, role: null } as MemberProfile)])), |
| 307 | { slug: job.slug, channel, row: { id: message.id, author: G1T_KEY, body, thread_root: null, created_at: message.created_at }, handles: [] }, |
| 308 | ).catch((error) => console.error("chat could not notify about", message.id, error)); |
| 309 | } |
| 310 | |
| 311 | override async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer): Promise<void> { |
| 312 | const who = socket.deserializeAttachment() as RoomMember | null; |
| 313 | if (!who || await this.ctx.storage.get("erased") || await this.ctx.storage.get(`erased:${principalKey(who.member)}`)) return; |
| 314 | // The only thing a client says over the socket: it is typing |
| 315 | // (`{"type":"typing","channel_id":"<id>"}`, at most every 3 s). |
| 316 | // Everything else (posting, reading) goes through the site, which |
| 317 | // checks it. |
| 318 | if (!isTypingFrame(message, who.channel_id)) return; |
| 319 | const now = Date.now(); |
| 320 | if (now - (this.typedAt.get(socket) ?? 0) < TYPING_EVERY_MS) return; |
| 321 | this.typedAt.set(socket, now); |
| 322 | this.broadcast( |
| 323 | { |
| 324 | type: "typing", |
| 325 | channel_id: who.channel_id, |
| 326 | member: who.member, |
| 327 | until: new Date(now + PERSON_TYPING_MS).toISOString(), |
| 328 | }, |
| 329 | principalKey(who.member), |
| 330 | ); |
| 331 | } |
| 332 | |
| 333 | override async webSocketClose(socket: WebSocket, code: number, reason: string): Promise<void> { |
| 334 | try { |
| 335 | socket.close(code, reason); |
| 336 | } catch { |
| 337 | // Already closed. |
| 338 | } |
| 339 | } |
| 340 | |
| 341 | override async webSocketError(): Promise<void> { |
| 342 | // Nothing to keep: the socket is gone, and getWebSockets no longer lists it. |
| 343 | } |
| 344 | } |