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