Skip to content
Sign up

flagon-io/g1t

public
0
267 linesCodeBlameRaw
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 */
16import { DurableObject } from "cloudflare:workers";
17
18import {
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
32import { runAnswer } from "./answer.ts";
33import { G1T_PROFILE, answerAllowed, metaColumn } from "./g1t.ts";
34import { PERSON_TYPING_MS, isTypingFrame } from "./messages.ts";
35import { notifyMessage } from "./notify.ts";
36
37/** Header the Worker sets on a socket it forwards: who it is, as JSON (`RoomMember`). */
38export const ROOM_MEMBER_HEADER = "x-g1t-chat-member";
39
40export type RoomMember = { channel_id: string; member: MemberProfile };
41
42/** The shortest time between two "is typing" events from one socket. */
43const TYPING_EVERY_MS = 2_000;
44/** How long g1t shows as typing before its first words, at most. */
45const G1T_TYPING_MS = 30_000;
46
47type 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 */
59export 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
68export class ChannelRoom extends DurableObject<RoomEnv> {
69 /** When each socket last said it was typing; lost on hibernation, which only lets one more through. */
70 private typedAt = new WeakMap<WebSocket, number>();
71 /** The answer being written now, to stop; one at a time per room. */
72 private running: { id: string; stop: AbortController; superseded: boolean } | null = null;
73
74 constructor(ctx: DurableObjectState, env: RoomEnv) {
75 super(ctx, env);
76 // Keepalives are answered without waking the object.
77 ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong"));
78 }
79
80 override async fetch(request: Request): Promise<Response> {
81 if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") {
82 return new Response("Expected a WebSocket upgrade\n", { status: 426 });
83 }
84 let who: RoomMember;
85 try {
86 who = JSON.parse(request.headers.get(ROOM_MEMBER_HEADER) ?? "") as RoomMember;
87 } catch {
88 return new Response("Missing member\n", { status: 400 });
89 }
90 const pair = new WebSocketPair();
91 const [client, server] = Object.values(pair) as [WebSocket, WebSocket];
92 // Tagged with the member's key, so their sockets can be found to drop.
93 this.ctx.acceptWebSocket(server, [principalKey(who.member)]);
94 server.serializeAttachment(who);
95 return new Response(null, { status: 101, webSocket: client });
96 }
97
98 /** Sends `event` to every socket in the room, except those of `except` (a principal key). */
99 broadcast(event: ChatLiveEvent, except: string | null = null): void {
100 const text = JSON.stringify(event);
101 for (const socket of this.ctx.getWebSockets()) {
102 if (except && this.ctx.getTags(socket).includes(except)) continue;
103 try {
104 socket.send(text);
105 } catch {
106 // Closing already; webSocketClose tidies up.
107 }
108 }
109 }
110
111 /** Closes a member's sockets: they left a private channel and may no longer read it. */
112 drop(principal: string): void {
113 for (const socket of this.ctx.getWebSockets(principal)) {
114 try {
115 socket.close(4403, "No longer a member");
116 } catch {
117 // Already closed.
118 }
119 }
120 }
121
122 // ── @g1t's answers (src/answer.ts) ──────────────────────────────────────
123
124 /**
125 * Answers a question in this direct message with g1t. One answer at a
126 * time: a newer question takes over, stopping the one being written
127 * (what came of it stays; one that had not started is taken away), and
128 * the new answer reads both questions. Run from an alarm, so a long
129 * answer is not cut short when the request that asked ends.
130 */
131 async startAnswer(job: AnswerJob): Promise<void> {
132 const times = answerAllowed((await this.ctx.storage.get<number[]>("answered")) ?? [], Date.now());
133 await this.ctx.storage.put("answered", times.times);
134 if (!times.ok) {
135 const answer: ChatAnswer = { state: "error", error: "Too many questions in a minute." };
136 await this.end(job, "You've asked a lot in the last minute. Wait a moment, then ask again.", answer, false);
137 return;
138 }
139 const pending = await this.ctx.storage.get<AnswerJob>("job");
140 if (pending) await this.dropAnswer(pending);
141 if (this.running) {
142 this.running.superseded = true;
143 this.running.stop.abort();
144 }
145 await this.ctx.storage.put("job", job);
146 await this.ctx.storage.setAlarm(Date.now());
147 }
148
149 /** Stops the answer being written, or about to be; what came stays. */
150 async stopAnswer(): Promise<void> {
151 const pending = await this.ctx.storage.get<AnswerJob>("job");
152 if (pending) {
153 await this.ctx.storage.delete("job");
154 await this.end(pending, "_Stopped before answering._", { ...pending.message.answer, state: "stopped" }, false);
155 }
156 this.running?.stop.abort();
157 }
158
159 override async alarm(): Promise<void> {
160 const job = await this.ctx.storage.get<AnswerJob>("job");
161 if (!job) return;
162 await this.ctx.storage.delete("job");
163 const run = { id: job.message.id, stop: new AbortController(), superseded: false };
164 this.running = run;
165 const ask = integrationsClient(this.env.INTEGRATIONS as ServiceBinding);
166 try {
167 await runAnswer(
168 {
169 slug: job.slug,
170 setup: async () => {
171 if (!this.env.INTEGRATIONS) return null;
172 const found = await ask.askSettings(job.slug, job.actor);
173 return found.ok ? found.value : { connection_id: "unknown", connection_name: null, provider: null, model: null, problem: found.error.message };
174 },
175 ask: (signal) => ask.ask(job.actor, job.slug, job.messages, job.page, signal),
176 typing: () =>
177 this.broadcast({ type: "typing", channel_id: job.channel.id, member: G1T_PROFILE, until: new Date(Date.now() + G1T_TYPING_MS).toISOString() }),
178 show: (body, answer) => this.broadcast({ type: "message.streaming", channel_id: job.channel.id, id: job.message.id, body, answer }),
179 save: (body, answer) => this.store(job.message.id, body, answer),
180 finish: (body, answer) => this.end(job, body, answer, run.superseded),
181 now: () => Date.now(),
182 },
183 run.stop.signal,
184 );
185 } catch (error) {
186 console.error("g1t could not answer in", job.channel.id, error);
187 } finally {
188 if (this.running === run) this.running = null;
189 }
190 }
191
192 /** Saves an answer as it is now. */
193 private async store(id: string, body: string, answer: ChatAnswer): Promise<void> {
194 await this.env.DB.prepare("UPDATE messages SET body = ?, meta = ? WHERE id = ? AND deleted_at IS NULL")
195 .bind(body, metaColumn({ answer }), id)
196 .run();
197 }
198
199 /** A question taken over before its answer started: its empty message goes. */
200 private async dropAnswer(job: AnswerJob): Promise<void> {
201 await this.ctx.storage.delete("job");
202 await this.env.DB.prepare("UPDATE messages SET deleted_at = ?, body = '' WHERE id = ?").bind(new Date().toISOString(), job.message.id).run();
203 this.broadcast({ type: "message.deleted", channel_id: job.channel.id, id: job.message.id });
204 }
205
206 /**
207 * Ends an answer: stored as a normal message, sent to the room, and the
208 * person notified (their direct message notifies them, as any would).
209 * One taken over by a newer question before a word of it came is taken
210 * away instead.
211 */
212 private async end(job: AnswerJob, body: string, answer: ChatAnswer, superseded: boolean): Promise<void> {
213 const { message, channel } = job;
214 if (superseded && answer.state === "stopped" && !message.body && body.startsWith("_Stopped")) {
215 await this.dropAnswer(job);
216 return;
217 }
218 const at = new Date().toISOString();
219 await this.env.DB.batch([
220 this.env.DB.prepare("UPDATE messages SET body = ?, meta = ? WHERE id = ?").bind(body, metaColumn({ answer }), message.id),
221 this.env.DB.prepare("UPDATE channels SET last_message_at = ? WHERE id = ?").bind(at, channel.id),
222 ]);
223 this.broadcast({ type: "message.updated", message: { ...message, body, answer } });
224 if (answer.state === "stopped") return;
225 await notifyMessage(
226 this.env.DB,
227 this.env.NOTIFY,
228 // The person's name only matters for mentions, which g1t's answers do not notify by.
229 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)])),
230 { slug: job.slug, channel, row: { id: message.id, author: G1T_KEY, body, thread_root: null, created_at: message.created_at }, handles: [] },
231 ).catch((error) => console.error("chat could not notify about", message.id, error));
232 }
233
234 override async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer): Promise<void> {
235 const who = socket.deserializeAttachment() as RoomMember | null;
236 if (!who) return;
237 // The only thing a client says over the socket: it is typing
238 // (`{"type":"typing","channel_id":"<id>"}`, at most every 3 s).
239 // Everything else (posting, reading) goes through the site, which
240 // checks it.
241 if (!isTypingFrame(message, who.channel_id)) return;
242 const now = Date.now();
243 if (now - (this.typedAt.get(socket) ?? 0) < TYPING_EVERY_MS) return;
244 this.typedAt.set(socket, now);
245 this.broadcast(
246 {
247 type: "typing",
248 channel_id: who.channel_id,
249 member: who.member,
250 until: new Date(now + PERSON_TYPING_MS).toISOString(),
251 },
252 principalKey(who.member),
253 );
254 }
255
256 override async webSocketClose(socket: WebSocket, code: number, reason: string): Promise<void> {
257 try {
258 socket.close(code, reason);
259 } catch {
260 // Already closed.
261 }
262 }
263
264 override async webSocketError(): Promise<void> {
265 // Nothing to keep: the socket is gone, and getWebSockets no longer lists it.
266 }
267}