| 1 | import { GHOST_ID } from "@g1t/contracts"; |
| 2 | |
| 3 | import { cloudflareVectors } from "./vectors.ts"; |
| 4 | import { fileStore, type FileStoreEnv } from "./files.ts"; |
| 5 | import type { IndexEnv } from "./indexer.ts"; |
| 6 | import type { PageRoom } from "./room.ts"; |
| 7 | import type { FolioRoom } from "./folios/room.ts"; |
| 8 | |
| 9 | type Env = IndexEnv & FileStoreEnv & { PAGES?: DurableObjectNamespace<PageRoom>; FOLIOS?: DurableObjectNamespace<FolioRoom> }; |
| 10 | type Subject = { kind: "user" | "workspace"; id: string; username?: string }; |
| 11 | const 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. */ |
| 13 | const PAGE = 200; |
| 14 | /** Documents destroyed per call; the rest wait for identity's next retry. */ |
| 15 | const DOCUMENTS = 25; |
| 16 | /** Sweep passes per call, so a large workspace stays within one invocation's subrequests. */ |
| 17 | const 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 | */ |
| 24 | export 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, |
| 27 | key = `user:${args.id}`; |
| 28 | await db.prepare("INSERT OR IGNORE INTO erased_subjects VALUES (?, ?)").bind(args.id, args.kind).run(); |
| 29 | if (args.kind === "workspace") { |
| 30 | await db.batch([ |
| 31 | db.prepare("INSERT OR IGNORE INTO purge_documents SELECT ?, 'page', id FROM pages WHERE workspace_id = ?").bind(args.id, args.id), |
| 32 | db.prepare("INSERT OR IGNORE INTO purge_documents SELECT ?, 'folio', id FROM folios WHERE workspace_id = ?").bind(args.id, args.id), |
| 33 | ]); |
| 34 | } else { |
| 35 | await db.batch([ |
| 36 | db |
| 37 | .prepare( |
| 38 | "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)", |
| 39 | ) |
| 40 | .bind(args.id, key), |
| 41 | db |
| 42 | .prepare( |
| 43 | "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)", |
| 44 | ) |
| 45 | .bind(args.id, key), |
| 46 | ]); |
| 47 | const access = ( |
| 48 | await db |
| 49 | .prepare("SELECT kind, document_id FROM purge_access WHERE subject_id = ? AND done = 0 LIMIT 50") |
| 50 | .bind(args.id) |
| 51 | .all<{ kind: string; document_id: string }>() |
| 52 | ).results; |
| 53 | for (const doc of access) { |
| 54 | const rooms = doc.kind === "folio" ? env.FOLIOS : env.PAGES; |
| 55 | if (!rooms) throw new Error("Document access cleanup unavailable"); |
| 56 | await rooms.get(rooms.idFromName(doc.document_id)).forgetPerson(key, args.username ?? ""); |
| 57 | 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(); |
| 58 | } |
| 59 | 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 }; |
| 60 | // Only private folios nobody else can open, in any workspace: the folio |
| 61 | // and everything under it the person's own, with no other principal's |
| 62 | // access or grant. Shared material needs a case-specific assessment. |
| 63 | await db |
| 64 | .prepare( |
| 65 | `INSERT OR IGNORE INTO purge_documents SELECT ?1, 'folio', f.id FROM folios f |
| 66 | WHERE f.owner = ?2 AND f.space_id IS NULL AND f.general_access = 'none' |
| 67 | AND NOT EXISTS (SELECT 1 FROM folios child WHERE substr(child.path, 1, length(f.path)) = f.path AND ( |
| 68 | child.owner != ?2 OR child.general_access != 'none' |
| 69 | OR EXISTS (SELECT 1 FROM folio_access a WHERE a.folio_id = child.id AND a.principal != ?2) |
| 70 | OR EXISTS (SELECT 1 FROM folio_grants g WHERE g.folio_id = child.id AND g.principal != ?2)))`, |
| 71 | ) |
| 72 | .bind(args.id, key) |
| 73 | .run(); |
| 74 | } |
| 75 | const docs = ( |
| 76 | await db |
| 77 | .prepare("SELECT kind, document_id FROM purge_documents WHERE subject_id = ? LIMIT ?") |
| 78 | .bind(args.id, DOCUMENTS) |
| 79 | .all<{ kind: string; document_id: string }>() |
| 80 | ).results; |
| 81 | for (const doc of docs) { |
| 82 | const folio = doc.kind === "folio", |
| 83 | id = doc.document_id; |
| 84 | // First, so an upload, an index run or a save racing this cannot write rows for it again (migration 0010, indexer.ts). |
| 85 | await db.prepare("INSERT OR IGNORE INTO erased_documents (id, kind) VALUES (?, ?)").bind(id, doc.kind).run(); |
| 86 | const rooms = folio ? env.FOLIOS : env.PAGES; |
| 87 | if (!rooms) throw new Error("Document room cleanup unavailable"); |
| 88 | await rooms.get(rooms.idFromName(id)).destroy(); |
| 89 | const chunks = ( |
| 90 | await db |
| 91 | .prepare(`SELECT id FROM ${folio ? "folio_chunks" : "doc_chunks"} WHERE ${folio ? "folio_id" : "page_id"} = ?`) |
| 92 | .bind(id) |
| 93 | .all<{ id: string }>() |
| 94 | ).results.map((r) => r.id); |
| 95 | const vectors = folio ? env.FOLIO_VECTORS : env.VECTORS; |
| 96 | if (vectors && chunks.length) await cloudflareVectors(vectors).delete(chunks); |
| 97 | const files = ( |
| 98 | await db |
| 99 | .prepare(`SELECT key FROM ${folio ? "folio_files" : "files"} WHERE ${folio ? "folio_id" : "page_id"} = ?`) |
| 100 | .bind(id) |
| 101 | .all<{ key: string }>() |
| 102 | ).results; |
| 103 | for (const { key: file } of files) await fileStore(env).delete(`docs/${file}`); |
| 104 | if (folio) { |
| 105 | const states = ( |
| 106 | await db |
| 107 | .prepare( |
| 108 | "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", |
| 109 | ) |
| 110 | .bind(id) |
| 111 | .all<{ key: string }>() |
| 112 | ).results; |
| 113 | for (const { key: file } of states) await fileStore(env).delete(file); |
| 114 | } |
| 115 | const table = folio ? "folios" : "pages", |
| 116 | column = folio ? "folio_id" : "page_id"; |
| 117 | await db.batch([ |
| 118 | // By document, so passages a late index run wrote without a chunk row go too. |
| 119 | db.prepare(`DELETE FROM ${folio ? "folio_chunks_fts WHERE folio_id" : "doc_chunks_fts WHERE doc_id"} = ?`).bind(id), |
| 120 | db.prepare(`DELETE FROM ${folio ? "folio_chunks" : "doc_chunks"} WHERE ${column} = ?`).bind(id), |
| 121 | db.prepare(`DELETE FROM ${folio ? "folios_fts" : "pages_fts"} WHERE ${column} = ?`).bind(id), |
| 122 | db.prepare(`DELETE FROM ${folio ? "folio_files" : "files"} WHERE ${column} = ?`).bind(id), |
| 123 | db.prepare(`UPDATE ${table} SET parent_id = NULL WHERE parent_id = ?`).bind(id), |
| 124 | db.prepare(`DELETE FROM ${table} WHERE id = ?`).bind(id), |
| 125 | db.prepare("DELETE FROM purge_documents WHERE subject_id = ? AND kind = ? AND document_id = ?").bind(args.id, doc.kind, id), |
| 126 | ]); |
| 127 | } |
| 128 | if (await db.prepare("SELECT 1 FROM purge_documents WHERE subject_id = ? LIMIT 1").bind(args.id).first()) return { ok: false, pending: true }; |
| 129 | if (args.kind === "workspace") { |
| 130 | // What no document holds any more: repository docs' passages, passages |
| 131 | // and uploads of documents deleted earlier. A page at a time, each |
| 132 | // deleted once its vectors or blobs are, until none is left. |
| 133 | let passes = PASSES; |
| 134 | for (const [table, store] of [ |
| 135 | ["doc_chunks", env.VECTORS], |
| 136 | ["folio_chunks", env.FOLIO_VECTORS], |
| 137 | ] as const) { |
| 138 | const fts = table === "doc_chunks" ? "doc_chunks_fts" : "folio_chunks_fts"; |
| 139 | for (;;) { |
| 140 | if (passes-- <= 0) return { ok: false, pending: true }; |
| 141 | const ids = (await db.prepare(`SELECT id FROM ${table} WHERE workspace_id = ? LIMIT ${PAGE}`).bind(args.id).all<{ id: string }>()).results.map( |
| 142 | (r) => r.id, |
| 143 | ); |
| 144 | if (!ids.length) break; |
| 145 | if (store) await cloudflareVectors(store).delete(ids); |
| 146 | await db.batch([ |
| 147 | db.prepare(`DELETE FROM ${fts} WHERE chunk_id IN (SELECT value FROM json_each(?))`).bind(JSON.stringify(ids)), |
| 148 | db.prepare(`DELETE FROM ${table} WHERE id IN (SELECT value FROM json_each(?))`).bind(JSON.stringify(ids)), |
| 149 | ]); |
| 150 | } |
| 151 | } |
| 152 | for (const table of ["files", "folio_files"]) { |
| 153 | for (;;) { |
| 154 | if (passes-- <= 0) return { ok: false, pending: true }; |
| 155 | const rows = (await db.prepare(`SELECT key FROM ${table} WHERE workspace_id = ? LIMIT ${PAGE}`).bind(args.id).all<{ key: string }>()).results; |
| 156 | if (!rows.length) break; |
| 157 | for (const row of rows) await fileStore(env).delete(`docs/${row.key}`); |
| 158 | await db |
| 159 | .prepare(`DELETE FROM ${table} WHERE key IN (SELECT value FROM json_each(?))`) |
| 160 | .bind(JSON.stringify(rows.map((r) => r.key))) |
| 161 | .run(); |
| 162 | } |
| 163 | } |
| 164 | await db.batch([ |
| 165 | db.prepare("DELETE FROM repo_files_fts WHERE space_id IN (SELECT id FROM repo_spaces WHERE workspace_id = ?)").bind(args.id), |
| 166 | ...["repo_spaces", "templates", "folio_templates", "spaces", "doc_index_runs", "doc_embed_usage", "workspace_billing"].map((table) => |
| 167 | db.prepare(`DELETE FROM ${table} WHERE workspace_id = ?`).bind(args.id), |
| 168 | ), |
| 169 | ]); |
| 170 | } else { |
| 171 | const statements: D1PreparedStatement[] = []; |
| 172 | for (const table of ["page_views", "favorites", "folio_visits", "folio_favorites", "folio_access_requests", "space_joins"]) |
| 173 | statements.push(db.prepare(`DELETE FROM ${table} WHERE user_id = ?`).bind(args.id)); |
| 174 | for (const table of ["space_members", "page_owners", "folio_grants", "folio_access"]) |
| 175 | statements.push(db.prepare(`DELETE FROM ${table} WHERE principal = ?`).bind(key)); |
| 176 | const authors: Record<string, string[]> = { |
| 177 | spaces: ["created_by", "deleted_by"], |
| 178 | pages: ["created_by", "updated_by", "archived_by"], |
| 179 | folios: ["owner", "created_by", "updated_by", "edited_by", "trashed_by"], |
| 180 | templates: ["created_by"], |
| 181 | folio_templates: ["created_by"], |
| 182 | files: ["created_by"], |
| 183 | folio_files: ["created_by"], |
| 184 | suggestions: ["author", "asked_by", "decided_by"], |
| 185 | folio_suggestions: ["author", "asked_by", "decided_by"], |
| 186 | folio_proposals: ["author", "asked_by", "decided_by"], |
| 187 | space_members: ["added_by"], |
| 188 | folio_grants: ["granted_by"], |
| 189 | repo_spaces: ["added_by"], |
| 190 | page_changes: ["cleared_by"], |
| 191 | folio_changes: ["cleared_by"], |
| 192 | }; |
| 193 | for (const [table, columns] of Object.entries(authors)) |
| 194 | for (const column of columns) statements.push(db.prepare(`UPDATE ${table} SET ${column} = ? WHERE ${column} = ?`).bind(GHOST, key)); |
| 195 | // The cheap LIKE first, so json_each runs only on the versions that name them. |
| 196 | for (const table of ["page_versions", "folio_versions"]) |
| 197 | statements.push( |
| 198 | db |
| 199 | .prepare( |
| 200 | `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)`, |
| 201 | ) |
| 202 | .bind(key, GHOST), |
| 203 | ); |
| 204 | // Who was already told of a mention, by username. |
| 205 | const username = (args.username ?? "").toLowerCase(); |
| 206 | if (username) |
| 207 | for (const table of ["pages", "folios"]) |
| 208 | statements.push( |
| 209 | db |
| 210 | .prepare( |
| 211 | `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)`, |
| 212 | ) |
| 213 | .bind(username), |
| 214 | ); |
| 215 | for (let i = 0; i < statements.length; i += 80) await db.batch(statements.slice(i, i + 80)); |
| 216 | } |
| 217 | await db.prepare("DELETE FROM purge_access WHERE subject_id = ?").bind(args.id).run(); |
| 218 | return { ok: true }; |
| 219 | } |