Skip to content
Sign up

flagon-io/g1t

public
00

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

gitagentscloudflarerustforgemcp
244 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 * `forget_conversation`, when someone clears their direct message with
15 * @g1t (src/forget.ts);
16 * - events: `notify`, for every inbox item, and `set_inbox`, the count after
17 * items arrive or are marked, naming the person by username;
18 * - the site: `subscribe`, `unsubscribe`, `status`, `set_preferences`,
19 * `test`, `presence` and `set_presence`, for the person signed in;
20 * `register_device` and `unregister_device`, for the phone app they
21 * signed in to (through `/-/notify`); and
22 * `last_visit` and `mark_visit`, when they were last on a workspace's
23 * Home page (src/visits.ts);
24 * - the site: `workspace_presence`, how a workspace's people
25 * show now (away, in Do Not Disturb, offline, and their statuses);
26 * - integrations, later: `set_presence` with a status of `source`
27 * `calendar` or `integration`, for the person they act for;
28 * - events and actions: `repo_changed`, what changed in repositories,
29 * told to whoever watches their pages; and the site `watch_repo`, for a
30 * tab on a repository page whose person it checked may read it;
31 * - events: `workspace_changed`, what changed to workspaces' work items,
32 * told to whoever watches their Work pages; and the site
33 * `watch_workspace`, for a tab on a Work page whose person it checked is
34 * a member.
35 *
36 * Presence: each feed works out its person's from their tabs and tells
37 * one room per workspace (src/room.ts), which tells everyone there who is
38 * online. See docs.g1t.sh/guides/chat/, "Presence and status".
39 *
40 * Live repository pages: one topic per repository (src/topic.ts) keeps
41 * who is watching it, and hands each change to their feeds. See
42 * docs.g1t.sh/guides/notifications/, "Pages that update live".
43 */
44import {
45 NOTIFY_VIEWER_HEADER,
46 identityClient,
47 type FeedDelivery,
48 type ServiceBinding,
49 type Viewer,
50} from "@g1t/contracts";
51
52import { FEED_USERNAME_HEADER, FEED_USER_ID_HEADER, Feed, type FeedEnv } from "./feed.ts";
53import { cleanChannelId } from "./forget.ts";
54import { onlyPeople } from "./presence.ts";
55import { visitWorkspace } from "./visits.ts";
56import { settledThreads } from "./prefs.ts";
57import { byRepo, byWorkspace, cleanRepoId, cleanRepoPath, cleanWorkspace, workspaceTopic } from "./watch.ts";
58
59export { Feed } from "./feed.ts";
60export { Room } from "./room.ts";
61export { Topic } from "./topic.ts";
62
63type Env = FeedEnv & {
64 IDENTITY: ServiceBinding;
65 FEEDS: DurableObjectNamespace<Feed>;
66};
67
68function feed(env: Env, userId: string) {
69 return env.FEEDS.get(env.FEEDS.idFromName(userId));
70}
71
72/** A user id from the arguments, or one looked up by username. */
73async function userIdOf(env: Env, given: { user_id?: unknown; username?: unknown; target?: unknown }): Promise<string | null> {
74 // Who it's for, at the top level or under `target`.
75 const target = given.target && typeof given.target === "object" ? (given.target as { user_id?: unknown; username?: unknown }) : null;
76 const args = { user_id: given.user_id ?? target?.user_id, username: given.username ?? target?.username };
77 if (typeof args.user_id === "string" && args.user_id) return args.user_id;
78 if (typeof args.username !== "string" || !args.username) return null;
79 const user = await identityClient(env.IDENTITY)
80 .userByUsername(args.username)
81 .catch(() => null);
82 return user?.id ?? null;
83}
84
85/** The most deliveries in one call; chat sends one per person in the conversation. */
86const MAX_DELIVERIES = 1000;
87
88async function answer(env: Env, method: string, args: any): Promise<Response> {
89 if (method === "purge_personal_data") {
90 if (typeof args?.id !== "string" || !args.id) return Response.json({ ok: false }, { status: 400 });
91 if (args.kind === "user") return Response.json(await feed(env, args.id).erase(args.id));
92 if (args.kind !== "workspace" || typeof args.slug !== "string" || !args.slug) return Response.json({ ok: false }, { status: 400 });
93 // Every slug it went by, and everyone identity knew as a member when it
94 // was purged, besides those its presence rooms know: their feeds keep
95 // notifications, counts and visits under the slug.
96 const strings = (value: unknown): string[] => (Array.isArray(value) ? value.filter((v): v is string => typeof v === "string" && v.length > 0) : []);
97 const slugs = [...new Set([args.slug, ...strings(args.old_slugs)].map((slug) => slug.toLowerCase()))];
98 const members = strings(args.members);
99 for (const slug of slugs) {
100 const room = env.ROOMS?.get(env.ROOMS.idFromName(slug));
101 const people = room ? await room.eraseMembers() : [];
102 const everyone = [...new Set([...people, ...members])];
103 for (let i = 0; i < everyone.length; i += 20) {
104 await Promise.all(everyone.slice(i, i + 20).map((userId) => feed(env, userId).eraseWorkspace(slug)));
105 }
106 // Only once their feeds are cleared, so a failure leaves them listed for the retry.
107 if (room) for (const userId of people) await room.dropPerson(userId);
108 }
109 return Response.json({ ok: true });
110 }
111 if (method === "workspace_presence") {
112 // For services: how a workspace's people show now, as its room keeps them.
113 const slug = typeof args?.workspace === "string" ? args.workspace.trim().toLowerCase() : "";
114 if (!slug || !env.ROOMS) return Response.json([]);
115 const everyone = await env.ROOMS.get(env.ROOMS.idFromName(slug)).people();
116 return Response.json(onlyPeople(everyone, Array.isArray(args.user_ids) ? args.user_ids : null));
117 }
118 if (method === "repo_changed") {
119 // For services: each repository's changes to its topic, which tells its watchers.
120 const topics = env.TOPICS;
121 if (!topics) return Response.json({ ok: true });
122 const grouped = byRepo(args?.changes);
123 const results = await Promise.allSettled([...grouped].map(([id, changes]) => topics.get(topics.idFromName(id)).changed(id, changes)));
124 for (const result of results) if (result.status === "rejected") console.error("notify: a repository's change was not told", result.reason);
125 return Response.json({ ok: true });
126 }
127 if (method === "workspace_changed") {
128 // For services: each workspace's work item changes to its topic, which tells whoever watches its Work pages.
129 const topics = env.TOPICS;
130 if (!topics) return Response.json({ ok: true });
131 const grouped = byWorkspace(args?.changes);
132 const results = await Promise.allSettled(
133 [...grouped].map(([slug, changes]) => {
134 const key = workspaceTopic(slug);
135 return topics.get(topics.idFromName(key)).workspaceChanged(key, changes);
136 }),
137 );
138 for (const result of results) if (result.status === "rejected") console.error("notify: a workspace's change was not told", result.reason);
139 return Response.json({ ok: true });
140 }
141 if (method === "deliver") {
142 const items = (Array.isArray(args?.items) ? args.items : []).slice(0, MAX_DELIVERIES) as FeedDelivery[];
143 const byUser = new Map<string, FeedDelivery[]>();
144 for (const item of items) {
145 if (typeof item?.user_id !== "string" || !item.user_id) continue;
146 byUser.set(item.user_id, [...(byUser.get(item.user_id) ?? []), item]);
147 }
148 const results = await Promise.allSettled([...byUser].map(([id, mine]) => feed(env, id).deliver(mine)));
149 for (const result of results) if (result.status === "rejected") console.error("notify: a delivery failed", result.reason);
150 return Response.json({ ok: true });
151 }
152 const userId = await userIdOf(env, args ?? {});
153 if (!userId) return Response.json({ ok: false });
154 const stub = feed(env, userId);
155 switch (method) {
156 case "notify":
157 return Response.json(await stub.notify(args.notification));
158 case "set_inbox":
159 return Response.json(await stub.setInbox(Number(args.unread), settledThreads(args.settled)));
160 case "subscribe":
161 return Response.json(await stub.subscribe(args.subscription, typeof args.user_agent === "string" ? args.user_agent : null));
162 case "unsubscribe":
163 return Response.json(await stub.unsubscribe(String(args.endpoint ?? "")));
164 case "register_device":
165 return Response.json(await stub.registerDevice(args.device));
166 case "unregister_device":
167 return Response.json(await stub.unregisterDevice(String(args.token ?? "")));
168 case "status":
169 return Response.json(await stub.status(typeof args.endpoint === "string" ? args.endpoint : null));
170 case "set_preferences":
171 return Response.json(await stub.setPreferences(args.preferences));
172 case "last_visit":
173 case "mark_visit": {
174 const workspace = visitWorkspace(args.workspace);
175 if (!workspace) return Response.json({ seen_at: null });
176 return Response.json(method === "last_visit" ? await stub.lastVisit(workspace) : await stub.markVisit(workspace, args.at));
177 }
178 case "watch_repo": {
179 const repoId = cleanRepoId(args.repo_id);
180 const repo = cleanRepoPath(args.repo);
181 if (!repoId || !repo) return Response.json({ ok: false, until: null });
182 return Response.json(await stub.watchRepo(userId, repoId, repo));
183 }
184 case "watch_workspace": {
185 // The site checked the person is a member: members see all of its work items.
186 const workspace = cleanWorkspace(args.workspace);
187 if (!workspace) return Response.json({ ok: false, until: null });
188 return Response.json(await stub.watchWorkspace(userId, workspace));
189 }
190 case "forget_conversation": {
191 // For chat, after the person cleared their direct message with @g1t (src/forget.ts).
192 const channelId = cleanChannelId(args.channel_id);
193 const workspace = visitWorkspace(args.workspace);
194 if (!channelId || !workspace) return Response.json({ ok: false });
195 return Response.json(await stub.forgetConversation(workspace, channelId));
196 }
197 case "test":
198 return Response.json(await stub.test(typeof args.username === "string" ? args.username : ""));
199 case "presence":
200 return Response.json(await stub.own({ user_id: userId, username: typeof args.username === "string" ? args.username : "" }));
201 case "set_presence":
202 return Response.json(
203 await stub.setPresence({ user_id: userId, username: typeof args.username === "string" ? args.username : "" }, args.change ?? {}),
204 );
205 default:
206 return new Response("Unknown method\n", { status: 404 });
207 }
208}
209
210/**
211 * `GET /live`, upgraded: the viewer comes in NOTIFY_VIEWER_HEADER, set by
212 * the site after checking the session; trusted only because this Worker
213 * is reachable through service bindings alone.
214 */
215async function live(request: Request, env: Env): Promise<Response> {
216 if (request.headers.get("upgrade")?.toLowerCase() !== "websocket") {
217 return new Response("Expected a WebSocket upgrade\n", { status: 426 });
218 }
219 let viewer: Viewer = null;
220 try {
221 viewer = JSON.parse(request.headers.get(NOTIFY_VIEWER_HEADER) ?? "null") as Viewer;
222 } catch {
223 viewer = null;
224 }
225 if (!viewer?.id) return new Response("Sign in first\n", { status: 401 });
226 const headers = new Headers(request.headers);
227 headers.delete(NOTIFY_VIEWER_HEADER);
228 // Who the feed is for, so it can name them to the presence rooms.
229 headers.set(FEED_USER_ID_HEADER, viewer.id);
230 headers.set(FEED_USERNAME_HEADER, viewer.username);
231 return feed(env, viewer.id).fetch(new Request(request.url, { method: "GET", headers }));
232}
233
234export default {
235 async fetch(request: Request, env: Env): Promise<Response> {
236 const url = new URL(request.url);
237 if (request.method === "GET" && url.pathname === "/live") return live(request, env);
238 const match = url.pathname.match(/^\/rpc\/([a-z_]+)$/);
239 if (request.method !== "POST" || !match) return new Response("Not found\n", { status: 404 });
240 const args = (await request.json().catch(() => ({}))) as any;
241 return answer(env, match[1], args);
242 },
243} satisfies ExportedHandler<Env>;
244