Something went wrong. Try again.
a tool for shared writing and social publishing
Something went wrong. Try again.
20 kB · 557 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558import { afterAll, describe, expect, test } from "vitest";import { Pool } from "pg";import { createClient } from "@supabase/supabase-js";import type { Database } from "supabase/database.types";import { randomUUID } from "node:crypto";import type { PatchOperation, ReadonlyJSONValue } from "replicache";import { fetchPullData, computePull } from "./serverPullData";import { parseCVRCookie } from "./cvr";import { makeCVRStore, type RedisLike } from "./cvrStore";
// computePull is exercised with the real store logic over a Map-backed fake// redis; only the ioredis transport itself is out of scope here.let redisMap = new Map<string, string>();let fakeRedis: RedisLike = { async get(key) { return redisMap.get(key) ?? null; }, async setex(key, _seconds, value) { redisMap.set(key, value); },};let store = makeCVRStore(fakeRedis);
// These tests run against the local supabase stack (which must be migrated// and running: `supabase start`): PostgREST for the pull function like// production, plus a direct pg connection for seeding and parity queries.// They never read .env.local so they can't accidentally point at production.const TEST_DB_URL = process.env.TEST_DB_URL ?? "postgresql://postgres:postgres@127.0.0.1:54322/postgres";// The standard supabase-CLI local demo credentials (`supabase status`).const TEST_SUPABASE_URL = process.env.TEST_SUPABASE_URL ?? "http://127.0.0.1:54321";const TEST_SUPABASE_SERVICE_KEY = process.env.TEST_SUPABASE_SERVICE_KEY ?? "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6InNlcnZpY2Vfcm9sZSIsImV4cCI6MTk4MzgxMjk5Nn0.EGIM96RAZx35lJzdJsyH-qQwv8Hdp7fsn3W0YpN81IU";
let pool = new Pool({ connectionString: TEST_DB_URL, max: 5 });let supabase = createClient<Database>( TEST_SUPABASE_URL, TEST_SUPABASE_SERVICE_KEY,);
let dbAvailable = await pool .query( "select 1 from pg_proc where proname = 'pull_data_cvr' and exists (select 1 from pg_proc where proname = 'pull_data')", ) .then((r) => r.rowCount === 1) .catch(() => false);if (!dbAvailable) { // Skipping locally is a convenience; in CI it would silently pass a broken // stack, so fail instead. if (process.env.CI) throw new Error( `Pull integration tests require a migrated database at ${TEST_DB_URL}`, ); console.warn( `Skipping pull integration tests: no migrated database at ${TEST_DB_URL}`, );}
// ---------------------------------------------------------------------------// Seeding helpers. Cleanup relies on ON DELETE CASCADE from entity_sets for// everything except documents and replicache_clients, which we track.// ---------------------------------------------------------------------------
let createdEntitySets: string[] = [];let createdDocuments: string[] = [];let createdClients: string[] = [];
async function createDoc() { let set = randomUUID(); let root = randomUUID(); let token = randomUUID(); createdEntitySets.push(set); await pool.query(`insert into entity_sets (id) values ($1)`, [set]); await pool.query(`insert into entities (id, set) values ($1, $2)`, [ root, set, ]); await pool.query( `insert into permission_tokens (id, root_entity) values ($1, $2)`, [token, root], ); return { set, root, token };}
async function addEntity(set: string) { let id = randomUUID(); await pool.query(`insert into entities (id, set) values ($1, $2)`, [id, set]); return id;}
async function addFact(entity: string, attribute: string, data: unknown) { let id = randomUUID(); await pool.query( `insert into facts (id, entity, attribute, data) values ($1, $2, $3, $4)`, [id, entity, attribute, JSON.stringify(data)], ); return id;}
async function setClient( client_id: string, client_group: string, last_mutation: number,) { createdClients.push(client_id); await pool.query( `insert into replicache_clients (client_id, client_group, last_mutation) values ($1, $2, $3) on conflict (client_id) do update set last_mutation = excluded.last_mutation`, [client_id, client_group, last_mutation], );}
async function addPublicationFor(token: string, title: string) { let did = `did:plc:test-${randomUUID().slice(0, 13)}`; let uri = `at://${did}/pub.leaflet.publication/test`; await pool.query( `insert into identities (id, home_page, atp_did) values ($1, $2, $3)`, [randomUUID(), token, did], ); await pool.query( `insert into publications (uri, name, identity_did) values ($1, $2, $3)`, [uri, "Test Publication", did], ); await pool.query( `insert into leaflets_in_publications (publication, leaflet, doc, title, description, tags, preferences) values ($1, $2, null, $3, 'a description', $4, $5)`, [uri, token, title, ["tag1", "tag2"], JSON.stringify({ theme: "dark" })], ); return { did, uri };}
function sortById<T extends { id: string }>(rows: T[]) { return [...rows].sort((a, b) => a.id.localeCompare(b.id));}
async function naivePullData(token: string, clientGroup: string) { let { rows } = await pool.query( `select * from pull_data($1::uuid, $2)`, [token, clientGroup], ); return rows[0] as { client_groups: | { client_id: string; client_group: string; last_mutation: number }[] | null; facts: { id: string; [k: string]: unknown }[] | null; publications: Record<string, unknown>[] | null; draft_contributors: string[] | null; };}
function applyPatch( kv: Map<string, ReadonlyJSONValue>, patch: readonly PatchOperation[],) { for (let op of patch) { if (op.op === "clear") kv.clear(); else if (op.op === "put") kv.set(op.key, op.value); else if (op.op === "del") kv.delete(op.key); }}
afterAll(async () => { if (dbAvailable) { if (createdEntitySets.length) await pool.query(`delete from entity_sets where id = any($1)`, [ createdEntitySets, ]); if (createdDocuments.length) await pool.query(`delete from documents where uri = any($1)`, [ createdDocuments, ]); if (createdClients.length) await pool.query(`delete from replicache_clients where client_id = any($1)`, [ createdClients, ]); } await pool.end();});
// ---------------------------------------------------------------------------
describe.runIf(dbAvailable)("fetchPullData parity with pull_data RPC", () => { test("fact closure, clients, publication and contributors all match", async () => { let { set, root, token } = await createDoc(); let child = await addEntity(set); let grandchild = await addEntity(set); let orphan = await addEntity(set);
await addFact(root, "block/text", { type: "text", value: "hello world" }); await addFact(root, "card/block", { type: "ordered-reference", value: child, position: "a0", }); await addFact(child, "card/block", { type: "reference", value: grandchild, }); await addFact(grandchild, "canvas/block", { type: "spatial-reference", value: root, // cycle back to the root position: { x: 0, y: 0 }, }); // orphan is not reachable from root and must not appear await addFact(orphan, "block/text", { type: "text", value: "orphan" });
let clientGroup = `cg-${randomUUID()}`; await setClient(`c1-${randomUUID()}`, clientGroup, 4); await setClient(`c2-${randomUUID()}`, clientGroup, 9); await addPublicationFor(token, "Pub Title"); await pool.query( `insert into leaflet_contributors (leaflet, contributor_did) select $1, atp_did from identities where home_page = $1`, [token], );
let naive = await naivePullData(token, clientGroup); let mine = await fetchPullData(supabase, token, clientGroup, null);
// facts: same rows, same JSON shapes expect(sortById(mine.facts.map((f) => f.fact!))).toEqual( sortById(naive.facts ?? []), ); expect(mine.facts).toHaveLength(4); for (let f of mine.facts) { expect(typeof f.row_version).toBe("number"); expect(f.row_version).toBeGreaterThan(0); }
// clients let naiveClients: Record<string, number> = {}; for (let c of naive.client_groups ?? []) naiveClients[c.client_id] = c.last_mutation; expect(mine.clients).toEqual(naiveClients);
// publications + contributors expect(mine.publications).toEqual(naive.publications); expect(mine.publications?.[0]?.title).toBe("Pub Title"); expect(mine.draft_contributors).toEqual(naive.draft_contributors); expect(mine.draft_contributors).toHaveLength(1); });
test("empty/nonexistent cases match", async () => { // token that exists but has no facts beyond nothing on the root let { token } = await createDoc(); let naive = await naivePullData(token, "no-such-group"); let mine = await fetchPullData(supabase, token, "no-such-group", null); expect(mine.facts.map((f) => f.fact!)).toEqual(naive.facts ?? []); expect(mine.clients).toEqual({}); expect(mine.publications).toEqual(naive.publications); expect(mine.draft_contributors).toEqual(naive.draft_contributors);
// valid-uuid but nonexistent token let ghost = randomUUID(); let naiveGhost = await naivePullData(ghost, "no-such-group"); let mineGhost = await fetchPullData(supabase, ghost, "no-such-group", null); expect(mineGhost.facts).toEqual([]); expect(naiveGhost.facts).toBeNull(); expect(mineGhost.publications).toBeNull(); expect(mineGhost.draft_contributors).toBeNull();
// malformed token id errors in both await expect(naivePullData("not-a-uuid", "g")).rejects.toThrow(); await expect(fetchPullData(supabase, "not-a-uuid", "g", null)).rejects.toThrow(); });
test("hydrates full rows only for facts that differ from the passed base", async () => { let { root, token } = await createDoc(); let factA = await addFact(root, "block/text", { type: "text", value: "a" }); let factB = await addFact(root, "block/text", { type: "text", value: "b" });
let full = await fetchPullData(supabase, token, "g", null); expect(full.facts.every((f) => f.fact !== null)).toBe(true); let base = Object.fromEntries(full.facts.map((f) => [f.id, f.row_version]));
// base matches current state exactly: versions only, no rows hydrated let unchanged = await fetchPullData(supabase, token, "g", base); expect(unchanged.facts).toHaveLength(2); expect(unchanged.facts.every((f) => f.fact === null)).toBe(true); expect( Object.fromEntries(unchanged.facts.map((f) => [f.id, f.row_version])), ).toEqual(base);
// one stale version + one unknown id: only those hydrate await pool.query(`update facts set data = $2 where id = $1`, [ factA, JSON.stringify({ type: "text", value: "a2" }), ]); let partial = await fetchPullData(supabase, token, "g", base); let partialA = partial.facts.find((f) => f.id === factA)!; let partialB = partial.facts.find((f) => f.id === factB)!; expect(partialA.fact!.data).toEqual({ type: "text", value: "a2" }); expect(partialB.fact).toBeNull();
let factC = await addFact(root, "block/text", { type: "text", value: "c" }); let withNew = await fetchPullData(supabase, token, "g", base); expect(withNew.facts.find((f) => f.id === factC)!.fact).not.toBeNull(); });
test("row_version tracks updates via xmin", async () => { let { root, token } = await createDoc(); let factA = await addFact(root, "block/text", { type: "text", value: "a", }); let factB = await addFact(root, "block/text", { type: "text", value: "b", });
let before = await fetchPullData(supabase, token, "g", null); let beforeA = before.facts.find((f) => f.id === factA)!; let beforeB = before.facts.find((f) => f.id === factB)!;
await pool.query(`update facts set data = $2 where id = $1`, [ factA, JSON.stringify({ type: "text", value: "a2" }), ]);
let after = await fetchPullData(supabase, token, "g", null); let afterA = after.facts.find((f) => f.id === factA)!; let afterB = after.facts.find((f) => f.id === factB)!;
expect(afterA.row_version).not.toBe(beforeA.row_version); expect(afterA.fact!.data).toEqual({ type: "text", value: "a2" }); expect(afterB.row_version).toBe(beforeB.row_version); });});
describe.runIf(dbAvailable)("computePull end-to-end", () => { test("first pull resets, later pulls are incremental, no-op pulls are free", async () => { let { set, root, token } = await createDoc(); let child = await addEntity(set); let factRoot = await addFact(root, "card/block", { type: "reference", value: child, }); let factChild = await addFact(child, "block/text", { type: "text", value: "original", }); let clientGroup = `cg-${randomUUID()}`; let clientID = `c-${randomUUID()}`; await setClient(clientID, clientGroup, 1);
// --- first pull: legacy-style full snapshot let kv = new Map<string, ReadonlyJSONValue>(); let first = await computePull( supabase, store, { cookie: null, clientGroupID: clientGroup }, token, Date.now(), ); if ("error" in first) throw new Error("unexpected pull error"); expect(first.patch[0]).toEqual({ op: "clear" }); expect(first.lastMutationIDChanges).toEqual({ [clientID]: 1 }); applyPatch(kv, first.patch); expect(kv.get("initialized")).toBe(true); expect(kv.has(factRoot)).toBe(true); expect(kv.has(factChild)).toBe(true); expect(parseCVRCookie(first.cookie)).not.toBeNull(); let slotKey = `replicache-cvr:${token}:${clientGroup}`; expect(JSON.parse(redisMap.get(slotKey)!).id).toBe( parseCVRCookie(first.cookie)!.cvrID, );
// --- no-op pull: nothing changed let noop = await computePull( supabase, store, { cookie: JSON.parse(JSON.stringify(first.cookie)), clientGroupID: clientGroup }, token, Date.now(), ); if ("error" in noop) throw new Error("unexpected pull error"); expect(noop.patch).toEqual([]); expect(noop.lastMutationIDChanges).toEqual({}); expect(noop.cookie).toEqual(JSON.parse(JSON.stringify(first.cookie))); // a no-op pull writes nothing: the slot still holds the first CVR expect(JSON.parse(redisMap.get(slotKey)!).id).toBe( parseCVRCookie(first.cookie)!.cvrID, );
// --- mutate: update one fact, add one, delete none, confirm a mutation await pool.query(`update facts set data = $2 where id = $1`, [ factChild, JSON.stringify({ type: "text", value: "edited" }), ]); let factNew = await addFact(child, "block/heading", { type: "number", value: 2, }); await setClient(clientID, clientGroup, 2);
let second = await computePull( supabase, store, { cookie: noop.cookie, clientGroupID: clientGroup }, token, Date.now(), ); if ("error" in second) throw new Error("unexpected pull error"); // incremental: no clear, and only the two touched facts expect(second.patch.some((op) => op.op === "clear")).toBe(false); expect( second.patch.map((op) => ({ op: op.op, key: "key" in op ? op.key : "" })), ).toEqual( expect.arrayContaining([ { op: "put", key: factChild }, { op: "put", key: factNew }, ]), ); expect(second.patch).toHaveLength(2); expect(second.lastMutationIDChanges).toEqual({ [clientID]: 2 }); applyPatch(kv, second.patch);
// --- delete a fact (which unlinks the child closure too) await pool.query(`delete from facts where id = $1`, [factRoot]); let third = await computePull( supabase, store, { cookie: second.cookie, clientGroupID: clientGroup }, token, Date.now(), ); if ("error" in third) throw new Error("unexpected pull error"); // deleting the reference fact removes the whole child subtree from the view expect(new Set(third.patch.map((op) => op.op))).toEqual(new Set(["del"])); expect(new Set(third.patch.map((op) => ("key" in op ? op.key : "")))).toEqual( new Set([factRoot, factChild, factNew]), ); applyPatch(kv, third.patch);
// final client view must equal a from-scratch snapshot let fresh = await computePull( supabase, store, { cookie: null, clientGroupID: clientGroup }, token, Date.now(), ); if ("error" in fresh) throw new Error("unexpected pull error"); let freshKv = new Map<string, ReadonlyJSONValue>(); applyPatch(freshKv, fresh.patch); expect(Object.fromEntries(kv)).toEqual(Object.fromEntries(freshKv)); });
test("publication metadata changes flow through incrementally", async () => { let { token } = await createDoc(); let first = await computePull( supabase, store, { cookie: null, clientGroupID: "g" }, token, Date.now(), ); if ("error" in first) throw new Error("unexpected pull error"); let kv = new Map<string, ReadonlyJSONValue>(); applyPatch(kv, first.patch); expect(kv.has("publication_title")).toBe(false);
await addPublicationFor(token, "First Title"); let second = await computePull( supabase, store, { cookie: first.cookie, clientGroupID: "g" }, token, Date.now(), ); if ("error" in second) throw new Error("unexpected pull error"); applyPatch(kv, second.patch); expect(kv.get("publication_title")).toBe("First Title"); expect(kv.get("publication_tags")).toEqual(["tag1", "tag2"]); expect(kv.get("post_preferences")).toEqual({ theme: "dark" }); expect(kv.get("draft_contributors")).toEqual([]);
await pool.query( `update leaflets_in_publications set title = 'Renamed' where leaflet = $1`, [token], ); let third = await computePull( supabase, store, { cookie: second.cookie, clientGroupID: "g" }, token, Date.now(), ); if ("error" in third) throw new Error("unexpected pull error"); expect(third.patch).toEqual([ { op: "put", key: "publication_title", value: "Renamed" }, ]); });
test("legacy numeric cookie falls back to a full snapshot with a larger order", async () => { let { root, token } = await createDoc(); await addFact(root, "block/text", { type: "text", value: "hi" }); let legacyCookie = Date.now() - 1000; let res = await computePull( supabase, store, { cookie: legacyCookie, clientGroupID: "g" }, token, Date.now(), ); if ("error" in res) throw new Error("unexpected pull error"); expect(res.patch[0]).toEqual({ op: "clear" }); expect((res.cookie as { order: number }).order).toBeGreaterThan( legacyCookie, ); expect(parseCVRCookie(res.cookie)).not.toBeNull(); });
test("an evicted CVR slot falls back to a full snapshot and re-converges", async () => { let { root, token } = await createDoc(); await addFact(root, "block/text", { type: "text", value: "hi" }); let first = await computePull( supabase, store, { cookie: null, clientGroupID: "g" }, token, Date.now(), ); if ("error" in first) throw new Error("unexpected pull error"); let slotKey = `replicache-cvr:${token}:g`; redisMap.delete(slotKey);
let res = await computePull( supabase, store, { cookie: first.cookie, clientGroupID: "g" }, token, Date.now(), ); if ("error" in res) throw new Error("unexpected pull error"); expect(res.patch[0]).toEqual({ op: "clear" }); // a fresh slot was written for the new cookie expect(JSON.parse(redisMap.get(slotKey)!).id).toBe( parseCVRCookie(res.cookie)!.cvrID, ); });
test("malformed token id returns ClientStateNotFound like the legacy handler", async () => { let res = await computePull( supabase, store, { cookie: null, clientGroupID: "g" }, "not-a-uuid", Date.now(), ); expect(res).toEqual({ error: "ClientStateNotFound" }); });});