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
115 linesCodeBlameRaw
1import { GHOST_ID } from "@g1t/contracts";
2
3import { g1tDmPerson } from "./g1t.ts";
4import type { ChannelRoom } from "./room.ts";
5import { clearConversation } from "./clear.ts";
6
7type Env = { DB: D1Database; ROOMS: DurableObjectNamespace<ChannelRoom>; AVATARS: KVNamespace };
8export type Erasure = { kind: "user" | "workspace"; id: string; username?: string; slug?: string };
9
10const GHOST = `user:${GHOST_ID}`;
11/** Messages deleted per statement, and per call, so one huge channel never holds the database in one write. */
12const CHUNK = 500;
13const BUDGET = 5000;
14/** Channels handled per call; the rest wait for identity's next retry. */
15const 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 */
22export 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,
25 key = `user:${args.id}`;
26 // The channels to visit are listed once, in the transaction that writes
27 // the tombstone: a retry works through the list instead of scanning
28 // every message again, and the list can never be skipped.
29 const unseeded = "NOT EXISTS (SELECT 1 FROM erased_subjects WHERE id = ?1)";
30 await db.batch([
31 args.kind === "workspace"
32 ? db.prepare(`INSERT OR IGNORE INTO purge_channels SELECT ?1, id FROM channels WHERE workspace_id = ?1 AND ${unseeded}`).bind(args.id)
33 : db
34 .prepare(
35 `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}`,
36 )
37 .bind(args.id, key),
38 db.prepare("INSERT OR IGNORE INTO erased_subjects (id, kind) VALUES (?, ?)").bind(args.id, args.kind),
39 ]);
40 let budget = BUDGET;
41 const channels = (
42 await db.prepare("SELECT channel_id FROM purge_channels WHERE subject_id = ? LIMIT ?").bind(args.id, CHANNELS).all<{ channel_id: string }>()
43 ).results;
44 for (const { channel_id: id } of channels) {
45 if (budget <= 0) break;
46 const channel = await db.prepare("SELECT kind FROM channels WHERE id = ?").bind(id).first<{ kind: string }>();
47 const members = (await db.prepare("SELECT principal FROM channel_members WHERE channel_id = ?").bind(id).all<{ principal: string }>()).results.map(
48 (r) => r.principal,
49 );
50 const destroy = args.kind === "workspace" || (channel?.kind === "dm" && g1tDmPerson(members) === key);
51 // References stay in purge_channels until both the room and database confirm cleanup.
52 await env.ROOMS.get(env.ROOMS.idFromName(id)).erase(id, destroy ? null : key);
53 if (destroy) {
54 // In chunks, newest last, until the channel is empty or this call's budget is spent.
55 let left = true;
56 while (budget > 0) {
57 const [, gone] = await db.batch([
58 db.prepare(`DELETE FROM reactions WHERE message_id IN (SELECT id FROM messages WHERE channel_id = ? ORDER BY id LIMIT ${CHUNK})`).bind(id),
59 db.prepare(`DELETE FROM messages WHERE id IN (SELECT id FROM messages WHERE channel_id = ? ORDER BY id LIMIT ${CHUNK})`).bind(id),
60 ]);
61 const changes = Number(gone?.meta?.changes ?? 0);
62 budget -= Math.max(changes, 1);
63 if (changes < CHUNK) {
64 left = false;
65 break;
66 }
67 }
68 if (left) break;
69 await clearConversation(db, id);
70 await db.batch([
71 db.prepare("DELETE FROM channel_members WHERE channel_id = ?").bind(id),
72 db.prepare("DELETE FROM channel_removals WHERE channel_id = ?").bind(id),
73 db.prepare("DELETE FROM channels WHERE id = ?").bind(id),
74 ]);
75 } else {
76 await db.batch([
77 db.prepare("UPDATE messages SET author = ? WHERE channel_id = ? AND author = ?").bind(GHOST, id, key),
78 db.prepare("DELETE FROM channel_members WHERE channel_id = ? AND principal = ?").bind(id, key),
79 db
80 .prepare(
81 "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",
82 )
83 .bind(id, key, GHOST),
84 ]);
85 budget -= 1;
86 }
87 await db.prepare("DELETE FROM purge_channels WHERE subject_id = ? AND channel_id = ?").bind(args.id, id).run();
88 }
89 if (await db.prepare("SELECT 1 FROM purge_channels WHERE subject_id = ? LIMIT 1").bind(args.id).first()) return { ok: false, pending: true };
90 if (args.kind === "user") {
91 await db.batch([
92 db.prepare("DELETE FROM reactions WHERE principal = ?").bind(key),
93 db.prepare("DELETE FROM general_joined WHERE principal = ?").bind(key),
94 db.prepare("UPDATE custom_emoji SET created_by = ? WHERE created_by = ?").bind(GHOST, key),
95 // Invitations, requests and removals name them; what someone else sent goes with them too.
96 db.prepare("DELETE FROM channel_invitations WHERE invitee = ?1 OR invited_by = ?1").bind(key),
97 db.prepare("UPDATE channel_requests SET answered_by = ?2 WHERE answered_by = ?1").bind(key, GHOST),
98 db.prepare("DELETE FROM channel_requests WHERE requester = ?").bind(key),
99 db.prepare("DELETE FROM channel_removals WHERE principal = ?1 OR removed_by = ?1").bind(key),
100 ]);
101 } else {
102 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 }>())
103 .results;
104 for (const { file } of files) {
105 const used = await db.prepare("SELECT 1 FROM custom_emoji WHERE workspace_id != ? AND file = ? LIMIT 1").bind(args.id, file).first();
106 if (!used) await env.AVATARS.delete(`emoji/${file}`);
107 }
108 await db.batch(
109 ["general_joined", "custom_emoji", "chat_settings", "former_agents", "chat_meter", "channel_invitations", "channel_requests"].map((table) =>
110 db.prepare(`DELETE FROM ${table} WHERE workspace_id = ?`).bind(args.id),
111 ),
112 );
113 }
114 return { ok: true };
115}