import { 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"
})
]);
});
});