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
264 linesCodeBlameRaw
1import assert from "node:assert/strict";
2import { readFileSync, readdirSync } from "node:fs";
3import { DatabaseSync } from "node:sqlite";
4import { test } from "node:test";
5import { build } from "esbuild";
6
7// Execute the real service code against SQLite, with only the Cloudflare host replaced.
8async function service(path) {
9 const bundle = await build({ entryPoints: [path], bundle: true, write: false, format: "esm", platform: "node", plugins: [{ name: "worker-host", setup(build) {
10 build.onResolve({ filter: /^yjs$/ }, () => ({ path: import.meta.resolve("yjs"), external: true }));
11 build.onResolve({ filter: /^cloudflare:workers$/ }, () => ({ path: "host", namespace: "host" }));
12 build.onLoad({ filter: /.*/, namespace: "host" }, () => ({ contents: "export class DurableObject { constructor(ctx, env) { this.ctx = ctx; this.env = env; } }" }));
13 }}] });
14 return import(`data:text/javascript;base64,${Buffer.from(bundle.outputFiles[0].text).toString("base64")}`);
15}
16
17function database(service) {
18 const raw = new DatabaseSync(":memory:");
19 const dir = new URL(`../services/${service}/migrations/`, import.meta.url);
20 for (const name of readdirSync(dir).sort()) raw.exec(readFileSync(new URL(name, dir), "utf8"));
21 raw.exec("PRAGMA foreign_keys = ON");
22 const run = (s) => ({ meta: { changes: Number(raw.prepare(s.sql).run(...s.params).changes) } });
23 const prepare = (sql, params = []) => ({ sql, params, bind: (...p) => prepare(sql, p), first: async () => raw.prepare(sql).get(...params) ?? null, all: async () => ({ results: raw.prepare(sql).all(...params) }), run: async () => run({ sql, params }) });
24 return { raw, prepare, batch: async (statements) => { raw.exec("BEGIN"); try { const result = statements.map(run); raw.exec("COMMIT"); return result; } catch (e) { raw.exec("ROLLBACK"); throw e; } } };
25}
26const namespace = (get) => ({ idFromName: id => id, get });
27const AT = "2026-10-14T00:00:00.000Z";
28
29test("identity purge work is atomic, idempotent and a prior restore prevents scheduling", () => {
30 const db = database("identity").raw;
31 const enqueue = db.prepare(readFileSync(new URL("../services/identity/src/purge_job.sql", import.meta.url), "utf8"));
32 db.prepare("INSERT INTO users (id, username, password_hash, created_at, deleted_at, purge_after) VALUES ('usr_fixture', 'fixture', 'fixture', ?, ?, ?)").run(AT, AT, AT);
33 const args = { "?1": "usr_fixture", "?2": "user", "?3": "CHAT", "?4": "{}", "?5": AT };
34 db.exec("BEGIN");
35 enqueue.run(args);
36 db.exec("DELETE FROM users WHERE id = 'usr_fixture'");
37 db.exec("ROLLBACK");
38 assert.equal(db.prepare("SELECT count(*) n FROM purge_jobs").get().n, 0);
39 assert.ok(db.prepare("SELECT 1 FROM users WHERE id = 'usr_fixture'").get());
40 db.exec("UPDATE users SET deleted_at = NULL, purge_after = NULL WHERE id = 'usr_fixture'");
41 enqueue.run(args);
42 assert.equal(db.prepare("SELECT count(*) n FROM purge_jobs").get().n, 0);
43 db.prepare("UPDATE users SET deleted_at = ?, purge_after = ? WHERE id = 'usr_fixture'").run(AT, AT);
44 db.exec("BEGIN");
45 enqueue.run(args);
46 enqueue.run(args);
47 db.exec("DELETE FROM users WHERE id = 'usr_fixture'; COMMIT");
48 assert.equal(db.prepare("SELECT count(*) n FROM purge_jobs").get().n, 1);
49 assert.equal(db.prepare("SELECT completed_at FROM purge_jobs").get().completed_at, null);
50});
51
52test("chat purges the private AI conversation, preserves other tenants, retries failed rooms and rejects late writes", async () => {
53 const { purgePersonalData } = await service("services/chat/src/erasure.ts");
54 const db = database("chat");
55 const { G1T_KEY } = await service("packages/contracts/src/chat.ts");
56 for (const [id, person] of [["mine", "usr_a"], ["other", "usr_b"]]) {
57 db.raw.prepare("INSERT INTO channels (id, workspace_id, kind, created_by, created_at) VALUES (?, ?, 'dm', ?, ?)").run(id, id, `user:${person}`, AT);
58 for (const key of [`user:${person}`, G1T_KEY]) db.raw.prepare("INSERT INTO channel_members (channel_id, principal, joined_at) VALUES (?, ?, ?)").run(id, key, AT);
59 db.raw.prepare("INSERT INTO messages (id, channel_id, author, body, created_at) VALUES (?, ?, ?, 'private prompt', ?)").run(id, id, `user:${person}`, AT);
60 }
61 let fail = true;
62 const env = { DB: db, ROOMS: namespace(() => ({ erase: async () => { if (fail) throw Error("unavailable"); } })), AVATARS: { delete: async () => {} } };
63 await assert.rejects(purgePersonalData(env, { kind: "user", id: "usr_a" }));
64 assert.equal(db.raw.prepare("SELECT count(*) n FROM messages").get().n, 2);
65 fail = false;
66 assert.deepEqual(await purgePersonalData(env, { kind: "user", id: "usr_a" }), { ok: true });
67 assert.deepEqual(await purgePersonalData(env, { kind: "user", id: "usr_a" }), { ok: true });
68 assert.equal(db.raw.prepare("SELECT count(*) n FROM messages").get().n, 1);
69 assert.equal(db.raw.prepare("SELECT body FROM messages WHERE id = 'other'").get().body, "private prompt");
70 db.raw.prepare("INSERT INTO messages (id, channel_id, author, body, created_at) VALUES ('late', 'mine', 'user:usr_a', 'late', ?)").run(AT);
71 assert.equal(db.raw.prepare("SELECT count(*) n FROM messages").get().n, 1);
72});
73
74test("artifacts keep retry references when file deletion fails, then remove private state with FK enforcement", async () => {
75 const { purgePersonalData } = await service("services/artifacts/src/erasure.ts");
76 const db = database("artifacts");
77 for (const [id, owner] of [["mine", "usr_a"], ["other", "usr_b"]]) {
78 db.raw.prepare("INSERT INTO folios (id, workspace_id, kind, owner, position, acl_root, path, created_by, created_at, updated_at, edited_at) VALUES (?, ?, 'doc', ?, 1, ?, ?, ?, ?, ?, ?)").run(id, id, `user:${owner}`, id, `/${id}/`, `user:${owner}`, AT, AT, AT);
79 db.raw.prepare("INSERT INTO folio_files (id, workspace_id, folio_id, key, name, content_type, bytes, created_by, created_at) VALUES (?, ?, ?, ?, 'private.txt', 'text/plain', 10, ?, ?)").run(id, id, id, id, `user:${owner}`, AT);
80 }
81 let fail = true; const deleted = [], rooms = [];
82 const env = { DB: db, FOLIOS: namespace(id => ({ forgetPerson: async () => {}, destroy: async () => { rooms.push(id); } })), FILES: { delete: async key => { if (fail) throw Error("storage down"); deleted.push(key); } } };
83 await assert.rejects(purgePersonalData(env, { kind: "user", id: "usr_a" }));
84 assert.equal(db.raw.prepare("SELECT count(*) n FROM folio_files").get().n, 2);
85 fail = false;
86 assert.deepEqual(await purgePersonalData(env, { kind: "user", id: "usr_a" }), { ok: true });
87 assert.deepEqual(await purgePersonalData(env, { kind: "user", id: "usr_a" }), { ok: true });
88 assert.deepEqual(deleted, ["docs/mine"]);
89 assert.equal(db.raw.prepare("SELECT count(*) n FROM folios").get().n, 1);
90 assert.equal(db.raw.prepare("SELECT id FROM folio_files").get().id, "other");
91 assert.ok(rooms.every(id => id === "mine"));
92 assert.deepEqual(await purgePersonalData(env, { kind: "workspace", id: "other" }), { ok: true });
93 assert.equal(db.raw.prepare("SELECT count(*) n FROM folios").get().n, 0);
94});
95
96function objectState() {
97 const raw = new DatabaseSync(":memory:");
98 const kv = new Map();
99 const sql = { exec(query, ...params) {
100 const stmt = raw.prepare(query);
101 if (stmt.columns().length) { const rows = stmt.all(...params); return { toArray: () => rows, one: () => rows[0], [Symbol.iterator]: () => rows[Symbol.iterator]() }; }
102 const result = stmt.run(...params); return { toArray: () => [], rowsWritten: Number(result.changes) };
103 } };
104 return { raw, blockConcurrencyWhile: fn => fn(), storage: { sql, get: async key => kv.get(key), put: async (key, value) => { kv.set(key, value); }, delete: async key => kv.delete(key), deleteAlarm: async () => {}, transactionSync(fn) { raw.exec("BEGIN"); try { fn(); raw.exec("COMMIT"); } catch (e) { raw.exec("ROLLBACK"); throw e; } } }, getWebSockets: () => [], setWebSocketAutoResponse: () => {}, waitUntil: () => {} };
105}
106
107test("notification erasure retries presence cleanup, clears credentials and permanently blocks late delivery", async () => {
108 globalThis.WebSocketRequestResponsePair = class {};
109 const { Feed } = await service("services/notify/src/feed.ts");
110 const state = objectState(); let fail = true, calls = 0;
111 const feed = new Feed(state, { ROOMS: namespace(() => ({ forget: async () => { calls++; if (fail) throw Error("offline"); } })) });
112 state.raw.exec(`INSERT INTO meta VALUES ('workspaces', '["acme"]'); INSERT INTO devices VALUES ('token', 'ios', 'phone', 1); INSERT INTO notifications VALUES ('n', 1, '{}');`);
113 await assert.rejects(feed.erase("usr_a"));
114 // A late delivery is dropped quietly, never refused: senders would log and retry it.
115 assert.deepEqual(await feed.deliver([{ workspace: "acme", notification: { id: "late", created_at: AT } }]), { ok: true });
116 assert.deepEqual(await feed.notify({ id: "late", kind: "mention", workspace: "acme", created_at: AT }), { ok: true });
117 fail = false;
118 assert.deepEqual(await feed.erase("usr_a"), { ok: true });
119 assert.equal(calls, 2);
120 for (const table of ["devices", "notifications", "counts", "subscriptions", "visits", "watching"]) assert.equal(state.raw.prepare(`SELECT count(*) n FROM ${table}`).get().n, 0);
121 assert.deepEqual(state.raw.prepare("SELECT key FROM meta").all().map(r => r.key), ["erased"]);
122 await assert.rejects(feed.registerDevice({ token: "late", platform: "ios" }), /Account erased/);
123 assert.equal((await feed.fetch(new Request("https://test/live"))).status, 410);
124 assert.deepEqual(await feed.erase("usr_a"), { ok: true });
125});
126
127test("destroyed document rooms reject late ensure calls and keep only a tombstone", async () => {
128 globalThis.WebSocketRequestResponsePair = class {};
129 for (const [file, name, init] of [
130 ["services/artifacts/src/room.ts", "PageRoom", { page_id: "p", workspace_slug: "acme", markdown: "private" }],
131 ["services/artifacts/src/folios/room.ts", "FolioRoom", { folio_id: "f", kind: "doc", workspace_slug: "acme", text: "private" }],
132 ]) {
133 const module = await service(file), state = objectState();
134 const room = new module[name](state, {});
135 state.raw.exec("INSERT INTO meta VALUES ('workspace_slug', '\"acme\"'); INSERT INTO snapshot VALUES (1, x'01');");
136 await room.destroy();
137 await room.destroy();
138 assert.equal(state.raw.prepare("SELECT count(*) n FROM snapshot").get().n, 0);
139 assert.deepEqual(state.raw.prepare("SELECT key FROM meta").all().map(r => r.key), ["erased"]);
140 await assert.rejects(room.ensure(init), /Document erased/);
141 assert.equal((await room.fetch(new Request("https://test/live"))).status, 410);
142 assert.deepEqual(state.raw.prepare("SELECT key FROM meta").all().map(r => r.key), ["erased"]);
143 }
144});
145
146test("erasing one member of a shared channel leaves everyone else's conversation alone", async () => {
147 const { purgePersonalData } = await service("services/chat/src/erasure.ts");
148 const db = database("chat");
149 db.raw.prepare("INSERT INTO channels (id, workspace_id, kind, name, created_by, created_at) VALUES ('general', 'ws', 'channel', 'general', 'user:usr_a', ?)").run(AT);
150 for (const person of ["usr_a", "usr_b"]) db.raw.prepare("INSERT INTO channel_members (channel_id, principal, joined_at) VALUES ('general', ?, ?)").run(`user:${person}`, AT);
151 db.raw.prepare("INSERT INTO messages (id, channel_id, author, body, created_at) VALUES ('m1', 'general', 'user:usr_a', 'from a', ?), ('m2', 'general', 'user:usr_b', 'from b', ?)").run(AT, AT);
152 db.raw.prepare("INSERT INTO reactions (message_id, principal, emoji, created_at) VALUES ('m2', 'user:usr_a', 'ok', ?), ('m2', 'user:usr_b', 'ok', ?)").run(AT, AT);
153 const erased = [];
154 const env = { DB: db, ROOMS: namespace(() => ({ erase: async (id, principal) => { erased.push([id, principal]); } })), AVATARS: { delete: async () => {} } };
155 assert.deepEqual(await purgePersonalData(env, { kind: "user", id: "usr_a" }), { ok: true });
156 // The room shuts out only the erased person; it is never cleared for everyone.
157 assert.deepEqual(erased, [["general", "user:usr_a"]]);
158 assert.deepEqual(db.raw.prepare("SELECT id, author, body FROM messages ORDER BY id").all().map(r => ({ ...r })), [
159 { id: "m1", author: "user:usr_ghost", body: "from a" },
160 { id: "m2", author: "user:usr_b", body: "from b" },
161 ]);
162 assert.deepEqual(db.raw.prepare("SELECT principal FROM channel_members").all().map(r => r.principal), ["user:usr_b"]);
163 assert.deepEqual(db.raw.prepare("SELECT principal FROM reactions").all().map(r => r.principal), ["user:usr_b"]);
164 assert.equal(db.raw.prepare("SELECT created_by FROM channels").get().created_by, "user:usr_ghost");
165});
166
167test("a shared channel's room shuts out only the erased member and tells nobody it was cleared", async () => {
168 globalThis.WebSocketRequestResponsePair = class {};
169 const { ChannelRoom } = await service("services/chat/src/room.ts");
170 const state = objectState();
171 const socket = (who) => ({ who, sent: [], closed: null, send(text) { this.sent.push(text); }, close(code) { this.closed = code; } });
172 const sockets = [socket("user:usr_a"), socket("user:usr_b")];
173 state.getWebSockets = (tag) => sockets.filter(s => !tag || s.who === tag);
174 await state.storage.put("job", { message: { id: "answer_for_b" } });
175 const deletes = [];
176 const room = new ChannelRoom(state, { DB: { prepare: () => ({ bind: (id) => ({ run: async () => { deletes.push(id); } }) }) } });
177 await room.erase("general", "user:usr_a");
178 assert.equal(sockets[0].closed, 4403);
179 assert.equal(sockets[1].closed, null);
180 assert.deepEqual(sockets[1].sent, []);
181 // Someone else's pending answer carries on.
182 assert.deepEqual(await state.storage.get("job"), { message: { id: "answer_for_b" } });
183 assert.deepEqual(deletes, []);
184 assert.equal((await room.fetch(new Request("https://test/live"))).status, 426);
185});
186
187test("a workspace's large channel is deleted in chunks across calls, and nothing of it comes back", async () => {
188 const { purgePersonalData } = await service("services/chat/src/erasure.ts");
189 const db = database("chat");
190 db.raw.prepare("INSERT INTO channels (id, workspace_id, kind, name, created_by, created_at) VALUES ('big', 'ws', 'channel', 'big', 'user:usr_a', ?), ('kept', 'other', 'channel', 'kept', 'user:usr_b', ?)").run(AT, AT);
191 const insert = db.raw.prepare("INSERT INTO messages (id, channel_id, author, body, created_at) VALUES (?, ?, 'user:usr_a', 'x', ?)");
192 for (let i = 0; i < 5600; i++) insert.run(`m${String(i).padStart(5, "0")}`, "big", AT);
193 insert.run("other", "kept", AT);
194 const env = { DB: db, ROOMS: namespace(() => ({ erase: async () => {} })), AVATARS: { delete: async () => {} } };
195 assert.deepEqual(await purgePersonalData(env, { kind: "workspace", id: "ws" }), { ok: false, pending: true });
196 assert.ok(db.raw.prepare("SELECT count(*) n FROM messages WHERE channel_id = 'big'").get().n > 0);
197 assert.deepEqual(await purgePersonalData(env, { kind: "workspace", id: "ws" }), { ok: true });
198 assert.equal(db.raw.prepare("SELECT count(*) n FROM messages WHERE channel_id = 'big'").get().n, 0);
199 assert.equal(db.raw.prepare("SELECT count(*) n FROM channels WHERE workspace_id = 'ws'").get().n, 0);
200 assert.equal(db.raw.prepare("SELECT body FROM messages WHERE id = 'other'").get().body, "x");
201 db.raw.prepare("INSERT INTO channels (id, workspace_id, kind, name, created_by, created_at) VALUES ('late', 'ws', 'channel', 'late', 'user:usr_b', ?)").run(AT);
202 db.raw.prepare("INSERT INTO chat_settings (workspace_id) VALUES ('ws')").run();
203 assert.equal(db.raw.prepare("SELECT count(*) n FROM channels WHERE workspace_id = 'ws'").get().n, 0);
204 assert.equal(db.raw.prepare("SELECT count(*) n FROM chat_settings").get().n, 0);
205});
206
207test("a purged workspace's slug can be made again: only what was said before the purge is dropped", async () => {
208 globalThis.WebSocketRequestResponsePair = class {};
209 const { Feed } = await service("services/notify/src/feed.ts");
210 const { Room } = await service("services/notify/src/room.ts");
211 const state = objectState();
212 const feed = new Feed(state, {});
213 const before = new Date(Date.now() - 60_000).toISOString();
214 state.raw.prepare("INSERT INTO notifications VALUES ('old', 1, ?)").run(JSON.stringify({ id: "old", workspace: "alice", created_at: before }));
215 state.raw.prepare("INSERT INTO meta VALUES ('complete:alice', '1'), ('forgot:alice:ch_1', ?), ('forgot:alicex:ch_1', ?)").run(before, before);
216 assert.deepEqual(await feed.eraseWorkspace("alice"), { ok: true });
217 assert.equal(state.raw.prepare("SELECT count(*) n FROM notifications").get().n, 0);
218 assert.deepEqual(state.raw.prepare("SELECT key FROM meta WHERE key LIKE 'forgot:%' OR key LIKE 'complete:%'").all().map(r => r.key), ["forgot:alicex:ch_1"]);
219 // A late notification from before the purge is dropped; one from the workspace made again is kept.
220 await feed.notify({ id: "late", kind: "mention", workspace: "alice", title: "t", url: "/alice", created_at: before });
221 await feed.notify({ id: "new", kind: "mention", workspace: "alice", title: "t", url: "/alice", created_at: new Date(Date.now() + 1000).toISOString() });
222 assert.deepEqual(state.raw.prepare("SELECT id FROM notifications").all().map(r => r.id), ["new"]);
223
224 const room = new Room(objectState(), {});
225 await room.report("alice", { user_id: "usr_b", at: 1, presence: "online" });
226 assert.deepEqual(await room.eraseMembers(), ["usr_b"]);
227 await room.dropPerson("usr_b");
228 assert.equal(await room.report("alice", { user_id: "usr_c", at: 2, presence: "online" }), null);
229 await room.report("alice", { user_id: "usr_b", at: Date.now() + 1000, presence: "online" });
230 assert.deepEqual((await room.people()).map(p => p.user_id), ["usr_b"]);
231});
232
233test("vectors written while a document was being erased are taken back", async () => {
234 const { dropIfErased, erasedSource } = await service("services/artifacts/src/indexer.ts");
235 const db = database("artifacts");
236 const deleted = [];
237 const store = { delete: async ids => { deleted.push(...ids); } };
238 await dropIfErased(db, store, "folio", "f1", "ws", ["f1:0"]);
239 assert.deepEqual(deleted, []);
240 db.raw.prepare("INSERT INTO erased_documents (id, kind) VALUES ('f1', 'folio')").run();
241 assert.equal(await erasedSource(db, "folio", "f1", "ws"), true);
242 assert.equal(await erasedSource(db, "page", "f1", "ws"), false);
243 await dropIfErased(db, store, "folio", "f1", "ws", ["f1:0", "f1:1"]);
244 assert.deepEqual(deleted, ["f1:0", "f1:1"]);
245 db.raw.prepare("INSERT INTO erased_subjects VALUES ('ws2', 'workspace')").run();
246 assert.equal(await erasedSource(db, null, "repo_file", "ws2"), true);
247});
248
249test("a workspace's uploads are swept a page at a time, resuming after a failure", async () => {
250 const { purgePersonalData } = await service("services/artifacts/src/erasure.ts");
251 const db = database("artifacts");
252 const insert = db.raw.prepare("INSERT INTO files (id, workspace_id, page_id, key, name, content_type, bytes, created_by, created_at) VALUES (?, ?, NULL, ?, 'f', 'text/plain', 1, 'user:usr_a', ?)");
253 for (let i = 0; i < 2100; i++) insert.run(`f${i}`, "ws", `k${i}`, AT);
254 insert.run("kept", "other", "kept", AT);
255 let fail = 3; const deleted = new Set();
256 const env = { DB: db, FILES: { delete: async key => { if (key === "docs/k5" && fail-- > 0) throw Error("storage down"); deleted.add(key); } } };
257 await assert.rejects(purgePersonalData(env, { kind: "workspace", id: "ws" }));
258 let result;
259 for (let i = 0; i < 6 && !result?.ok; i++) result = await purgePersonalData(env, { kind: "workspace", id: "ws" }).catch(() => null);
260 assert.deepEqual(result, { ok: true });
261 assert.equal(deleted.size, 2100);
262 assert.ok(!deleted.has("docs/kept"));
263 assert.equal(db.raw.prepare("SELECT id FROM files").get().id, "kept");
264});