| 1 | import { GHOST_ID } from "@g1t/contracts"; |
| 2 | |
| 3 | import { g1tDmPerson } from "./g1t.ts"; |
| 4 | import type { ChannelRoom } from "./room.ts"; |
| 5 | import { clearConversation } from "./clear.ts"; |
| 6 | |
| 7 | type Env = { DB: D1Database; ROOMS: DurableObjectNamespace<ChannelRoom>; AVATARS: KVNamespace }; |
| 8 | export type Erasure = { kind: "user" | "workspace"; id: string; username?: string; slug?: string }; |
| 9 | |
| 10 | const GHOST = `user:${GHOST_ID}`; |
| 11 | /** Messages deleted per statement, and per call, so one huge channel never holds the database in one write. */ |
| 12 | const CHUNK = 500; |
| 13 | const BUDGET = 5000; |
| 14 | /** Channels handled per call; the rest wait for identity's next retry. */ |
| 15 | const CHANNELS = 25; |
| 16 | |
| 17 | /** |
| 18 | * Called only by identity's durable purge jobs, through a service binding. |
| 19 | * Safe to repeat: every step is idempotent, and work left over answers |
| 20 | * `pending` so identity calls again rather than recording success. |
| 21 | */ |
| 22 | export async function purgePersonalData(env: Env, args: Erasure): Promise<{ ok: boolean; pending?: boolean }> { |
| 23 | if (!args?.id || !["user", "workspace"].includes(args.kind)) throw new Error("Invalid purge subject"); |
| 24 | const db = env.DB, key = `user:${args.id}`; |
| 25 | // The channels to visit are listed once, in the transaction that writes |
| 26 | // the tombstone: a retry works through the list instead of scanning |
| 27 | // every message again, and the list can never be skipped. |
| 28 | const unseeded = "NOT EXISTS (SELECT 1 FROM erased_subjects WHERE id = ?1)"; |
| 29 | await db.batch([ |
| 30 | args.kind === "workspace" |
| 31 | ? db.prepare(`INSERT OR IGNORE INTO purge_channels SELECT ?1, id FROM channels WHERE workspace_id = ?1 AND ${unseeded}`).bind(args.id) |
| 32 | : db.prepare(`INSERT OR IGNORE INTO purge_channels SELECT ?1, id FROM channels WHERE (created_by = ?2 OR id IN (SELECT channel_id FROM channel_members WHERE principal = ?2 UNION SELECT channel_id FROM messages WHERE author = ?2)) AND ${unseeded}`).bind(args.id, key), |
| 33 | db.prepare("INSERT OR IGNORE INTO erased_subjects (id, kind) VALUES (?, ?)").bind(args.id, args.kind), |
| 34 | ]); |
| 35 | let budget = BUDGET; |
| 36 | const channels = (await db.prepare("SELECT channel_id FROM purge_channels WHERE subject_id = ? LIMIT ?").bind(args.id, CHANNELS).all<{ channel_id: string }>()).results; |
| 37 | for (const { channel_id: id } of channels) { |
| 38 | if (budget <= 0) break; |
| 39 | const channel = await db.prepare("SELECT kind FROM channels WHERE id = ?").bind(id).first<{ kind: string }>(); |
| 40 | const members = (await db.prepare("SELECT principal FROM channel_members WHERE channel_id = ?").bind(id).all<{ principal: string }>()).results.map(r => r.principal); |
| 41 | const destroy = args.kind === "workspace" || (channel?.kind === "dm" && g1tDmPerson(members) === key); |
| 42 | // References stay in purge_channels until both the room and database confirm cleanup. |
| 43 | await env.ROOMS.get(env.ROOMS.idFromName(id)).erase(id, destroy ? null : key); |
| 44 | if (destroy) { |
| 45 | // In chunks, newest last, until the channel is empty or this call's budget is spent. |
| 46 | let left = true; |
| 47 | while (budget > 0) { |
| 48 | const [, gone] = await db.batch([ |
| 49 | db.prepare(`DELETE FROM reactions WHERE message_id IN (SELECT id FROM messages WHERE channel_id = ? ORDER BY id LIMIT ${CHUNK})`).bind(id), |
| 50 | db.prepare(`DELETE FROM messages WHERE id IN (SELECT id FROM messages WHERE channel_id = ? ORDER BY id LIMIT ${CHUNK})`).bind(id), |
| 51 | ]); |
| 52 | const changes = Number(gone?.meta?.changes ?? 0); |
| 53 | budget -= Math.max(changes, 1); |
| 54 | if (changes < CHUNK) { left = false; break; } |
| 55 | } |
| 56 | if (left) break; |
| 57 | await clearConversation(db, id); |
| 58 | await db.batch([ |
| 59 | db.prepare("DELETE FROM channel_members WHERE channel_id = ?").bind(id), |
| 60 | db.prepare("DELETE FROM channels WHERE id = ?").bind(id), |
| 61 | ]); |
| 62 | } else { |
| 63 | await db.batch([ |
| 64 | db.prepare("UPDATE messages SET author = ? WHERE channel_id = ? AND author = ?").bind(GHOST, id, key), |
| 65 | db.prepare("DELETE FROM channel_members WHERE channel_id = ? AND principal = ?").bind(id, key), |
| 66 | db.prepare("UPDATE channels SET created_by = CASE WHEN created_by = ?2 THEN ?3 ELSE created_by END, dm_key = CASE WHEN kind = 'dm' THEN NULL ELSE dm_key END WHERE id = ?1").bind(id, key, GHOST), |
| 67 | ]); |
| 68 | budget -= 1; |
| 69 | } |
| 70 | await db.prepare("DELETE FROM purge_channels WHERE subject_id = ? AND channel_id = ?").bind(args.id, id).run(); |
| 71 | } |
| 72 | if (await db.prepare("SELECT 1 FROM purge_channels WHERE subject_id = ? LIMIT 1").bind(args.id).first()) return { ok: false, pending: true }; |
| 73 | if (args.kind === "user") { |
| 74 | await db.batch([ |
| 75 | db.prepare("DELETE FROM reactions WHERE principal = ?").bind(key), |
| 76 | db.prepare("DELETE FROM general_joined WHERE principal = ?").bind(key), |
| 77 | db.prepare("UPDATE custom_emoji SET created_by = ? WHERE created_by = ?").bind(GHOST, key), |
| 78 | ]); |
| 79 | } else { |
| 80 | const files = (await db.prepare("SELECT DISTINCT file FROM custom_emoji WHERE workspace_id = ? AND file IS NOT NULL").bind(args.id).all<{ file: string }>()).results; |
| 81 | for (const { file } of files) { |
| 82 | const used = await db.prepare("SELECT 1 FROM custom_emoji WHERE workspace_id != ? AND file = ? LIMIT 1").bind(args.id, file).first(); |
| 83 | if (!used) await env.AVATARS.delete(`emoji/${file}`); |
| 84 | } |
| 85 | await db.batch(["general_joined", "custom_emoji", "chat_settings", "former_agents", "chat_meter"].map(table => db.prepare(`DELETE FROM ${table} WHERE workspace_id = ?`).bind(args.id))); |
| 86 | } |
| 87 | return { ok: true }; |
| 88 | } |