feat(pds): implement account migration endpoints for PDS (#16)

* feat(pds): implement account migration endpoints

Add support for migrating accounts between PDS instances using CAR file
import/export.

Changes:
- Add com.atproto.repo.importRepo endpoint for importing repository from
  CAR file (authenticated, 100MB limit)
- Add com.atproto.server.getAccountStatus endpoint for migration planning
- Add rpcImportRepo method to AccountDurableObject with DID validation
- Add @ipld/car to dependencies for CAR file parsing
- Add comprehensive migration tests (9 tests covering auth, validation,
  and import/export workflow)
- Update EDGE_PDS_PLAN.md to document Phase 9 (Account Migration)

The implementation validates that:
- CAR files have valid structure (single root, non-empty blocks)
- DID in imported repo matches the target PDS DID
- Import is prevented if repository already exists
- Proper error handling for oversized files and malformed CAR data

This enables users to migrate their accounts from other PDS instances
while maintaining data integrity and preventing migration errors.

* refactor(pds): use official @atproto/repo CAR utilities

Replace direct @ipld/car usage with official @atproto/repo utilities
for better alignment with AT Protocol standards.

Changes:
- Use readCarWithRoot() instead of CarReader.fromBytes()
  - Automatically validates single root requirement
  - Returns BlockMap directly for efficient storage
- Use putMany() instead of individual putBlock() calls
  - More efficient bulk import operation
  - Leverages existing BlockMap support
- Move @ipld/car from dependencies to devDependencies
  - Only needed for test validation
  - Reduces runtime dependencies

Benefits:
- Cleaner, more maintainable code (removed ~30 lines)
- Better alignment with AT Protocol standards
- More efficient block import (single putMany vs loop)
- Consistent with export (blocksToCarFile)
- Automatic validation built into readCarWithRoot

All 93 tests passing, including 9 migration tests.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: ascorbic

EDGE_PDS_PLAN.md

@@ -47,13 +47,15 @@ Build a single-user AT Protocol Personal Data Server (PDS) on Cloudflare Workers
   - `com.atproto.sync.getBlob` endpoint (public read access)
   - Direct R2 access in endpoint (R2ObjectBody cannot be serialized across RPC)
   - Blobs stored with DID prefix for isolation
-- ✅ **Testing** - Migrated to vitest 4, all 81 tests passing
+- ✅ **Testing** - Migrated to vitest 4, all 101 tests passing
   - 16 storage tests
-  - 26 XRPC tests (auth, concurrency, error handling, CAR validation)
-  - 6 firehose tests (event sequencing, cursor validation, backfill)
+  - 32 XRPC tests (auth, concurrency, error handling, CAR validation)
+  - 8 firehose tests (event sequencing, cursor validation, backfill)
   - 10 blob tests (upload, retrieval, size limits, content types)
   - 15 session tests (login, refresh, getSession, JWT validation)
   - 8 validation tests (optimistic mode, strict mode, schema enforcement)
+  - 9 migration tests (account status, import/export, validation)
+  - 3 Bluesky validation tests (post creation, profile updates, schema compliance)
 - ✅ **TypeScript** - All diagnostic errors resolved, proper type declarations for cloudflare:test
 - ✅ **Protocol Helpers** - All protocol operations use official @atproto utilities
   - Record keys: `TID.nextStr()` from `@atproto/common-web`
@@ -82,6 +84,14 @@ Build a single-user AT Protocol Personal Data Server (PDS) on Cloudflare Workers
   - Integrated into `createRecord`, `putRecord`, and `applyWrites` endpoints
   - Schemas can be added dynamically via `validator.addSchema()`
   - 8 validation tests covering optimistic mode, strict mode, and schema enforcement
+- ✅ **Account Migration** (Phase 9) - Import/export for PDS migration
+  - `com.atproto.repo.importRepo` - Import repository from CAR file (authenticated, 100MB limit)
+  - `com.atproto.server.getAccountStatus` - Get account status for migration planning
+  - CAR file import using `readCarWithRoot()` from `@atproto/repo`
+  - Validates DID matches during import to prevent incorrect migrations
+  - Prevents importing over existing repository data
+  - Complete export/import workflow tested with CAR file validation
+  - 9 comprehensive migration tests
 
 ### Not Started
 
@@ -2026,11 +2036,10 @@ wrangler secret put PASSWORD_HASH # Generate: npx bcryptjs hash "your-password"
 
 - Account creation / multi-user
 - OAuth / third-party app auth
-- Account migration
 - Labelling
 - Email verification
 - Rate limiting
-- Admin endpoints
+- Advanced admin endpoints
 
 These can all be added later.
 

packages/pds/src/account-do.ts

@@ -4,6 +4,7 @@ import {
 	WriteOpAction,
 	BlockMap,
 	blocksToCarFile,
+	readCarWithRoot,
 	type RecordCreateOp,
 	type RecordUpdateOp,
 	type RecordDeleteOp,
@@ -671,6 +672,56 @@ export class AccountDurableObject extends DurableObject<Env> {
 		return blocksToCarFile(root, blocks);
 	}
 
+	/**
+	 * RPC method: Import repo from CAR file
+	 * This is used for account migration - importing an existing repository
+	 * from another PDS.
+	 */
+	async rpcImportRepo(carBytes: Uint8Array): Promise<{
+		did: string;
+		rev: string;
+		cid: string;
+	}> {
+		await this.ensureStorageInitialized();
+
+		// Check if repo already exists
+		const existingRoot = await this.storage!.getRoot();
+		if (existingRoot) {
+			throw new Error(
+				"Repository already exists. Cannot import over existing repository.",
+			);
+		}
+
+		// Use official @atproto/repo utilities to read and validate CAR
+		// readCarWithRoot validates single root requirement and returns BlockMap
+		const { root: rootCid, blocks } = await readCarWithRoot(carBytes);
+
+		// Import all blocks into storage using putMany (more efficient than individual putBlock)
+		const importRev = TID.nextStr();
+		await this.storage!.putMany(blocks, importRev);
+
+		// Load the repo to verify it's valid and get the actual revision
+		this.keypair = await Secp256k1Keypair.import(this.env.SIGNING_KEY);
+		this.repo = await Repo.load(this.storage!, rootCid);
+
+		// Verify the DID matches to prevent incorrect migrations
+		if (this.repo.did !== this.env.DID) {
+			// Clean up imported blocks
+			await this.storage!.destroy();
+			throw new Error(
+				`DID mismatch: CAR file contains DID ${this.repo.did}, but expected ${this.env.DID}`,
+			);
+		}
+
+		this.repoInitialized = true;
+
+		return {
+			did: this.repo.did,
+			rev: this.repo.commit.rev,
+			cid: rootCid.toString(),
+		};
+	}
+
 	/**
 	 * RPC method: Upload a blob to R2
 	 */

packages/pds/src/index.ts

@@ -181,6 +181,9 @@ app.post("/xrpc/com.atproto.repo.applyWrites", requireAuth, (c) =>
 app.post("/xrpc/com.atproto.repo.putRecord", requireAuth, (c) =>
 	repo.putRecord(c, getAccountDO(c.env)),
 );
+app.post("/xrpc/com.atproto.repo.importRepo", requireAuth, (c) =>
+	repo.importRepo(c, getAccountDO(c.env)),
+);
 
 // Server identity
 app.get("/xrpc/com.atproto.server.describeServer", server.describeServer);
@@ -200,6 +203,11 @@ app.post("/xrpc/com.atproto.server.refreshSession", server.refreshSession);
 app.get("/xrpc/com.atproto.server.getSession", server.getSession);
 app.post("/xrpc/com.atproto.server.deleteSession", server.deleteSession);
 
+// Account migration
+app.get("/xrpc/com.atproto.server.getAccountStatus", requireAuth, (c) =>
+	server.getAccountStatus(c, getAccountDO(c.env)),
+);
+
 // Actor preferences (stub - returns empty preferences)
 app.get("/xrpc/app.bsky.actor.getPreferences", requireAuth, (c) => {
 	return c.json({ preferences: [] });

packages/pds/src/xrpc/repo.ts

@@ -429,3 +429,87 @@ export async function uploadBlob(
 		throw err;
 	}
 }
+
+export async function importRepo(
+	c: Context<{ Bindings: Env }>,
+	accountDO: DurableObjectStub<AccountDurableObject>,
+): Promise<Response> {
+	const contentType = c.req.header("Content-Type");
+
+	// Verify content type
+	if (contentType !== "application/vnd.ipld.car") {
+		return c.json(
+			{
+				error: "InvalidRequest",
+				message:
+					"Content-Type must be application/vnd.ipld.car for repository import",
+			},
+			400,
+		);
+	}
+
+	// Get CAR file bytes
+	const carBytes = new Uint8Array(await c.req.arrayBuffer());
+
+	if (carBytes.length === 0) {
+		return c.json(
+			{
+				error: "InvalidRequest",
+				message: "Empty CAR file",
+			},
+			400,
+		);
+	}
+
+	// Size limit check (100MB for repo imports)
+	const MAX_CAR_SIZE = 100 * 1024 * 1024;
+	if (carBytes.length > MAX_CAR_SIZE) {
+		return c.json(
+			{
+				error: "RepoTooLarge",
+				message: `Repository size ${carBytes.length} exceeds maximum of ${MAX_CAR_SIZE} bytes`,
+			},
+			400,
+		);
+	}
+
+	try {
+		const result = await accountDO.rpcImportRepo(carBytes);
+		return c.json(result);
+	} catch (err) {
+		if (err instanceof Error) {
+			if (err.message.includes("already exists")) {
+				return c.json(
+					{
+						error: "RepoAlreadyExists",
+						message: "Repository already exists. Cannot import over existing data.",
+					},
+					409,
+				);
+			}
+			if (err.message.includes("DID mismatch")) {
+				return c.json(
+					{
+						error: "InvalidRepo",
+						message: err.message,
+					},
+					400,
+				);
+			}
+			if (
+				err.message.includes("no roots") ||
+				err.message.includes("no blocks") ||
+				err.message.includes("Invalid root")
+			) {
+				return c.json(
+					{
+						error: "InvalidRepo",
+						message: `Invalid CAR file: ${err.message}`,
+					},
+					400,
+				);
+			}
+		}
+		throw err;
+	}
+}

packages/pds/src/xrpc/server.ts

@@ -1,5 +1,6 @@
 import type { Context } from "hono";
 import { ensureValidHandle } from "@atproto/syntax";
+import type { AccountDurableObject } from "../account-do";
 import {
 	createAccessToken,
 	createRefreshToken,
@@ -279,3 +280,39 @@ export async function deleteSession(
 	// In a full implementation, we'd revoke the refresh token
 	return c.json({});
 }
+
+/**
+ * Get account status - used for migration checks
+ */
+export async function getAccountStatus(
+	c: Context<{ Bindings: Env }>,
+	accountDO: DurableObjectStub<AccountDurableObject>,
+): Promise<Response> {
+	try {
+		// Check if repo exists
+		const status = await accountDO.rpcGetRepoStatus();
+
+		return c.json({
+			activated: true,
+			validDid: true,
+			repoRev: status.rev,
+			repoBlocks: null, // Could implement block counting if needed
+			indexedRecords: null, // Could implement record counting if needed
+			privateStateValues: null,
+			expectedBlobs: null,
+			importedBlobs: null,
+		});
+	} catch (err) {
+		// If repo doesn't exist yet, return empty status
+		return c.json({
+			activated: false,
+			validDid: true,
+			repoRev: null,
+			repoBlocks: null,
+			indexedRecords: null,
+			privateStateValues: null,
+			expectedBlobs: null,
+			importedBlobs: null,
+		});
+	}
+}