Skip to content
Sign up

flagon-io/g1t

public
0
163 linesCodeBlameRaw
1/**
2 * The notify service: live notifications, unread counts, and push to
3 * browsers (Web Push) and to the phone app (through Expo, src/expopush.ts).
4 * One feed per person (src/feed.ts).
5 *
6 * Reached through service bindings only: `POST /rpc/<method>` with
7 * snake_case bodies (`notifyClient` in @g1t/contracts), and `GET /live`,
8 * the person's feed socket, which the site forwards after checking the
9 * session (NOTIFY_VIEWER_HEADER), with the counts it read (NOTIFY_SEED_HEADER).
10 *
11 * Who sends what:
12 * - chat: `deliver`, for every message (counts for each person in the
13 * conversation, a notification for those it is for) and every read;
14 * - events: `notify`, for every inbox item, and `set_inbox`, the count after
15 * items arrive or are marked, naming the person by username;
16 * - the site: `subscribe`, `unsubscribe`, `status`, `set_preferences`,
17 * `test`, `presence` and `set_presence`, for the person signed in;
18 * `register_device` and `unregister_device`, for the phone app they
19 * signed in to (through `/-/notify`); and
20 * `last_visit` and `mark_visit`, when they were last on a workspace's
21 * Home page (src/visits.ts);
22 * - agents and the site: `workspace_presence`, how a workspace's people
23 * show now (away, in Do Not Disturb, offline, and their statuses);
24 * - integrations, later: `set_presence` with a status of `source`
25 * `calendar` or `integration`, for the person they act for.
26 *
27 * Presence: each feed works out its person's from their tabs and tells
28 * one room per workspace (src/room.ts), which tells everyone there who is
29 * online. See docs.g1t.sh/guides/chat/, "Presence and status".
30 */
31import {
32 NOTIFY_VIEWER_HEADER,
33 identityClient,
34 type FeedDelivery,
35 type ServiceBinding,
36 type Viewer,
37} from "@g1t/contracts";
38
39import { FEED_USERNAME_HEADER, FEED_USER_ID_HEADER, Feed, type FeedEnv } from "./feed.ts";
40import { onlyPeople } from "./presence.ts";
41import { visitWorkspace } from "./visits.ts";
42
43export { Feed } from "./feed.ts";
44export { Room } from "./room.ts";
45
46type Env = FeedEnv & {
47 IDENTITY: ServiceBinding;
48 FEEDS: DurableObjectNamespace<Feed>;
49};
50
51function feed(env: Env, userId: string) {
52 return env.FEEDS.get(env.FEEDS.idFromName(userId));
53}
54
55/** A user id from the arguments, or one looked up by username. */
56async function userIdOf(env: Env, given: { user_id?: unknown; username?: unknown; target?: unknown }): Promise<string | null> {
57 // Who it's for, at the top level or under `target` as the agents service sends it.
58 const target = given.target && typeof given.target === "object" ? (given.target as { user_id?: unknown; username?: unknown }) : null;
59 const args = { user_id: given.user_id ?? target?.user_id, username: given.username ?? target?.username };
60 if (typeof args.user_id === "string" && args.user_id) return args.user_id;
61 if (typeof args.username !== "string" || !args.username) return null;
62 const user = await identityClient(env.IDENTITY)
63 .userByUsername(args.username)
64 .catch(() => null);
65 return user?.id ?? null;
66}
67
68/** The most deliveries in one call; chat sends one per person in the conversation. */
69const MAX_DELIVERIES = 1000;
70
71async function answer(env: Env, method: string, args: any): Promise<Response> {
72 if (method === "workspace_presence") {
73 // For services: how a workspace's people show now, as its room keeps them.
74 const slug = typeof args?.workspace === "string" ? args.workspace.trim().toLowerCase() : "";
75 if (!slug || !env.ROOMS) return Response.json([]);
76 const everyone = await env.ROOMS.get(env.ROOMS.idFromName(slug)).people();
77 return Response.json(onlyPeople(everyone, Array.isArray(args.user_ids) ? args.user_ids : null));
78 }
79 if (method === "deliver") {
80 const items = (Array.isArray(args?.items) ? args.items : []).slice(0, MAX_DELIVERIES) as FeedDelivery[];
81 const byUser = new Map<string, FeedDelivery[]>();
82 for (const item of items) {
83 if (typeof item?.user_id !== "string" || !item.user_id) continue;
84 byUser.set(item.user_id, [...(byUser.get(item.user_id) ?? []), item]);
85 }
86 const results = await Promise.allSettled([...byUser].map(([id, mine]) => feed(env, id).deliver(mine)));
87 for (const result of results) if (result.status === "rejected") console.error("notify: a delivery failed", result.reason);
88 return Response.json({ ok: true });
89 }
90 const userId = await userIdOf(env, args ?? {});
91 if (!userId) return Response.json({ ok: false });
92 const stub = feed(env, userId);
93 switch (method) {
94 case "notify":
95 return Response.json(await stub.notify(args.notification));
96 case "set_inbox":
97 return Response.json(await stub.setInbox(Number(args.unread)));
98 case "subscribe":
99 return Response.json(await stub.subscribe(args.subscription, typeof args.user_agent === "string" ? args.user_agent : null));
100 case "unsubscribe":
101 return Response.json(await stub.unsubscribe(String(args.endpoint ?? "")));
102 case "register_device":
103 return Response.json(await stub.registerDevice(args.device));
104 case "unregister_device":
105 return Response.json(await stub.unregisterDevice(String(args.token ?? "")));
106 case "status":
107 return Response.json(await stub.status(typeof args.endpoint === "string" ? args.endpoint : null));
108 case "set_preferences":
109 return Response.json(await stub.setPreferences(args.preferences));
110 case "last_visit":
111 case "mark_visit": {
112 const workspace = visitWorkspace(args.workspace);
113 if (!workspace) return Response.json({ seen_at: null });
114 return Response.json(method === "last_visit" ? await stub.lastVisit(workspace) : await stub.markVisit(workspace, args.at));
115 }
116 case "test":
117 return Response.json(await stub.test(typeof args.username === "string" ? args.username : ""));
118 case "presence":
119 return Response.json(await stub.own({ user_id: userId, username: typeof args.username === "string" ? args.username : "" }));
120 case "set_presence":
121 return Response.json(
122 await stub.setPresence({ user_id: userId, username: typeof args.username === "string" ? args.username : "" }, args.change ?? {}),
123 );
124 default:
125 return new Response("Unknown method\n", { status: 404 });
126 }
127}
128
129/**
130 * `GET /live`, upgraded: the viewer comes in NOTIFY_VIEWER_HEADER, set by
131 * the site after checking the session; trusted only because this Worker
132 * is reachable through service bindings alone.
133 */
134async function live(request: Request, env: Env): Promise<Response> {
135 if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") {
136 return new Response("Expected a WebSocket upgrade\n", { status: 426 });
137 }
138 let viewer: Viewer = null;
139 try {
140 viewer = JSON.parse(request.headers.get(NOTIFY_VIEWER_HEADER) ?? "null") as Viewer;
141 } catch {
142 viewer = null;
143 }
144 if (!viewer?.id) return new Response("Sign in first\n", { status: 401 });
145 const headers = new Headers(request.headers);
146 headers.delete(NOTIFY_VIEWER_HEADER);
147 // Who the feed is for, so it can name them to the presence rooms.
148 headers.set(FEED_USER_ID_HEADER, viewer.id);
149 headers.set(FEED_USERNAME_HEADER, viewer.username);
150 return feed(env, viewer.id).fetch(new Request(request.url, { method: "GET", headers }));
151}
152
153export default {
154 async fetch(request: Request, env: Env): Promise<Response> {
155 const url = new URL(request.url);
156 if (request.method === "GET" && url.pathname === "/live") return live(request, env);
157 const match = url.pathname.match(/^\/rpc\/([a-z_]+)$/);
158 if (request.method !== "POST" || !match) return new Response("Not found\n", { status: 404 });
159 const args = (await request.json().catch(() => ({}))) as any;
160 return answer(env, match[1], args);
161 },
162} satisfies ExportedHandler<Env>;
163