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.

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