Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428import { describe, expect, it, vi } from "vitest";import type { OAuthUserAgent } from "@atcute/oauth-browser-client";import { batchWrites, importFromLiar, missingImportScopes, MigrationScopeError} from "#lib/migration/import.ts";import { toBytes, toCidLink } from "@atcute/cbor";import { CODEC_RAW, create, fromString, toString } from "@atcute/cid";import { buildRepoCar } from "#lib/migration/repoCar.ts";import type { Nsid, Rkey } from "#lib/api/syntax.ts";import type { Fetch } from "#lib/fetch.ts";
const did = "did:plc:alice";const liarPdsUrl = "https://liar.test";const bobbinUrl = "https://bobbin.test";const unmark = "rpc:org.tangled.foster.unmarkLiarAccount?aud=*";const CID = "bafyreihltcnuuyqp2jm24aqydpnlj7b6w3ogwrplomrjtg5rifv44mmjey";const createdAt = "2026-01-01T00:00:00.000Z";
const star = (index: number) => ({ collection: "org.tangled.feed.star", rkey: `3star${String(index).padStart(8, "0")}`, value: { $type: "org.tangled.feed.star", createdAt, index }});
const liarFetchFor = ( records: Parameters<typeof buildRepoCar>[1], options?: Parameters<typeof buildRepoCar>[2]) => vi.fn<Fetch>(async (input) => { const url = new URL(input instanceof Request ? input.url : String(input)); if (url.pathname === "/xrpc/com.atproto.sync.listBlobs") { return new Response('{"cids":[]}', { headers: { "content-type": "application/json" } }); } return new Response(await buildRepoCar(did, records, options)); });
const fakePds = ({ scope = `atproto repo:org.tangled.feed.star repo:org.tangled.actor.profile ${unmark}`, aud = "https://pds.test", failApplyOnCall, rateLimit, sniffedMimeType = "application/octet-stream", lostBlobs = []}: { scope?: string; aud?: string; failApplyOnCall?: number; rateLimit?: { onCall: number; resetSeconds: number }; sniffedMimeType?: string; lostBlobs?: string[];} = {}) => { const records = new Map<string, unknown>(); let applyCalls = 0; const batchSizes: number[] = []; const validated: unknown[] = []; const json = (body: unknown, status = 200) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });
const agent = { sub: did, session: { info: { aud }, token: { scope } }, handle: async (pathname: string, init?: RequestInit) => { const url = new URL(pathname, "https://pds.test"); const body = typeof init?.body === "string" ? JSON.parse(init.body) : undefined; switch (url.pathname) { case "/xrpc/com.atproto.repo.listRecords": { const collection = url.searchParams.get("collection"); const found = [...records.keys()].filter((key) => key.startsWith(`${collection}/`) ); return json({ records: found.map((key) => ({ uri: `at://${did}/${key}`, cid: CID, value: records.get(key) })) }); } case "/xrpc/com.atproto.repo.uploadBlob": { const bytes = new Uint8Array(await new Response(init?.body).arrayBuffer()); return json({ blob: { $type: "blob", ref: { $link: toString(await create(CODEC_RAW, bytes)) }, mimeType: sniffedMimeType, size: bytes.length } }); } case "/xrpc/com.atproto.repo.applyWrites": { applyCalls++; if (applyCalls === failApplyOnCall) return json({ error: "InternalServerError" }, 500); if (applyCalls === rateLimit?.onCall) { return new Response(JSON.stringify({ error: "RateLimitExceeded" }), { status: 429, headers: { "content-type": "application/json", "ratelimit-reset": String(rateLimit.resetSeconds) } }); } const lost = lostBlobs.find((cid) => JSON.stringify(body.writes).includes(cid)); if (lost) { return json( { error: "BlobNotFound", message: `Could not find blob: ${lost}` }, 400 ); } validated.push(body.validate); batchSizes.push(body.writes.length); for (const write of body.writes) { const key = `${write.collection}/${write.rkey}`; if (write.$type.endsWith("#create") && records.has(key)) return json( { error: "InvalidRequest", message: "record already exists" }, 400 ); records.set(key, write.value); } return json({}); } } return json({ error: "MethodNotImplemented" }, 501); } }; return { agent: agent as unknown as OAuthUserAgent, records, batchSizes, validated };};
describe("importFromLiar", () => { it("writes every org.tangled record in batches and skips the rest", async () => { const pds = fakePds(); const phases: string[] = []; const fetch = liarFetchFor([ ...Array.from({ length: 250 }, (_, index) => star(index)), { collection: "org.tangled.actor.profile", rkey: "self", value: { $type: "org.tangled.actor.profile", description: "hi" } }, { collection: "app.bsky.feed.post", rkey: "3post00000000", value: { $type: "app.bsky.feed.post", text: "x", createdAt } } ]);
const result = await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch, onProgress: ({ phase }) => void (phases.at(-1) === phase || phases.push(phase)) });
expect(result).toEqual({ written: 251, skipped: 0, lostBlobs: [] }); expect(pds.batchSizes).toEqual([100, 100, 51]); expect(pds.records.get("org.tangled.actor.profile/self")).toEqual({ $type: "org.tangled.actor.profile", description: "hi" }); expect([...pds.records.keys()].some((key) => key.startsWith("app.bsky"))).toBe(false); expect(phases).toEqual(["reading", "blobs", "importing"]); const requested = new URL(String(fetch.mock.calls[0]?.[0])); expect(requested.origin + requested.pathname).toBe( `${liarPdsUrl}/xrpc/com.atproto.sync.getRepo` ); expect(requested.searchParams.get("did")).toBe(did); });
it("resumes after a failed batch without rewriting records", async () => { const pds = fakePds({ failApplyOnCall: 2 }); const records = Array.from({ length: 150 }, (_, index) => star(index)); const fetch = liarFetchFor(records);
await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).rejects.toThrow(); expect(pds.records.size).toBe(100);
pds.records.set(`org.tangled.feed.star/${records[0]!.rkey}`, { edited: true }); const result = await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch });
expect(result).toEqual({ written: 50, skipped: 100, lostBlobs: [] }); expect(pds.records.get(`org.tangled.feed.star/${records[0]!.rkey}`)).toEqual({ edited: true }); });
it("waits until the pds's rate limit resets and carries on", async () => { vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); try { const resetSeconds = Math.ceil(Date.now() / 1000) + 60; const pds = fakePds({ rateLimit: { onCall: 2, resetSeconds } }); const fetch = liarFetchFor(Array.from({ length: 150 }, (_, index) => star(index))); const waits: number[] = []; const running = importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch, onProgress: ({ resumesAt }) => void (resumesAt && waits.push(resumesAt)) });
await vi.waitFor(() => expect(waits).toEqual([resetSeconds * 1000])); await vi.advanceTimersByTimeAsync(resetSeconds * 1000 - Date.now() - 1); expect(pds.records.size).toBe(100); await vi.advanceTimersByTimeAsync(1);
expect(await running).toEqual({ written: 150, skipped: 0, lostBlobs: [] }); expect(pds.batchSizes).toEqual([100, 50]); } finally { vi.useRealTimers(); } });
it("stops before writing anything when the grant misses a collection", async () => { const pds = fakePds({ scope: "atproto repo:org.tangled.feed.star" }); const fetch = liarFetchFor([ star(0), { collection: "org.tangled.actor.profile", rkey: "self", value: { $type: "org.tangled.actor.profile" } } ]);
const failure = await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch }).catch( (cause: unknown) => cause );
expect(failure).toBeInstanceOf(MigrationScopeError); expect((failure as MigrationScopeError).missing).toEqual([ { resource: "repo", collection: "org.tangled.actor.profile", actions: ["create"] }, { resource: "rpc", lxm: "org.tangled.foster.unmarkLiarAccount", aud: "did:web:bobbin.test" } ]); expect(pds.records.size).toBe(0); });
it("doesn't ask for scopes over collections that are already on the pds", async () => { const pds = fakePds({ scope: `atproto repo:org.tangled.feed.star ${unmark}` }); const profile = { $type: "org.tangled.actor.profile", description: "hi" }; pds.records.set("org.tangled.actor.profile/self", profile); const fetch = liarFetchFor([ star(0), { collection: "org.tangled.actor.profile", rkey: "self", value: profile } ]);
await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).resolves.toEqual({ written: 1, skipped: 1, lostBlobs: [] }); });
it("fills in a profile that holds nothing but its type", async () => { const pds = fakePds(); pds.records.set("org.tangled.actor.profile/self", { $type: "org.tangled.actor.profile" }); const profile = { $type: "org.tangled.actor.profile", description: "tangled guy" }; const fetch = liarFetchFor([ { collection: "org.tangled.actor.profile", rkey: "self", value: profile } ]);
await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).resolves.toEqual({ written: 1, skipped: 0, lostBlobs: [] }); expect(pds.records.get("org.tangled.actor.profile/self")).toEqual(profile); });
it("refuses an incomplete repo instead of unmarking over missing data", async () => { const pds = fakePds(); const fetch = liarFetchFor([star(0)], { dropRecordBlocks: true });
await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).rejects.toThrow( "incomplete" ); expect(pds.records.size).toBe(0); });
it("points blob refs at the type the pds sniffed from the upload", async () => { const script = new TextEncoder().encode("#!/usr/bin/env bash\necho hi\n"); const blobCid = toString(await create(CODEC_RAW, script)); const pds = fakePds({ scope: `atproto repo:org.tangled.snippet.string ${unmark}`, sniffedMimeType: "text/x-shellscript" }); const snippet = { $type: "org.tangled.snippet.string", content: { $type: "blob", ref: toCidLink(fromString(blobCid)), mimeType: "text/plain", size: script.length }, digest: toBytes(new Uint8Array([1, 2, 3])) }; const car = await buildRepoCar(did, [ { collection: "org.tangled.snippet.string", rkey: "3snippet00000", value: snippet } ]); const fetch = vi.fn<Fetch>(async (input) => { const url = new URL(input instanceof Request ? input.url : String(input)); switch (url.pathname) { case "/xrpc/com.atproto.sync.listBlobs": return Response.json({ cids: [blobCid] }); case "/xrpc/com.atproto.sync.getBlob": return new Response(script, { headers: { "content-type": "text/plain" } }); } return new Response(car); });
await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch });
const written = JSON.parse( JSON.stringify(pds.records.get("org.tangled.snippet.string/3snippet00000")) ); expect(written.content).toEqual({ $type: "blob", ref: { $link: blobCid }, mimeType: "text/x-shellscript", size: script.length }); expect(written.digest).toEqual({ $bytes: "AQID" }); });
it("migrates a record without a blob that no pds has and says which", async () => { const gone = "bafkreidfir3gpv7hnpzzilr2fenaqswqqd2bazegq3nx5ix4vpfo4xbooi"; const pds = fakePds({ lostBlobs: [gone] }); const profile = { $type: "org.tangled.actor.profile", description: "hi", avatar: { $type: "blob", ref: toCidLink(fromString(gone)), mimeType: "image/jpeg", size: 10 } }; const fetch = liarFetchFor([ star(0), { collection: "org.tangled.actor.profile", rkey: "self", value: profile } ]);
const result = await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch });
expect(result.lostBlobs).toEqual([ { collection: "org.tangled.actor.profile", rkey: "self", field: "avatar" } ]); expect(pds.records.get("org.tangled.actor.profile/self")).toEqual({ $type: "org.tangled.actor.profile", description: "hi" }); expect(pds.records.has(`org.tangled.feed.star/${star(0).rkey}`)).toBe(true); });
it("asks the pds not to validate records it may not know", async () => { const pds = fakePds(); await importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch: liarFetchFor([star(0)]) }); expect(pds.validated).toEqual([false]); });
it("refuses to import into the liar pds itself", async () => { const pds = fakePds({ aud: `${liarPdsUrl}/` }); const fetch = liarFetchFor([star(0)]);
await expect(importFromLiar(pds.agent, { liarPdsUrl, bobbinUrl, fetch })).rejects.toThrow( "nowhere to move" ); expect(fetch).not.toHaveBeenCalled(); });});
describe("batchWrites", () => { const write = (size: number) => ({ $type: "com.atproto.repo.applyWrites#create" as const, collection: "org.tangled.feed.star" as Nsid, rkey: "3aaaaaaaaaaaa" as Rkey, value: { $type: "org.tangled.feed.star", text: "x".repeat(size) } });
it("closes a batch at the byte budget and lets one oversized record through alone", () => { const batches = batchWrites([ write(40_000), write(40_000), write(40_000), write(200_000), write(10) ]); expect(batches.map((batch) => batch.length)).toEqual([2, 1, 1, 1]); });
it("returns nothing for nothing", () => { expect(batchWrites([])).toEqual([]); });});
describe("missingImportScopes", () => { it("always asks for the unmark rpc too", () => { expect( missingImportScopes( "atproto", [ { $type: "com.atproto.repo.applyWrites#create", collection: "org.tangled.feed.star" as Nsid } ], bobbinUrl ) ).toEqual([ expect.objectContaining({ resource: "repo", collection: "org.tangled.feed.star" }), expect.objectContaining({ resource: "rpc", lxm: "org.tangled.foster.unmarkLiarAccount" }) ]); });});