Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp
898 linesCodeBlameRaw
1/**
2 * One person's feed, as a Durable Object named by their user id.
3 *
4 * It holds a socket per open tab with the WebSocket Hibernation API, so a
5 * person with g1t open all day costs nothing between notifications. Plain
6 * `ping` keepalives are answered at the edge without waking it, and the
7 * time of the last one says whether a tab is still there
8 * (`getWebSocketAutoResponseTimestamp`). Each socket carries the tab's
9 * focus and page (`serializeAttachment`), which survive hibernation.
10 *
11 * Kept in the object's own SQLite storage: the latest notifications, the
12 * unread counts per conversation, push subscriptions (browsers) and
13 * devices (the phone app, pushed through Expo: src/expopush.ts) and
14 * preferences, and
15 * the person's status, Do Not Disturb and whether they set themselves away,
16 * and when they were last on each workspace's Home page (src/visits.ts),
17 * and which repositories' pages their tabs are watching (src/watch.ts).
18 *
19 * Presence is worked out here from the tabs (src/presence.ts) and told to
20 * the room of every workspace the person belongs to (src/room.ts), which
21 * passes it to everyone there who is online. A status or Do Not Disturb
22 * that runs out is cleared by an alarm and told the same way.
23 *
24 * Live repository pages: a tab on a repository's page has the site ask
25 * the feed to watch it (`watchRepo`), which tells the repository's topic
26 * (src/topic.ts); the topic hands back what changes there (`repoChanged`),
27 * and the feed sends it to the tabs as a `repo` event. A workspace's Work
28 * pages the same way, on the workspace's topic (`watchWorkspace`,
29 * `workspaceChanged`, a `workspace` event).
30 *
31 * The feed authorizes nothing: the Worker reaches it only for the person
32 * the site checked (src/index.ts).
33 */
34import { DurableObject } from "cloudflare:workers";
35
36import {
37 NOTIFY_SEED_HEADER,
38 type ChannelCounts,
39 type OwnPresence,
40 type PresenceChange,
41 type PresenceEntry,
42 type FeedDelivery,
43 type FeedEvent,
44 type FeedNotification,
45 type FeedSeed,
46 type RepoChange,
47 type WorkspaceChange,
48 type NotifyPreferences,
49 type NotifyStatus,
50 type PushDevice,
51 type PushSubscriptionJson,
52} from "@g1t/contracts";
53
54import { applyCounts, totals } from "./counts.ts";
55import { beforeForgetting, forgotKey } from "./forget.ts";
56import { cleanDevice, sendExpo } from "./expopush.ts";
57import { cleanNotification, decide, mergePreferences, pushPayload, readPreferences, type PushPayload, type TabState } from "./prefs.ts";
58import {
59 OFFLINE_GRACE_MS,
60 applyChange,
61 cleanWorkspaces,
62 current,
63 dndOn,
64 entryOf,
65 nextExpiry,
66 ownOf,
67 presenceOf,
68 readKept,
69 sameEntry,
70 type Kept,
71} from "./presence.ts";
72import type { Room } from "./room.ts";
73import type { Topic } from "./topic.ts";
74import { type Visit, lastVisit, markVisit } from "./visits.ts";
75import { MAX_WATCHING, stillWatching, watchUntil, workspaceTopic } from "./watch.ts";
76import { sendPush, type Vapid } from "./webpush.ts";
77
78export type FeedEnv = {
79 VAPID_PUBLIC_KEY?: string;
80 VAPID_PRIVATE_KEY?: string;
81 VAPID_SUBJECT?: string;
82 /**
83 * A secret, optional: the access token Expo's push service asks for once
84 * the Expo project turns on enhanced push security. Without it, pushes to
85 * phones go unauthenticated, which Expo takes otherwise.
86 */
87 EXPO_ACCESS_TOKEN?: string;
88 /**
89 * `off` sends phones nothing: a self-hosted g1t that would rather not pass
90 * notifications through Expo's push service (and Google's and Apple's)
91 * sets PHONE_PUSH=off (deploy/self-host/configs.mjs). Phones then do not
92 * register, and the site's /.well-known/g1t.json says so to the app.
93 */
94 PHONE_PUSH?: string;
95 /** One presence room per workspace (src/room.ts); without it, presence stays in the person's own tabs. */
96 ROOMS?: DurableObjectNamespace<Room>;
97 /** One live topic per repository (src/topic.ts); without it, repository pages refresh on their own. */
98 TOPICS?: DurableObjectNamespace<Topic>;
99};
100
101/** The headers the Worker passes the person's id and username in, with a socket (src/index.ts). */
102export const FEED_USERNAME_HEADER = "x-g1t-notify-username";
103export const FEED_USER_ID_HEADER = "x-g1t-notify-user-id";
104
105/** What each socket carries: focus, page, whether it has gone idle, and the workspace it is open in. */
106type Tab = { focused: boolean; path: string; at: number; idle?: boolean; workspace?: string | null };
107
108/** Notifications kept for a tab that opens later. */
109const KEPT = 100;
110/** How long after a workspace is purged its counts, seeds and visits, which carry no time, are refused. */
111const ERASED_WORKSPACE_MS = 24 * 60 * 60 * 1000;
112/** Sent with `hello`: the latest few. */
113const HELLO = 20;
114/** The most browsers one person gets pushes on. */
115const MAX_SUBSCRIPTIONS = 20;
116/** The longest endpoint taken. */
117const MAX_ENDPOINT = 2000;
118/** The most phones one person gets pushes on. */
119const MAX_DEVICES = 10;
120
121export class Feed extends DurableObject<FeedEnv> {
122 private readonly sql: SqlStorage;
123
124 constructor(ctx: DurableObjectState, env: FeedEnv) {
125 super(ctx, env);
126 this.sql = ctx.storage.sql;
127 // Keepalives are answered without waking the object.
128 ctx.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong"));
129 this.sql.exec(`CREATE TABLE IF NOT EXISTS notifications (id TEXT PRIMARY KEY, at INTEGER NOT NULL, json TEXT NOT NULL)`);
130 this.sql.exec(`CREATE INDEX IF NOT EXISTS notifications_at ON notifications (at)`);
131 this.sql.exec(
132 `CREATE TABLE IF NOT EXISTS counts (workspace TEXT NOT NULL, channel_id TEXT NOT NULL, unread INTEGER NOT NULL, mentions INTEGER NOT NULL, muted INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (workspace, channel_id))`,
133 );
134 this.sql.exec(
135 `CREATE TABLE IF NOT EXISTS subscriptions (endpoint TEXT PRIMARY KEY, p256dh TEXT NOT NULL, auth TEXT NOT NULL, user_agent TEXT, created_at INTEGER NOT NULL)`,
136 );
137 // The phone app's Expo push tokens (src/expopush.ts).
138 this.sql.exec(`CREATE TABLE IF NOT EXISTS devices (token TEXT PRIMARY KEY, platform TEXT NOT NULL, name TEXT, created_at INTEGER NOT NULL)`);
139 this.sql.exec(`CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)`);
140 // When the person was last on each workspace's Home page (src/visits.ts).
141 this.sql.exec(`CREATE TABLE IF NOT EXISTS visits (workspace TEXT PRIMARY KEY, seen_at INTEGER NOT NULL, previous_at INTEGER)`);
142 // The repositories whose pages the person's tabs are watching, until when (src/watch.ts).
143 this.sql.exec("CREATE TABLE IF NOT EXISTS presence_rooms (workspace TEXT PRIMARY KEY)");
144 this.sql.exec(`CREATE TABLE IF NOT EXISTS watching (repo_id TEXT PRIMARY KEY, until INTEGER NOT NULL)`);
145 }
146
147 private erased(): boolean { return this.meta("erased") === "1"; }
148
149 /** When `workspace` was purged (RFC 3339), if it was. */
150 private workspaceErasedAt(workspace: string | null | undefined): string | null {
151 return workspace ? this.meta(`erased_workspace:${workspace.toLowerCase()}`) : null;
152 }
153
154 /** Purged within the last day: a count, seed or visit, which carry no time, is a late one from before it. */
155 private workspaceGone(workspace: string | null | undefined): boolean {
156 const at = Date.parse(this.workspaceErasedAt(workspace) ?? "");
157 return Number.isFinite(at) && Date.now() - at < ERASED_WORKSPACE_MS;
158 }
159
160 /** A notification of a purged workspace from before the purge. */
161 private fromErasedWorkspace(notification: Pick<FeedNotification, "workspace" | "created_at">): boolean {
162 return beforeForgetting(notification, this.workspaceErasedAt(notification.workspace));
163 }
164
165 /** A permanent tombstone prevents delayed deliveries or stale sockets recreating a feed. */
166 async erase(userId: string): Promise<{ ok: boolean }> {
167 // Keep cleanup references until remote removals succeed, so any failure is retryable.
168 if (!this.erased()) {
169 this.setMeta("erase_rooms", JSON.stringify([...new Set([...this.memberOf(), ...this.workspaces(), ...this.sql.exec<{ workspace: string }>("SELECT workspace FROM visits UNION SELECT workspace FROM presence_rooms").toArray().map(r => r.workspace)])]));
170 this.setMeta("erased", "1");
171 }
172 for (const socket of this.ctx.getWebSockets()) { try { socket.close(4403, "Account erased"); } catch {} }
173 await this.ctx.storage.deleteAlarm();
174 const rooms = JSON.parse(this.meta("erase_rooms") ?? "[]") as string[];
175 for (const workspace of rooms) {
176 // Without the namespace there is no room to clean.
177 if (!this.env.ROOMS) break;
178 await this.env.ROOMS.get(this.env.ROOMS.idFromName(workspace)).forget(userId);
179 }
180 for (const row of this.sql.exec<{ repo_id: string }>("SELECT repo_id FROM watching").toArray()) {
181 if (!this.env.TOPICS) break;
182 await this.env.TOPICS.get(this.env.TOPICS.idFromName(row.repo_id)).forget(userId);
183 }
184 this.ctx.storage.transactionSync(() => {
185 for (const table of ["notifications", "counts", "subscriptions", "devices", "visits", "watching", "presence_rooms"]) this.sql.exec(`DELETE FROM ${table}`);
186 this.sql.exec("DELETE FROM meta WHERE key != 'erased'");
187 });
188 return { ok: true };
189 }
190
191 /**
192 * A workspace was purged. Its slug can be made again (by the person whose
193 * username it is), so the mark is a time, not a ban: what was said before
194 * it is dropped, and only for a day are counts and seeds without a time
195 * of their own refused (`workspaceGone`).
196 */
197 async eraseWorkspace(workspace: string): Promise<{ ok: boolean }> {
198 if (this.erased()) return { ok: true };
199 workspace = workspace.toLowerCase();
200 this.setMeta(`erased_workspace:${workspace}`, new Date().toISOString());
201 this.sql.exec("DELETE FROM notifications WHERE json_extract(json, '$.workspace') = ?", workspace);
202 this.sql.exec("DELETE FROM meta WHERE key = ?1 OR substr(key, 1, length(?2)) = ?2", `complete:${workspace}`, `forgot:${workspace}:`);
203 const preferences = this.preferences();
204 if (workspace in preferences.workspaces) {
205 const { [workspace]: _gone, ...rest } = preferences.workspaces;
206 this.setMeta("preferences", JSON.stringify({ ...preferences, workspaces: rest }));
207 }
208 this.sql.exec("DELETE FROM counts WHERE workspace = ?", workspace);
209 this.sql.exec("DELETE FROM visits WHERE workspace = ?", workspace);
210 this.sql.exec("DELETE FROM presence_rooms WHERE workspace = ?", workspace);
211 this.setMeta("workspaces", JSON.stringify(this.memberOf().filter(w => w !== workspace)));
212 // Force clients to reload counts instead of retaining the deleted workspace's snapshot.
213 for (const socket of this.ctx.getWebSockets()) { try { socket.close(1012, "Workspace erased"); } catch {} }
214 return { ok: true };
215 }
216
217 // ── Kept state ──────────────────────────────────────────────────────────
218
219 private meta(key: string): string | null {
220 const row = this.sql.exec<{ value: string }>("SELECT value FROM meta WHERE key = ?", key).toArray()[0];
221 return row?.value ?? null;
222 }
223
224 private setMeta(key: string, value: string): void {
225 if (key !== "erased" && this.erased()) return;
226 this.sql.exec("INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value", key, value);
227 }
228
229 private preferences(): NotifyPreferences {
230 return readPreferences(this.meta("preferences"));
231 }
232
233 private inboxUnread(): number {
234 return Number(this.meta("inbox_unread") ?? 0) || 0;
235 }
236
237 private channelRows(workspace: string): ChannelCounts[] {
238 return this.sql
239 .exec<{ channel_id: string; unread: number; mentions: number; muted: number }>(
240 "SELECT channel_id, unread, mentions, muted FROM counts WHERE workspace = ?",
241 workspace,
242 )
243 .toArray()
244 .map((r) => ({ channel_id: r.channel_id, unread: r.unread, mentions: r.mentions, muted: !!r.muted }));
245 }
246
247 private counts(workspace: string): FeedEvent {
248 return { type: "counts", ...totals(workspace, this.channelRows(workspace), this.inboxUnread(), this.meta(`complete:${workspace}`) === "1") };
249 }
250
251 private workspaces(): string[] {
252 return this.sql.exec<{ workspace: string }>("SELECT DISTINCT workspace FROM counts").toArray().map((r) => r.workspace);
253 }
254
255 private vapid(): Vapid | null {
256 const { VAPID_PUBLIC_KEY, VAPID_PRIVATE_KEY, VAPID_SUBJECT } = this.env;
257 return VAPID_PUBLIC_KEY && VAPID_PRIVATE_KEY ? { publicKey: VAPID_PUBLIC_KEY, privateKey: VAPID_PRIVATE_KEY, subject: VAPID_SUBJECT || "https://g1t.sh" } : null;
258 }
259
260 private subscriptionCount(): number {
261 return this.sql.exec<{ n: number }>("SELECT COUNT(*) AS n FROM subscriptions").one().n;
262 }
263
264 private deviceCount(): number {
265 if (!this.phonePush()) return 0;
266 return this.sql.exec<{ n: number }>("SELECT COUNT(*) AS n FROM devices").one().n;
267 }
268
269 /** Whether this g1t sends phones anything (PHONE_PUSH). */
270 private phonePush(): boolean {
271 return this.env.PHONE_PUSH !== "off";
272 }
273
274 // ── Presence ────────────────────────────────────────────────────────────
275
276 private kept(): Kept {
277 return readKept(this.meta("presence"));
278 }
279
280 /** Who this feed is for, as the site last said. */
281 private person(): { user_id: string; username: string } {
282 return { user_id: this.meta("user_id") ?? "", username: this.meta("username") ?? "" };
283 }
284
285 /** The open tabs as presence sees them, leaving out one that is closing. */
286 private presenceTabs(closing?: WebSocket): { idle: boolean }[] {
287 return this.ctx
288 .getWebSockets()
289 .filter((socket) => socket !== closing && socket.readyState === WebSocket.OPEN)
290 .map((socket) => ({ idle: (socket.deserializeAttachment() as Tab | null)?.idle === true }));
291 }
292
293 private entry(closing?: WebSocket): PresenceEntry {
294 const kept = this.kept();
295 return entryOf(this.person(), presenceOf(this.presenceTabs(closing), kept.away_manual), kept, Date.now());
296 }
297
298 private room(workspace: string) {
299 const rooms = this.env.ROOMS;
300 return rooms ? rooms.get(rooms.idFromName(workspace)) : null;
301 }
302
303 private memberOf(): string[] {
304 try {
305 return cleanWorkspaces(JSON.parse(this.meta("workspaces") ?? "[]")).filter(workspace => !this.workspaceGone(workspace));
306 } catch {
307 return [];
308 }
309 }
310
311 /**
312 * Works presence out again and, when how the person shows has changed,
313 * tells their own tabs and every workspace's room. `force` tells them
314 * anyway (a tab just connected, so a room may have missed the last word).
315 */
316 private async refresh(options: { closing?: WebSocket; force?: boolean } = {}): Promise<PresenceEntry> {
317 if (this.erased()) return this.entry();
318 const now = Date.now();
319 // What ran out goes, so nobody is told of it again.
320 const kept = this.kept();
321 const live = current(kept, now);
322 if (JSON.stringify(live) !== JSON.stringify(kept)) this.setMeta("presence", JSON.stringify(live));
323 const entry = this.entry(options.closing);
324 let before: PresenceEntry | null = null;
325 try {
326 before = JSON.parse(this.meta("reported") ?? "null") as PresenceEntry | null;
327 } catch {
328 before = null;
329 }
330 const changed = !sameEntry(before, entry);
331 if (changed || options.force) {
332 for (const workspace of this.memberOf()) this.sql.exec("INSERT OR IGNORE INTO presence_rooms (workspace) VALUES (?)", workspace);
333 this.setMeta("reported", JSON.stringify(entry));
334 this.send({ type: "me", me: ownOf(entry, live) }, options.closing);
335 // Without an id there is nobody to name: the site always sends one with a socket.
336 if (entry.user_id) await Promise.allSettled(this.memberOf().map((workspace) => this.room(workspace)?.report(workspace, entry)));
337 }
338 await this.schedule();
339 return entry;
340 }
341
342 /** The next alarm: when a status or Do Not Disturb runs out, or when a person whose last tab closed shows offline. */
343 private async schedule(): Promise<void> {
344 if (this.erased()) return;
345 const now = Date.now();
346 const times = [nextExpiry(this.kept(), now), Number(this.meta("offline_at") ?? 0) || null].filter((at): at is number => at != null && at > now);
347 if (times.length) await this.ctx.storage.setAlarm(Math.min(...times));
348 else await this.ctx.storage.deleteAlarm();
349 }
350
351 override async alarm(): Promise<void> {
352 if (this.erased()) return;
353 const offlineAt = Number(this.meta("offline_at") ?? 0);
354 if (offlineAt && offlineAt <= Date.now()) this.setMeta("offline_at", "0");
355 await this.refresh();
356 }
357
358 /** A tab just connected: everyone hears it is here, and it hears how everyone in its workspace shows. */
359 private async connected(socket: WebSocket, workspace: string | null): Promise<void> {
360 if (this.erased()) return;
361 this.setMeta("offline_at", "0");
362 const entry = await this.refresh({ force: true });
363 if (!workspace || !entry.user_id) return;
364 const people = await this.room(workspace)
365 ?.report(workspace, entry, true)
366 .catch((error: unknown) => {
367 console.error("notify: reading a presence room failed", error);
368 return null;
369 });
370 if (!people) return;
371 try {
372 socket.send(JSON.stringify({ type: "presence", workspace, people, full: true } satisfies FeedEvent));
373 } catch {
374 // Gone already.
375 }
376 }
377
378 // ── Sockets ─────────────────────────────────────────────────────────────
379
380 private tabs(): TabState[] {
381 return this.ctx.getWebSockets().map((socket) => {
382 const tab = (socket.deserializeAttachment() as Tab | null) ?? { focused: false, path: "", at: 0 };
383 const pinged = this.ctx.getWebSocketAutoResponseTimestamp(socket)?.getTime() ?? 0;
384 return { focused: tab.focused, seen_at: Math.max(tab.at, pinged) };
385 });
386 }
387
388 private send(event: FeedEvent, except?: WebSocket): void {
389 const text = JSON.stringify(event);
390 for (const socket of this.ctx.getWebSockets()) {
391 if (socket === except) continue;
392 try {
393 socket.send(text);
394 } catch {
395 // Closing already.
396 }
397 }
398 }
399
400 override async fetch(request: Request): Promise<Response> {
401 if (this.erased()) return new Response("Account erased", { status: 410 });
402 if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") {
403 return new Response("Expected a WebSocket upgrade\n", { status: 426 });
404 }
405 let seed: FeedSeed | null = null;
406 try {
407 seed = JSON.parse(request.headers.get(NOTIFY_SEED_HEADER) ?? "null") as FeedSeed | null;
408 } catch {
409 seed = null;
410 }
411 if (seed) this.applySeed(seed);
412 const username = request.headers.get(FEED_USERNAME_HEADER);
413 const userId = request.headers.get(FEED_USER_ID_HEADER);
414 this.remember({ user_id: userId ?? "", username: username ?? "" });
415 if (Array.isArray(seed?.workspaces)) this.setMeta("workspaces", JSON.stringify(cleanWorkspaces(seed.workspaces)));
416 const workspace = seed?.workspace?.toLowerCase() || null;
417 const pair = new WebSocketPair();
418 const [client, server] = Object.values(pair) as [WebSocket, WebSocket];
419 this.ctx.acceptWebSocket(server);
420 server.serializeAttachment({ focused: false, path: "", at: Date.now(), idle: false, workspace } satisfies Tab);
421 const hello: FeedEvent = {
422 type: "hello",
423 notifications: this.latest(HELLO),
424 vapid_public_key: this.env.VAPID_PUBLIC_KEY || null,
425 preferences: this.preferences(),
426 };
427 server.send(JSON.stringify(hello));
428 const shown = new Set(this.workspaces());
429 if (seed?.workspace) shown.add(seed.workspace);
430 for (const workspace of shown) server.send(JSON.stringify(this.counts(workspace)));
431 // Presence after the socket is handed back: the rooms are not waited on.
432 this.ctx.waitUntil(this.connected(server, workspace));
433 return new Response(null, { status: 101, webSocket: client });
434 }
435
436 /** Counts read from chat and the inbox just now replace what was kept for that workspace. */
437 private applySeed(seed: FeedSeed): void {
438 if (typeof seed.inbox_unread === "number") this.setMeta("inbox_unread", String(Math.max(0, Math.floor(seed.inbox_unread))));
439 const workspace = seed.workspace?.toLowerCase();
440 if (!workspace || this.workspaceGone(workspace) || !Array.isArray(seed.per_channel)) return;
441 this.ctx.storage.transactionSync(() => {
442 this.sql.exec("DELETE FROM counts WHERE workspace = ?", workspace);
443 for (const row of seed.per_channel!.slice(0, 2000)) {
444 if (typeof row?.channel_id !== "string") continue;
445 const kept = applyCounts(null, { ...row, set: true });
446 this.sql.exec(
447 "INSERT OR REPLACE INTO counts (workspace, channel_id, unread, mentions, muted) VALUES (?, ?, ?, ?, ?)",
448 workspace,
449 kept.channel_id,
450 kept.unread,
451 kept.mentions,
452 kept.muted ? 1 : 0,
453 );
454 }
455 this.setMeta(`complete:${workspace}`, "1");
456 });
457 }
458
459 override async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer): Promise<void> {
460 if (this.erased()) return;
461 if (typeof message !== "string" || message.length > 4096) return;
462 let frame: Record<string, unknown>;
463 try {
464 frame = JSON.parse(message);
465 } catch {
466 return;
467 }
468 if (frame.type === "state") {
469 const before = (socket.deserializeAttachment() as Tab | null) ?? { focused: false, path: "", at: 0 };
470 const idle = frame.idle === true;
471 socket.serializeAttachment({
472 focused: frame.focused === true,
473 path: String(frame.path ?? "").slice(0, 500),
474 at: Date.now(),
475 idle,
476 workspace: before.workspace ?? null,
477 } satisfies Tab);
478 // Gone idle, or back: away or active.
479 if (idle !== (before.idle === true)) await this.refresh();
480 } else if (frame.type === "inbox" && Number.isFinite(frame.unread)) {
481 // The inbox count a page just read: every tab shows it at once.
482 await this.setInbox(Number(frame.unread));
483 }
484 }
485
486 override async webSocketClose(socket: WebSocket, code: number, reason: string): Promise<void> {
487 try {
488 socket.close(code, reason);
489 } catch {
490 // Already closed.
491 }
492 await this.left(socket);
493 }
494
495 override async webSocketError(socket: WebSocket): Promise<void> {
496 await this.left(socket);
497 }
498
499 /**
500 * A tab went. The last one leaves the person showing as they were for a
501 * moment, so a reload or a switch of workspace is not leaving; the alarm
502 * then says offline if no tab came back.
503 */
504 private async left(socket: WebSocket): Promise<void> {
505 if (this.erased()) return;
506 if (this.presenceTabs(socket).length === 0) {
507 this.setMeta("offline_at", String(Date.now() + OFFLINE_GRACE_MS));
508 await this.schedule();
509 return;
510 }
511 await this.refresh({ closing: socket });
512 }
513
514 // ── Notifications ───────────────────────────────────────────────────────
515
516 private latest(limit: number): FeedNotification[] {
517 return this.sql
518 .exec<{ json: string }>("SELECT json FROM notifications ORDER BY at DESC LIMIT ?", limit)
519 .toArray()
520 .map((r) => JSON.parse(r.json) as FeedNotification);
521 }
522
523 /** Keeps it, unless it was told before. True when it is news. */
524 private keep(notification: FeedNotification): boolean {
525 const had = this.sql.exec("SELECT 1 FROM notifications WHERE id = ?", notification.id).toArray().length > 0;
526 if (had) return false;
527 this.sql.exec("INSERT INTO notifications (id, at, json) VALUES (?, ?, ?)", notification.id, Date.now(), JSON.stringify(notification));
528 this.sql.exec("DELETE FROM notifications WHERE id NOT IN (SELECT id FROM notifications ORDER BY at DESC LIMIT ?)", KEPT);
529 return true;
530 }
531
532 /** Tells the person: open tabs at once, then browsers and phones when none is in front of them. */
533 private async tell(notification: FeedNotification, test = false): Promise<number> {
534 if (this.erased() || this.fromErasedWorkspace(notification)) return 0;
535 if (!test && !this.keep(notification)) return 0;
536 // Somewhere to push: a browser or a phone.
537 const subscriptions = this.subscriptionCount() + this.deviceCount();
538 const now = Date.now();
539 // Do Not Disturb: kept and counted, neither toasted nor pushed.
540 const quiet = dndOn(this.kept().dnd_until, now);
541 const decision = decide({ prefs: this.preferences(), notification, tabs: this.tabs(), subscriptions, now, test, dnd: quiet });
542 this.send({ type: "notification", notification, toast: decision.toast });
543 return decision.push ? this.push(notification) : 0;
544 }
545
546 /** Pushes to every browser and phone at once; one failing does not stop the other. */
547 private async push(notification: FeedNotification): Promise<number> {
548 const payload = pushPayload(notification);
549 const results = await Promise.allSettled([this.pushBrowsers(payload), this.pushDevices(notification, payload)]);
550 let sent = 0;
551 for (const result of results) {
552 if (result.status === "fulfilled") sent += result.value;
553 else console.error("notify: pushing failed", result.reason);
554 }
555 return sent;
556 }
557
558 /** Pushes to every browser subscribed; drops the ones the push service says are gone. */
559 private async pushBrowsers(payload: PushPayload): Promise<number> {
560 const vapid = this.vapid();
561 if (!vapid) return 0;
562 const subscriptions = this.sql.exec<{ endpoint: string; p256dh: string; auth: string }>("SELECT endpoint, p256dh, auth FROM subscriptions").toArray();
563 const results = await Promise.allSettled(
564 subscriptions.map((s) => sendPush(s, payload, { vapid, urgency: payload.urgent ? "high" : "normal", topic: payload.tag, ttl: 12 * 3600 })),
565 );
566 let sent = 0;
567 for (const result of results) {
568 if (result.status === "rejected") {
569 console.error("notify: a push failed", result.reason);
570 continue;
571 }
572 if (result.value.gone) this.sql.exec("DELETE FROM subscriptions WHERE endpoint = ?", result.value.endpoint);
573 else if (result.value.status < 300) sent++;
574 else console.error("notify: a push service answered", result.value.status);
575 }
576 return sent;
577 }
578
579 /** Pushes to every phone through Expo; drops the ones Expo says the app is gone from. */
580 private async pushDevices(notification: FeedNotification, payload: PushPayload): Promise<number> {
581 if (!this.phonePush()) return 0;
582 const tokens = this.sql.exec<{ token: string }>("SELECT token FROM devices").toArray().map((r) => r.token);
583 if (!tokens.length) return 0;
584 const result = await sendExpo(tokens, payload, notification, { accessToken: this.env.EXPO_ACCESS_TOKEN });
585 for (const token of result.gone) this.sql.exec("DELETE FROM devices WHERE token = ?", token);
586 for (const error of result.errors) console.error("notify: a push to a phone failed", error);
587 return result.sent;
588 }
589
590 // ── RPC (called by the Worker, src/index.ts) ────────────────────────────
591
592 // Deliveries to an erased person are dropped quietly: the sender did nothing
593 // wrong, and a refusal would only be logged as a failure and retried.
594 async notify(value: unknown): Promise<{ ok: boolean }> {
595 if (this.erased()) return { ok: true };
596 const notification = cleanNotification(value);
597 if (!notification) return { ok: false };
598 if (this.fromErasedWorkspace(notification)) return { ok: true };
599 await this.tell(notification);
600 return { ok: true };
601 }
602
603 /** A batch for this person from chat: counts moved, notifications told. */
604 async deliver(items: FeedDelivery[]): Promise<{ ok: boolean }> {
605 if (this.erased()) return { ok: true };
606 const changed = new Set<string>();
607 const told: FeedNotification[] = [];
608 for (const item of items) {
609 const workspace = String(item.workspace ?? "").toLowerCase();
610 if (item.notification ? this.fromErasedWorkspace({ ...item.notification, workspace }) : this.workspaceGone(workspace)) continue;
611 // Of a message from a conversation cleared since: it is gone, so neither counted nor told.
612 const channel = item.counts?.channel_id ?? item.notification?.channel_id ?? null;
613 if (channel && workspace && item.notification && beforeForgetting(item.notification, this.meta(forgotKey(workspace, channel)))) continue;
614 if (item.counts?.channel_id && workspace) {
615 const before = this.sql
616 .exec<{ unread: number; mentions: number; muted: number }>(
617 "SELECT unread, mentions, muted FROM counts WHERE workspace = ? AND channel_id = ?",
618 workspace,
619 item.counts.channel_id,
620 )
621 .toArray()[0];
622 const after = applyCounts(
623 before ? { channel_id: item.counts.channel_id, unread: before.unread, mentions: before.mentions, muted: !!before.muted } : null,
624 item.counts,
625 );
626 this.sql.exec(
627 "INSERT OR REPLACE INTO counts (workspace, channel_id, unread, mentions, muted) VALUES (?, ?, ?, ?, ?)",
628 workspace,
629 after.channel_id,
630 after.unread,
631 after.mentions,
632 after.muted ? 1 : 0,
633 );
634 changed.add(workspace);
635 }
636 const notification = item.notification ? cleanNotification(item.notification) : null;
637 if (notification) told.push(notification);
638 }
639 for (const workspace of changed) this.send(this.counts(workspace));
640 await Promise.all(told.map((n) => this.tell(n)));
641 return { ok: true };
642 }
643
644 /**
645 * The person cleared a conversation (chat's `g1t_clear`): the
646 * notifications kept from it go, its counts go to nothing, and from now
647 * on a late one of a message from before is dropped (src/forget.ts).
648 * Pushes already shown by a browser or phone stay there: they cannot be
649 * taken back.
650 */
651 async forgetConversation(workspace: string, channelId: string): Promise<{ ok: boolean; forgotten: number }> {
652 if (this.erased()) return { ok: true, forgotten: 0 };
653 const slug = workspace.toLowerCase();
654 let forgotten = 0;
655 this.ctx.storage.transactionSync(() => {
656 forgotten = this.sql.exec("DELETE FROM notifications WHERE json_extract(json, '$.channel_id') = ?", channelId).rowsWritten;
657 this.sql.exec("UPDATE counts SET unread = 0, mentions = 0 WHERE workspace = ? AND channel_id = ?", slug, channelId);
658 this.setMeta(forgotKey(slug, channelId), new Date().toISOString());
659 });
660 this.send({ type: "forgotten", workspace: slug, channel_id: channelId });
661 this.send(this.counts(slug));
662 return { ok: true, forgotten };
663 }
664
665 /**
666 * The inbox count as it now is (events, after items arrive or are
667 * marked): every tab shows it at once. With `settled`, the threads of
668 * items that settled by themselves: their notifications are no longer
669 * kept to replay to a tab that opens later, and open tabs drop them.
670 */
671 async setInbox(value: number, settled: string[] = []): Promise<{ ok: boolean }> {
672 if (this.erased()) return { ok: true };
673 const unread = Number.isFinite(value) ? Math.max(0, Math.floor(value)) : 0;
674 if (settled.length > 0) {
675 this.sql.exec(
676 "DELETE FROM notifications WHERE json_extract(json, '$.inbox_thread') IN (SELECT value FROM json_each(?))",
677 JSON.stringify(settled),
678 );
679 }
680 if (unread === this.inboxUnread() && settled.length === 0) return { ok: true };
681 this.setMeta("inbox_unread", String(unread));
682 this.send(settled.length > 0 ? { type: "inbox", unread, settled } : { type: "inbox", unread });
683 return { ok: true };
684 }
685
686 async subscribe(subscription: PushSubscriptionJson, userAgent: string | null): Promise<{ ok: boolean }> {
687 if (this.erased()) throw new Error("Account erased");
688 const endpoint = String(subscription?.endpoint ?? "");
689 const { p256dh, auth } = subscription?.keys ?? ({} as PushSubscriptionJson["keys"]);
690 if (!/^https:\/\//.test(endpoint) || endpoint.length > MAX_ENDPOINT || typeof p256dh !== "string" || typeof auth !== "string") return { ok: false };
691 this.sql.exec(
692 "INSERT OR REPLACE INTO subscriptions (endpoint, p256dh, auth, user_agent, created_at) VALUES (?, ?, ?, ?, ?)",
693 endpoint,
694 p256dh.slice(0, 200),
695 auth.slice(0, 100),
696 userAgent ? userAgent.slice(0, 300) : null,
697 Date.now(),
698 );
699 // The oldest go first past the limit.
700 this.sql.exec("DELETE FROM subscriptions WHERE endpoint NOT IN (SELECT endpoint FROM subscriptions ORDER BY created_at DESC LIMIT ?)", MAX_SUBSCRIPTIONS);
701 return { ok: true };
702 }
703
704 async unsubscribe(endpoint: string): Promise<{ ok: boolean }> {
705 if (this.erased()) return { ok: true };
706 this.sql.exec("DELETE FROM subscriptions WHERE endpoint = ?", String(endpoint ?? ""));
707 return { ok: true };
708 }
709
710 /** A phone the app runs on, by its Expo push token; registering it again moves it to the newest. */
711 async registerDevice(device: PushDevice): Promise<{ ok: boolean; off?: boolean }> {
712 if (this.erased()) throw new Error("Account erased");
713 if (!this.phonePush()) return { ok: false, off: true };
714 const clean = cleanDevice(device);
715 if (!clean) return { ok: false };
716 this.sql.exec(
717 "INSERT OR REPLACE INTO devices (token, platform, name, created_at) VALUES (?, ?, ?, ?)",
718 clean.token,
719 clean.platform,
720 clean.name,
721 Date.now(),
722 );
723 // The oldest go first past the limit.
724 this.sql.exec("DELETE FROM devices WHERE token NOT IN (SELECT token FROM devices ORDER BY created_at DESC LIMIT ?)", MAX_DEVICES);
725 return { ok: true };
726 }
727
728 async unregisterDevice(token: string): Promise<{ ok: boolean }> {
729 if (this.erased()) return { ok: true };
730 this.sql.exec("DELETE FROM devices WHERE token = ?", String(token ?? ""));
731 return { ok: true };
732 }
733
734 async status(endpoint: string | null): Promise<NotifyStatus> {
735 const subscribed = !!endpoint && this.sql.exec("SELECT 1 FROM subscriptions WHERE endpoint = ?", endpoint).toArray().length > 0;
736 return {
737 preferences: this.preferences(),
738 subscriptions: this.subscriptionCount(),
739 devices: this.deviceCount(),
740 subscribed,
741 vapid_public_key: this.env.VAPID_PUBLIC_KEY || null,
742 };
743 }
744
745 async setPreferences(change: unknown): Promise<NotifyPreferences> {
746 if (this.erased()) throw new Error("Account erased");
747 const preferences = mergePreferences(this.preferences(), change);
748 this.setMeta("preferences", JSON.stringify(preferences));
749 this.send({ type: "preferences", preferences });
750 return preferences;
751 }
752
753 /** Tabs open in `workspace` hear how these people now show (from its room). */
754 async presence(workspace: string, people: PresenceEntry[]): Promise<void> {
755 if (this.erased()) return;
756 const text = JSON.stringify({ type: "presence", workspace, people, full: false } satisfies FeedEvent);
757 for (const socket of this.ctx.getWebSockets()) {
758 const tab = socket.deserializeAttachment() as Tab | null;
759 if (tab?.workspace !== workspace) continue;
760 try {
761 socket.send(text);
762 } catch {
763 // Closing already.
764 }
765 }
766 }
767
768 /** The person's own: presence, status, Do Not Disturb. */
769 async own(person: { user_id: string; username: string }): Promise<OwnPresence> {
770 if (this.erased()) throw new Error("Account erased");
771 this.remember(person);
772 const kept = current(this.kept(), Date.now());
773 return ownOf(this.entry(), kept);
774 }
775
776 /** A change to the person's own, told to their tabs and to every workspace they are in. */
777 async setPresence(person: { user_id: string; username: string }, change: PresenceChange): Promise<OwnPresence> {
778 if (this.erased()) throw new Error("Account erased");
779 this.remember(person);
780 const kept = applyChange(this.kept(), change, Date.now());
781 this.setMeta("presence", JSON.stringify(kept));
782 const entry = await this.refresh();
783 return ownOf(entry, current(this.kept(), Date.now()));
784 }
785
786 private remember(person: { user_id: string; username: string }): void {
787 if (person.user_id) this.setMeta("user_id", person.user_id.slice(0, 100));
788 if (person.username) this.setMeta("username", person.username.slice(0, 100));
789 }
790
791 // ── Live repository pages ───────────────────────────────────────────────
792
793 private openSockets(): WebSocket[] {
794 return this.ctx.getWebSockets().filter((socket) => socket.readyState === WebSocket.OPEN);
795 }
796
797 /**
798 * The person's tabs hear what changes in repository `repoId` (`repo` is
799 * its `owner/name`) for REPO_WATCH_MS. The site checked they may read it.
800 */
801 async watchRepo(userId: string, repoId: string, repo: string): Promise<{ ok: boolean; until: string | null }> {
802 if (this.erased()) throw new Error("Account erased");
803 const topics = this.env.TOPICS;
804 if (!topics) return { ok: false, until: null };
805 const now = Date.now();
806 const until = watchUntil(now);
807 this.sql.exec("DELETE FROM watching WHERE until <= ?", now);
808 this.sql.exec("INSERT OR REPLACE INTO watching (repo_id, until) VALUES (?, ?)", repoId, until);
809 this.sql.exec("DELETE FROM watching WHERE repo_id NOT IN (SELECT repo_id FROM watching ORDER BY until DESC LIMIT ?)", MAX_WATCHING);
810 await topics.get(topics.idFromName(repoId)).watch(userId, repo, until);
811 return { ok: true, until: new Date(until).toISOString() };
812 }
813
814 /**
815 * The person's tabs hear what changes to the work items of `workspace`
816 * (its Work pages) for REPO_WATCH_MS, on the workspace's topic, kept
817 * with the repositories they watch. The site checked they are a member.
818 */
819 async watchWorkspace(userId: string, workspace: string): Promise<{ ok: boolean; until: string | null }> {
820 return this.watchRepo(userId, workspaceTopic(workspace), workspace);
821 }
822
823 /**
824 * What changed in a repository, from its topic: sent to the tabs, which
825 * each keep what is about the page they show. False when nobody here is
826 * watching it any more, so the topic drops this feed.
827 */
828 async repoChanged(repoId: string, changes: RepoChange[]): Promise<boolean> {
829 return this.passOn(repoId, { type: "repo", changes });
830 }
831
832 /** What changed to a workspace's work items, from its topic (`key`), as `repoChanged`. */
833 async workspaceChanged(key: string, changes: WorkspaceChange[]): Promise<boolean> {
834 return this.passOn(key, { type: "workspace", changes });
835 }
836
837 private passOn(key: string, event: FeedEvent): boolean {
838 const row = this.sql.exec<{ until: number }>("SELECT until FROM watching WHERE repo_id = ?", key).toArray()[0];
839 if (!stillWatching(row?.until, Date.now())) return false;
840 const sockets = this.openSockets();
841 if (!sockets.length) return false;
842 const text = JSON.stringify(event);
843 for (const socket of sockets) {
844 try {
845 socket.send(text);
846 } catch {
847 // Closing already.
848 }
849 }
850 return true;
851 }
852
853 private visit(workspace: string): Visit | null {
854 return this.sql.exec<Visit>("SELECT seen_at, previous_at FROM visits WHERE workspace = ?", workspace).toArray()[0] ?? null;
855 }
856
857 /** When the person was last on `workspace`'s Home page, as it counts from (src/visits.ts); null when never. */
858 async lastVisit(workspace: string): Promise<{ seen_at: string | null }> {
859 const since = lastVisit(this.visit(workspace), Date.now());
860 return { seen_at: since != null ? new Date(since).toISOString() : null };
861 }
862
863 /** Marks the person's visit to `workspace`'s Home page at `at`; it only moves forward. */
864 async markVisit(workspace: string, at: unknown): Promise<{ seen_at: string | null }> {
865 if (this.erased() || this.workspaceGone(workspace)) return { seen_at: null };
866 const now = Date.now();
867 const next = markVisit(this.visit(workspace), at, now);
868 if (next) {
869 this.sql.exec(
870 "INSERT INTO visits (workspace, seen_at, previous_at) VALUES (?, ?, ?) ON CONFLICT (workspace) DO UPDATE SET seen_at = excluded.seen_at, previous_at = excluded.previous_at",
871 workspace,
872 next.seen_at,
873 next.previous_at,
874 );
875 }
876 return this.lastVisit(workspace);
877 }
878
879 async test(username: string): Promise<{ ok: boolean; pushed: number }> {
880 if (this.erased()) throw new Error("Account erased");
881 const pushed = await this.tell(
882 {
883 id: `test:${crypto.randomUUID()}`,
884 kind: "dm",
885 workspace: "",
886 title: "g1t",
887 body: `Notifications are on, ${username || "there"}. This is what a message looks like.`,
888 href: "/settings/notifications",
889 actor: { kind: "system", id: "g1t", name: "g1t", avatar: null, avatar_seed: null },
890 channel_id: null,
891 thread_root: null,
892 created_at: new Date().toISOString(),
893 },
894 true,
895 );
896 return { ok: true, pushed };
897 }
898}