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
139 linesCodeBlameRaw
1import { GHOST_ID } from "@g1t/contracts";
2
3import { cloudflareVectors } from "./vectors.ts";
4import { fileStore, type FileStoreEnv } from "./files.ts";
5import type { IndexEnv } from "./indexer.ts";
6import type { PageRoom } from "./room.ts";
7import type { FolioRoom } from "./folios/room.ts";
8
9type Env = IndexEnv & FileStoreEnv & { PAGES?: DurableObjectNamespace<PageRoom>; FOLIOS?: DurableObjectNamespace<FolioRoom> };
10type Subject = { kind: "user" | "workspace"; id: string; username?: string };
11const GHOST = `user:${GHOST_ID}`;
12/** Rows (chunks or files) a workspace sweep takes per pass; each pass deletes what it read, so a retry resumes. */
13const PAGE = 200;
14/** Documents destroyed per call; the rest wait for identity's next retry. */
15const DOCUMENTS = 25;
16/** Sweep passes per call, so a large workspace stays within one invocation's subrequests. */
17const PASSES = 10;
18
19/**
20 * Called only by identity's durable purge jobs. Failures leave the SQL
21 * references in place, every removal is idempotent, and work left over
22 * answers `pending`, so identity retries rather than recording success.
23 */
24export async function purgePersonalData(env: Env, args: Subject): Promise<{ ok: boolean; pending?: boolean }> {
25 if (!args?.id || !["user", "workspace"].includes(args.kind)) throw new Error("Invalid purge subject");
26 const db = env.DB, key = `user:${args.id}`;
27 await db.prepare("INSERT OR IGNORE INTO erased_subjects VALUES (?, ?)").bind(args.id, args.kind).run();
28 if (args.kind === "workspace") {
29 await db.batch([
30 db.prepare("INSERT OR IGNORE INTO purge_documents SELECT ?, 'page', id FROM pages WHERE workspace_id = ?").bind(args.id, args.id),
31 db.prepare("INSERT OR IGNORE INTO purge_documents SELECT ?, 'folio', id FROM folios WHERE workspace_id = ?").bind(args.id, args.id),
32 ]);
33 } else {
34 await db.batch([
35 db.prepare("INSERT OR IGNORE INTO purge_access (subject_id, kind, document_id) SELECT ?1, 'page', id FROM pages WHERE created_by = ?2 OR updated_by = ?2 OR id IN (SELECT page_id FROM page_views WHERE user_id = ?1)").bind(args.id, key),
36 db.prepare("INSERT OR IGNORE INTO purge_access (subject_id, kind, document_id) SELECT ?1, 'folio', id FROM folios WHERE owner = ?2 OR created_by = ?2 OR updated_by = ?2 OR id IN (SELECT folio_id FROM folio_visits WHERE user_id = ?1)").bind(args.id, key),
37 ]);
38 const access = (await db.prepare("SELECT kind, document_id FROM purge_access WHERE subject_id = ? AND done = 0 LIMIT 50").bind(args.id).all<{ kind: string; document_id: string }>()).results;
39 for (const doc of access) {
40 const rooms = doc.kind === "folio" ? env.FOLIOS : env.PAGES;
41 if (!rooms) throw new Error("Document access cleanup unavailable");
42 await rooms.get(rooms.idFromName(doc.document_id)).forgetPerson(key, args.username ?? "");
43 await db.prepare("UPDATE purge_access SET done = 1 WHERE subject_id = ? AND kind = ? AND document_id = ?").bind(args.id, doc.kind, doc.document_id).run();
44 }
45 if (await db.prepare("SELECT 1 FROM purge_access WHERE subject_id = ? AND done = 0 LIMIT 1").bind(args.id).first()) return { ok: false, pending: true };
46 // Only private folios nobody else can open, in any workspace: the folio
47 // and everything under it the person's own, with no other principal's
48 // access or grant. Shared material needs a case-specific assessment.
49 await db.prepare(`INSERT OR IGNORE INTO purge_documents SELECT ?1, 'folio', f.id FROM folios f
50 WHERE f.owner = ?2 AND f.space_id IS NULL AND f.general_access = 'none'
51 AND NOT EXISTS (SELECT 1 FROM folios child WHERE substr(child.path, 1, length(f.path)) = f.path AND (
52 child.owner != ?2 OR child.general_access != 'none'
53 OR EXISTS (SELECT 1 FROM folio_access a WHERE a.folio_id = child.id AND a.principal != ?2)
54 OR EXISTS (SELECT 1 FROM folio_grants g WHERE g.folio_id = child.id AND g.principal != ?2)))`).bind(args.id, key).run();
55 }
56 const docs = (await db.prepare("SELECT kind, document_id FROM purge_documents WHERE subject_id = ? LIMIT ?").bind(args.id, DOCUMENTS).all<{ kind: string; document_id: string }>()).results;
57 for (const doc of docs) {
58 const folio = doc.kind === "folio", id = doc.document_id;
59 // First, so an upload, an index run or a save racing this cannot write rows for it again (migration 0010, indexer.ts).
60 await db.prepare("INSERT OR IGNORE INTO erased_documents (id, kind) VALUES (?, ?)").bind(id, doc.kind).run();
61 const rooms = folio ? env.FOLIOS : env.PAGES;
62 if (!rooms) throw new Error("Document room cleanup unavailable");
63 await rooms.get(rooms.idFromName(id)).destroy();
64 const chunks = (await db.prepare(`SELECT id FROM ${folio ? "folio_chunks" : "doc_chunks"} WHERE ${folio ? "folio_id" : "page_id"} = ?`).bind(id).all<{ id: string }>()).results.map(r => r.id);
65 const vectors = folio ? env.FOLIO_VECTORS : env.VECTORS;
66 if (vectors && chunks.length) await cloudflareVectors(vectors).delete(chunks);
67 const files = (await db.prepare(`SELECT key FROM ${folio ? "folio_files" : "files"} WHERE ${folio ? "folio_id" : "page_id"} = ?`).bind(id).all<{ key: string }>()).results;
68 for (const { key: file } of files) await fileStore(env).delete(`docs/${file}`);
69 if (folio) {
70 const states = (await db.prepare("SELECT state_key AS key FROM folio_versions WHERE folio_id = ?1 AND state_key IS NOT NULL UNION SELECT update_key AS key FROM folio_proposals WHERE folio_id = ?1 AND update_key IS NOT NULL").bind(id).all<{ key: string }>()).results;
71 for (const { key: file } of states) await fileStore(env).delete(file);
72 }
73 const table = folio ? "folios" : "pages", column = folio ? "folio_id" : "page_id";
74 await db.batch([
75 // By document, so passages a late index run wrote without a chunk row go too.
76 db.prepare(`DELETE FROM ${folio ? "folio_chunks_fts WHERE folio_id" : "doc_chunks_fts WHERE doc_id"} = ?`).bind(id),
77 db.prepare(`DELETE FROM ${folio ? "folio_chunks" : "doc_chunks"} WHERE ${column} = ?`).bind(id),
78 db.prepare(`DELETE FROM ${folio ? "folios_fts" : "pages_fts"} WHERE ${column} = ?`).bind(id),
79 db.prepare(`DELETE FROM ${folio ? "folio_files" : "files"} WHERE ${column} = ?`).bind(id),
80 db.prepare(`UPDATE ${table} SET parent_id = NULL WHERE parent_id = ?`).bind(id),
81 db.prepare(`DELETE FROM ${table} WHERE id = ?`).bind(id),
82 db.prepare("DELETE FROM purge_documents WHERE subject_id = ? AND kind = ? AND document_id = ?").bind(args.id, doc.kind, id),
83 ]);
84 }
85 if (await db.prepare("SELECT 1 FROM purge_documents WHERE subject_id = ? LIMIT 1").bind(args.id).first()) return { ok: false, pending: true };
86 if (args.kind === "workspace") {
87 // What no document holds any more: repository docs' passages, passages
88 // and uploads of documents deleted earlier. A page at a time, each
89 // deleted once its vectors or blobs are, until none is left.
90 let passes = PASSES;
91 for (const [table, store] of [["doc_chunks", env.VECTORS], ["folio_chunks", env.FOLIO_VECTORS]] as const) {
92 const fts = table === "doc_chunks" ? "doc_chunks_fts" : "folio_chunks_fts";
93 for (;;) {
94 if (passes-- <= 0) return { ok: false, pending: true };
95 const ids = (await db.prepare(`SELECT id FROM ${table} WHERE workspace_id = ? LIMIT ${PAGE}`).bind(args.id).all<{ id: string }>()).results.map(r => r.id);
96 if (!ids.length) break;
97 if (store) await cloudflareVectors(store).delete(ids);
98 await db.batch([
99 db.prepare(`DELETE FROM ${fts} WHERE chunk_id IN (SELECT value FROM json_each(?))`).bind(JSON.stringify(ids)),
100 db.prepare(`DELETE FROM ${table} WHERE id IN (SELECT value FROM json_each(?))`).bind(JSON.stringify(ids)),
101 ]);
102 }
103 }
104 for (const table of ["files", "folio_files"]) {
105 for (;;) {
106 if (passes-- <= 0) return { ok: false, pending: true };
107 const rows = (await db.prepare(`SELECT key FROM ${table} WHERE workspace_id = ? LIMIT ${PAGE}`).bind(args.id).all<{ key: string }>()).results;
108 if (!rows.length) break;
109 for (const row of rows) await fileStore(env).delete(`docs/${row.key}`);
110 await db.prepare(`DELETE FROM ${table} WHERE key IN (SELECT value FROM json_each(?))`).bind(JSON.stringify(rows.map(r => r.key))).run();
111 }
112 }
113 await db.batch([
114 db.prepare("DELETE FROM repo_files_fts WHERE space_id IN (SELECT id FROM repo_spaces WHERE workspace_id = ?)").bind(args.id),
115 ...["repo_spaces", "templates", "folio_templates", "spaces", "doc_index_runs", "doc_embed_usage", "workspace_billing"].map(table => db.prepare(`DELETE FROM ${table} WHERE workspace_id = ?`).bind(args.id)),
116 ]);
117 } else {
118 const statements: D1PreparedStatement[] = [];
119 for (const table of ["page_views", "favorites", "folio_visits", "folio_favorites", "folio_access_requests", "space_joins"]) statements.push(db.prepare(`DELETE FROM ${table} WHERE user_id = ?`).bind(args.id));
120 for (const table of ["space_members", "page_owners", "folio_grants", "folio_access"]) statements.push(db.prepare(`DELETE FROM ${table} WHERE principal = ?`).bind(key));
121 const authors: Record<string, string[]> = {
122 spaces: ["created_by", "deleted_by"], pages: ["created_by", "updated_by", "archived_by"],
123 folios: ["owner", "created_by", "updated_by", "edited_by", "trashed_by"],
124 templates: ["created_by"], folio_templates: ["created_by"], files: ["created_by"], folio_files: ["created_by"],
125 suggestions: ["author", "asked_by", "decided_by"], folio_suggestions: ["author", "asked_by", "decided_by"], folio_proposals: ["author", "asked_by", "decided_by"],
126 space_members: ["added_by"], folio_grants: ["granted_by"], repo_spaces: ["added_by"],
127 page_changes: ["cleared_by"], folio_changes: ["cleared_by"],
128 };
129 for (const [table, columns] of Object.entries(authors)) for (const column of columns) statements.push(db.prepare(`UPDATE ${table} SET ${column} = ? WHERE ${column} = ?`).bind(GHOST, key));
130 // The cheap LIKE first, so json_each runs only on the versions that name them.
131 for (const table of ["page_versions", "folio_versions"]) statements.push(db.prepare(`UPDATE ${table} SET authors = (SELECT json_group_array(CASE WHEN value = ?1 THEN ?2 ELSE value END) FROM json_each(${table}.authors)) WHERE authors LIKE '%' || ?1 || '%' AND EXISTS (SELECT 1 FROM json_each(${table}.authors) WHERE value = ?1)`).bind(key, GHOST));
132 // Who was already told of a mention, by username.
133 const username = (args.username ?? "").toLowerCase();
134 if (username) for (const table of ["pages", "folios"]) statements.push(db.prepare(`UPDATE ${table} SET mentioned = (SELECT json_group_array(value) FROM json_each(${table}.mentioned) WHERE value != ?1) WHERE mentioned LIKE '%' || ?1 || '%' AND EXISTS (SELECT 1 FROM json_each(${table}.mentioned) WHERE value = ?1)`).bind(username));
135 for (let i = 0; i < statements.length; i += 80) await db.batch(statements.slice(i, i + 80));
136 }
137 await db.prepare("DELETE FROM purge_access WHERE subject_id = ?").bind(args.id).run();
138 return { ok: true };
139}