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
339 linesCodeBlameRaw
1/**
2 * Pages that cite code a change touched become possibly out of date
3 * (docs.g1t.sh/guides/artifacts/, "Docs that cite code"). The events
4 * service sends this service `git.push` and `pull.merged`
5 * (`SUBSCRIBER_DOCS`, crates/contracts subscribers.rs), and the lifecycle
6 * events every subscriber hears.
7 *
8 * - **What changed** is asked as g1t itself (a system viewer in the
9 * repository's workspace): a merged pull request's files from the work
10 * service, or a push's from comparing it with where the branch was.
11 * Only pushes to the default branch count.
12 * - **Cheap when nothing cites the repository:** the citations table is
13 * asked first, and nothing else is.
14 * - **Once per page and commit:** a merge is told twice (the push and the
15 * pull request); both land on one row, which the pull request names.
16 * Owners hear of it once, and `doc.page.stale` is published once.
17 * - **Privacy:** what is stored is the change itself; what a reader sees
18 * of it (the page, its banner, the inbox) is filtered by whether they can
19 * read the repository.
20 */
21import {
22 currentMovedPath,
23 identityClient,
24 notifyClient,
25 repoMove,
26 reposClient,
27 staleMovedPaths,
28 workClient,
29 type G1tEvent,
30 type ServiceBinding,
31 type User,
32} from "@g1t/contracts";
33
34import { touchedPaths } from "./citations.ts";
35import { foliosCite, recordFolioChanges } from "./folios/staleness.ts";
36import type { FolioRoom } from "./folios/room.ts";
37import type { PageRoom } from "./room.ts";
38import { publishDocEvent } from "./events.ts";
39import { pageSlug } from "./slugs.ts";
40
41export type StaleEnv = {
42 DB: D1Database;
43 IDENTITY: ServiceBinding;
44 REPOS?: ServiceBinding;
45 WORK?: ServiceBinding;
46 NOTIFY?: ServiceBinding;
47 EVENTS?: ServiceBinding;
48 PAGES?: DurableObjectNamespace<PageRoom>;
49 /** Folios' rooms (Artifacts mode): folios cite code too (src/folios/staleness.ts). */
50 FOLIOS?: DurableObjectNamespace<FolioRoom>;
51};
52
53/** What a change touched, as this module records it. */
54export type Change = {
55 repo: string;
56 repo_id: string;
57 commit: string;
58 pull: { number: number; title: string | null } | null;
59 /** Every path the change touched. */
60 changed: string[];
61 /** Who made it (an account id), for the events it causes. */
62 actor: string | null;
63};
64
65/** g1t itself, reading in a repository's workspace: how staleness asks what changed. */
66export function systemReader(namespace: string): User {
67 return { id: "g1t", username: "g1t", kind: "system", verified: true, workspaces: [{ slug: namespace.toLowerCase(), role: "owner" }] };
68}
69
70const ZERO = /^0+$/;
71
72async function pathById(repos: ServiceBinding, id: string): Promise<{ namespace: string; name: string } | null> {
73 const response = await repos.fetch("https://service/rpc/path_by_id", {
74 method: "POST",
75 headers: { "content-type": "application/json" },
76 body: JSON.stringify({ id }),
77 });
78 if (!response.ok) return null;
79 return (await response.json()) as { namespace: string; name: string } | null;
80}
81
82/** Whether any live page cites the repository, or any workspace shows its docs. */
83async function interested(db: D1Database, repo: string, repoId: string): Promise<{ cited: boolean; spaces: boolean }> {
84 const [cited, spaces] = await db.batch<{ yes: number }>([
85 db.prepare("SELECT 1 AS yes FROM citations c JOIN pages p ON p.id = c.page_id WHERE c.repo = ? AND p.archived_at IS NULL LIMIT 1").bind(repo),
86 db.prepare("SELECT 1 AS yes FROM repo_spaces WHERE repo_id = ? LIMIT 1").bind(repoId),
87 ]);
88 const folios = cited?.results.length ? false : await foliosCite(db, repo).catch(() => false);
89 return { cited: !!cited?.results.length || folios, spaces: !!spaces?.results.length };
90}
91
92/** What a push to the default branch changed: the files between where it was and where it is. */
93async function pushChange(
94 env: StaleEnv,
95 path: { namespace: string; name: string },
96 repoId: string,
97 before: string | undefined,
98 after: string,
99): Promise<string[] | null> {
100 if (!env.REPOS || !before || ZERO.test(before)) return null;
101 const compared = await reposClient(env.REPOS).compare(repoId, systemReader(path.namespace), before, after);
102 if (!compared.ok) return null;
103 return compared.value.files.map((f) => f.path);
104}
105
106/** What a merged pull request changed, and its title; null when it didn't merge into the default branch. */
107async function pullChange(
108 env: StaleEnv,
109 path: { namespace: string; name: string },
110 repoId: string,
111 number: number,
112): Promise<{ title: string; changed: string[] } | null> {
113 if (!env.WORK || !env.REPOS) return null;
114 const reader = systemReader(path.namespace);
115 const [found, repo] = await Promise.all([workClient(env.WORK).getPull(path, number, reader), reposClient(env.REPOS).getById(repoId, reader)]);
116 if (!found.ok || !repo.ok) return null;
117 const pull = found.value.pull;
118 if (pull.base && pull.base !== repo.value.defaultBranch) return null;
119 let changed = (pull.files ?? []).map((f) => f.path);
120 if (!changed.length && pull.mergeBase && pull.headCommit) {
121 const compared = await reposClient(env.REPOS).compare(repoId, reader, pull.mergeBase, pull.headCommit);
122 if (compared.ok) changed = compared.value.files.map((f) => f.path);
123 }
124 return { title: pull.title, changed };
125}
126
127type CitedRow = { page_id: string; path: string };
128type PageRow = { id: string; workspace_id: string; space_id: string; title: string; space_slug: string };
129
130/**
131 * Records a change against every page whose citations it touches.
132 * Returns the pages newly made stale by it.
133 */
134export async function record(env: StaleEnv, change: Change, now = new Date()): Promise<string[]> {
135 const db = env.DB;
136 const cited = (
137 await db
138 .prepare("SELECT c.page_id, c.path FROM citations c JOIN pages p ON p.id = c.page_id WHERE c.repo = ? AND p.archived_at IS NULL")
139 .bind(change.repo)
140 .all<CitedRow>()
141 ).results;
142 const byPage = new Map<string, CitedRow[]>();
143 for (const row of cited) byPage.set(row.page_id, [...(byPage.get(row.page_id) ?? []), row]);
144 const fresh: string[] = [];
145 const hits = new Map<string, string[]>();
146 for (const [pageId, rows] of byPage) {
147 const touched = touchedPaths(rows, change.changed);
148 if (!touched.length) continue;
149 hits.set(pageId, touched);
150 const at = now.toISOString();
151 const [inserted] = await db.batch<{ page_id: string }>([
152 db
153 .prepare(
154 "INSERT INTO page_changes (page_id, repo, repo_id, commit_sha, pull_number, pull_title, paths, detected_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (page_id, repo, commit_sha) DO NOTHING RETURNING page_id",
155 )
156 .bind(pageId, change.repo, change.repo_id, change.commit, change.pull?.number ?? null, change.pull?.title ?? null, JSON.stringify(touched), at),
157 ...(change.pull
158 ? [
159 db
160 .prepare("UPDATE page_changes SET pull_number = ?, pull_title = ? WHERE page_id = ? AND repo = ? AND commit_sha = ?")
161 .bind(change.pull.number, change.pull.title, pageId, change.repo, change.commit),
162 ]
163 : []),
164 ]);
165 if (inserted?.results.length) fresh.push(pageId);
166 }
167 if (fresh.length) await tellOf(env, change, fresh, hits);
168 return fresh;
169}
170
171/** Owners hear of pages newly stale, rooms ask their readers to look again, and `doc.page.stale` goes out. */
172async function tellOf(env: StaleEnv, change: Change, pageIds: string[], hits: Map<string, string[]>): Promise<void> {
173 const db = env.DB;
174 const marks = pageIds.map(() => "?").join(",");
175 const [pages, owners] = await Promise.all([
176 db
177 .prepare(
178 `SELECT p.id, p.workspace_id, p.space_id, p.title, s.slug AS space_slug FROM pages p JOIN spaces s ON s.id = p.space_id WHERE p.id IN (${marks})`,
179 )
180 .bind(...pageIds)
181 .all<PageRow>(),
182 db
183 .prepare(`SELECT page_id, principal FROM page_owners WHERE page_id IN (${marks})`)
184 .bind(...pageIds)
185 .all<{ page_id: string; principal: string }>(),
186 ]);
187 const workspaceIds = [...new Set(pages.results.map((p) => p.workspace_id))];
188 const slugs = await identityClient(env.IDENTITY)
189 .usernames(workspaceIds)
190 .catch(() => ({}) as Record<string, string>);
191 // Which owners can read the repository: they are told which change it was.
192 const ownerIds = [...new Set(owners.results.filter((o) => o.principal.startsWith("user:")).map((o) => o.principal.slice(5)))];
193 const people = ownerIds.length
194 ? await identityClient(env.IDENTITY)
195 .usersForAudience(ownerIds)
196 .catch(() => [] as User[])
197 : [];
198 const [namespace, name] = change.repo.split("/") as [string, string];
199 const canRead = new Set<string>();
200 if (env.REPOS) {
201 await Promise.all(
202 people.map(async (person) => {
203 const found = await reposClient(env.REPOS!)
204 .get({ namespace, name }, person)
205 .catch(() => null);
206 if (found?.ok) canRead.add(person.id);
207 }),
208 );
209 }
210 const what = change.pull ? `${change.repo}#${change.pull.number}` : `${change.repo}@${change.commit.slice(0, 7)}`;
211 const notify = env.NOTIFY ? notifyClient(env.NOTIFY) : null;
212 for (const page of pages.results) {
213 const slug = slugs[page.workspace_id];
214 if (!slug) continue;
215 const href = `/${slug}/-/docs/${page.space_slug}/${pageSlug(page.title, page.id)}`;
216 const paths = hits.get(page.id) ?? [];
217 const keys = owners.results.filter((o) => o.page_id === page.id).map((o) => o.principal);
218 const work: Promise<unknown>[] = [];
219 if (notify) {
220 for (const key of keys.filter((k) => k.startsWith("user:"))) {
221 const id = key.slice(5);
222 const known = canRead.has(id);
223 work.push(
224 notify
225 .notify(
226 { user_id: id },
227 {
228 id: `doc-stale:${page.id}:${change.commit}:${id}`,
229 kind: "inbox",
230 workspace: slug,
231 title: `${page.title || "Untitled"} may be out of date`,
232 body: known
233 ? `${what} changed ${paths.slice(0, 3).join(", ")}${paths.length > 3 ? ` and ${paths.length - 3} more` : ""}`
234 : "A change to code this page cites was merged.",
235 href,
236 actor: { kind: "system", id: "g1t", name: "g1t", avatar: null, avatar_seed: null },
237 created_at: new Date().toISOString(),
238 },
239 )
240 .catch(() => undefined),
241 );
242 }
243 }
244 if (env.PAGES) {
245 work.push(
246 env.PAGES.get(env.PAGES.idFromName(page.id))
247 .notice({ type: "page.staleness" })
248 .catch(() => undefined),
249 );
250 }
251 work.push(
252 publishDocEvent(
253 env.EVENTS,
254 "doc.page.stale",
255 {
256 workspace: slug,
257 workspaceId: page.workspace_id,
258 pageId: page.id,
259 spaceId: page.space_id,
260 title: page.title,
261 path: href,
262 repoId: change.repo_id,
263 repo: change.repo,
264 commit: change.commit,
265 pull: change.pull?.number ?? null,
266 paths,
267 owners: keys,
268 },
269 change.actor ? `user:${change.actor}` : null,
270 ),
271 );
272 await Promise.all(work);
273 }
274}
275
276/** A repository moved: rows kept under its old path follow it. */
277async function followMove(env: StaleEnv, event: G1tEvent): Promise<void> {
278 const move = repoMove(event);
279 if (!move || !env.REPOS) return;
280 const current = (await currentMovedPath(env.REPOS, move)).toLowerCase();
281 const stale = staleMovedPaths(move, current).map((p) => p.toLowerCase());
282 if (!stale.length) return;
283 const db = env.DB;
284 const statements: D1PreparedStatement[] = [];
285 for (const old of stale) {
286 statements.push(
287 db.prepare("UPDATE OR IGNORE citations SET repo = ? WHERE repo = ?").bind(current, old),
288 db.prepare("UPDATE OR IGNORE page_changes SET repo = ? WHERE repo = ?").bind(current, old),
289 db.prepare("UPDATE OR IGNORE page_projects SET repo = ? WHERE repo = ?").bind(current, old),
290 db.prepare("UPDATE OR IGNORE folio_citations SET repo = ? WHERE repo = ?").bind(current, old),
291 db.prepare("UPDATE OR IGNORE folio_changes SET repo = ? WHERE repo = ?").bind(current, old),
292 db.prepare("UPDATE OR IGNORE folio_projects SET repo = ? WHERE repo = ?").bind(current, old),
293 db.prepare("UPDATE OR IGNORE space_projects SET repo = ? WHERE repo = ?").bind(current, old),
294 db.prepare("UPDATE repo_spaces SET repo = ? WHERE repo_id = ?").bind(current, move.repoId),
295 );
296 }
297 await db.batch(statements);
298}
299
300/** One event, as this service acts on it. `reindex` reads a project's docs again (src/repo-spaces.ts). */
301export async function onEvent(env: StaleEnv, event: G1tEvent, reindex: (repoId: string, commit: string) => Promise<void>): Promise<void> {
302 if (event.type === "repo.renamed" || event.type === "repo.transferred") return followMove(env, event);
303 if (event.type === "repo.purged") {
304 await env.DB.prepare("DELETE FROM repo_spaces WHERE repo_id = ?").bind(event.data.repoId).run();
305 return;
306 }
307 if (event.type !== "git.push" && event.type !== "pull.merged") return;
308 if (!env.REPOS) return;
309 const repoId = event.repoId ?? (event.data as { repoId?: string }).repoId ?? null;
310 if (!repoId) return;
311 if (event.type === "git.push" && !event.data.defaultBranch) return;
312 const path = await pathById(env.REPOS, repoId);
313 if (!path) return;
314 const repo = `${path.namespace}/${path.name}`.toLowerCase();
315 const wants = await interested(env.DB, repo, repoId);
316 if (event.type === "git.push") {
317 if (wants.spaces) await reindex(repoId, event.data.after);
318 if (!wants.cited) return;
319 const changed = await pushChange(env, path, repoId, event.data.before, event.data.after);
320 if (!changed?.length) return;
321 const change: Change = { repo, repo_id: repoId, commit: event.data.after, pull: null, changed, actor: event.actor };
322 await record(env, change);
323 await recordFolioChanges(env, change);
324 return;
325 }
326 if (!wants.cited) return;
327 const pulled = await pullChange(env, path, repoId, event.data.number);
328 if (!pulled?.changed.length) return;
329 const change: Change = {
330 repo,
331 repo_id: repoId,
332 commit: event.data.commit,
333 pull: { number: event.data.number, title: pulled.title },
334 changed: pulled.changed,
335 actor: event.actor,
336 };
337 await record(env, change);
338 await recordFolioChanges(env, change);
339}