feat(pds): implement R2 blob storage with XRPC endpoints (#5)

- Add BlobStore class for R2 blob upload/retrieval using CID-based addressing
- Implement com.atproto.repo.uploadBlob endpoint with 5MB size limit
- Implement com.atproto.sync.getBlob endpoint with direct R2 access
- Add RPC methods to AccountDurableObject for blob operations
- Export BlobStore and BlobRef types for advanced users
- Fix TypeScript errors in sequencer (RecordWriteOp import, LexValue types)
- Remove unused storage variable in createRecord method
- Add comprehensive blob storage tests (10 tests)
- Update EDGE_PDS_PLAN.md to mark Phase 5 complete
- Update CLAUDE.md with @atproto library usage guidance

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: ascorbic

CLAUDE.md

@@ -110,34 +110,41 @@ Required environment variables (validated at module load using `cloudflare:worke
 
 ### Protocol Helpers and Dependencies
 
-The codebase uses official @atproto packages for all protocol operations:
+**CRITICAL: Always use @atproto libraries instead of low-level dependencies where available.**
+
+The codebase uses official @atproto packages for all protocol operations. When implementing new features:
+
+- **Always prefer @atproto packages** over direct use of `multiformats`, `uint8arrays`, `cborg`, etc.
+- **Reference the atproto monorepo** at `~/Repos/atproto` to understand available functions and patterns
+- The @atproto packages provide stable, tested abstractions over low-level primitives
 
 **Encoding and Data Structures:**
 
-- `@atproto/lex-cbor` - CBOR encoding/decoding with `encode()` and `cidForCbor()`
+- `@atproto/lex-cbor` - CBOR encoding/decoding with `encode()`, `cidForCbor()`, `cidForRawBytes()`
 - `@atproto/lex-data` - CID operations via stable interface wrapping multiformats
 - `@atproto/repo` - Repository operations, `BlockMap`, `blocksToCarFile()`
 
 **Protocol Utilities:**
 
 - `@atproto/common-web` - `TID.nextStr()` for record key generation
 - `@atproto/syntax` - `AtUri.make()`, `ensureValidDid()`, `ensureValidHandle()`
-- `@atproto/crypto` - `Secp256k1Keypair` for signing operations
+- `@atproto/crypto` - `Secp256k1Keypair` for signing operations, `sha256()` for hashing
 - `@atproto/lexicon` - Schema validation and type definitions
 
 **Important Notes:**
 
 - Never manually construct AT URIs - use `AtUri.make(did, collection, rkey).toString()`
 - Never manually generate record keys - use `TID.nextStr()`
 - Always validate DIDs and handles using `ensureValidDid()` / `ensureValidHandle()`
+- Use `cidForRawBytes()` from `@atproto/lex-cbor` for blob CID generation
 - Use `@atproto/lex-cbor` for test fixtures instead of direct `@ipld/dag-cbor`
 - CAR file export uses `blocksToCarFile()` from `@atproto/repo`
 
 ### Vitest Configuration Notes
 
 - **Module Shimming**: Uses `resolve: { conditions: ["node", "require"] }` to force CJS builds for multiformats
 - **BlockMap/CidSet**: Access internal Map/Set via `(blocks as unknown as { map: Map<...> }).map` when iterating
-- **Test Count**: 48 tests (16 storage tests, 26 XRPC tests, 6 firehose tests)
+- **Test Count**: 58 tests (16 storage tests, 26 XRPC tests, 6 firehose tests, 10 blob tests)
 
 ### Firehose Implementation
 

EDGE_PDS_PLAN.md

@@ -12,7 +12,7 @@ Build a single-user AT Protocol Personal Data Server (PDS) on Cloudflare Workers
 
 **Live at: https://pds.mk.gg**
 
-### Completed (Phase 1 + Phase 2 + Phase 3 + Phase 4 + Phase 6 + Phase 7)
+### Completed (Phase 1 + Phase 2 + Phase 3 + Phase 4 + Phase 5 + Phase 6 + Phase 7)
 
 - ✅ **Storage Layer** (Phase 1) - `SqliteRepoStorage` implementing `@atproto/repo` RepoStorage interface
 - ✅ **Durable Object** (Phase 2) - `AccountDurableObject` with Repo integration
@@ -41,17 +41,25 @@ Build a single-user AT Protocol Personal Data Server (PDS) on Cloudflare Workers
 - ✅ **Deployment** - Custom domain `pds.mk.gg` with auto-provisioned DNS
 - ✅ **Signing Keys** - secp256k1 keypair generated and configured
 - ✅ **Environment Validation** - Module-scope validation using `cloudflare:workers` env import
-- ✅ **Testing** - Migrated to vitest 4, all 48 tests passing
+- ✅ **Blob Storage** (Phase 5) - R2 integration for image/media uploads
+  - `BlobStore` class using `cidForRawBytes()` from `@atproto/lex-cbor`
+  - `com.atproto.repo.uploadBlob` endpoint (authenticated, 5MB limit)
+  - `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 58 tests passing
   - 16 storage tests
   - 26 XRPC tests (auth, concurrency, error handling, CAR validation)
   - 6 firehose tests (event sequencing, cursor validation, backfill)
+  - 10 blob tests (upload, retrieval, size limits, content types)
 - ✅ **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`
   - AT URI construction: `AtUri.make()` from `@atproto/syntax`
   - DID validation: `ensureValidDid()` from `@atproto/syntax`
   - Handle validation: `ensureValidHandle()` from `@atproto/syntax`
   - CBOR encoding: `@atproto/lex-cbor`
+  - Blob CID generation: `cidForRawBytes()` from `@atproto/lex-cbor`
   - CAR export: `blocksToCarFile()` from `@atproto/repo`
 - ✅ **Dependency Optimization** - Removed 6 low-level dependencies, added 3 @atproto helpers
   - Removed: `varint`, `@types/varint`, `cborg`, `uint8arrays`, `@ipld/dag-cbor`, `multiformats`
@@ -60,7 +68,7 @@ Build a single-user AT Protocol Personal Data Server (PDS) on Cloudflare Workers
 
 ### Not Started
 
-- ⬜ **Blob Storage** (Phase 5) - R2 integration (R2 needs enabling in dashboard)
+- None! All planned phases are complete.
 
 ### Testing & Development Notes
 
@@ -89,6 +97,8 @@ for (const [cidStr, bytes] of internalMap) { ... }
 
 **Durable Object RPC Types**: Using `Rpc.Serializable<any>` for RPC method return types to ensure TypeScript correctly infers serializable types instead of `never`.
 
+**R2 Blob Retrieval**: The `getBlob` endpoint accesses R2 directly rather than going through DO RPC because `R2ObjectBody` cannot be serialized (contains ReadableStream). Upload operations still use DO RPC since they only need to pass Uint8Array and return serializable metadata.
+
 ---
 
 ## Architecture
@@ -1720,6 +1730,125 @@ These can all be added later.
 
 ---
 
+## Deployment Architecture
+
+### Design Decision: Zero-Code Re-Export Pattern
+
+For maximum simplicity, users deploying a PDS should not need to write any code. The `@ascorbic/pds-worker` package provides everything needed, and users simply re-export it.
+
+#### User's Worker (Minimal)
+
+```typescript
+// src/index.ts
+export { default, AccountDurableObject } from '@ascorbic/pds-worker'
+```
+
+That's it. No additional code required.
+
+#### Package Exports
+
+The `@ascorbic/pds-worker` package exports:
+
+```typescript
+// Core exports for advanced users
+export { SqliteRepoStorage } from "./storage"
+export { AccountDurableObject } from "./account-do"
+export { BlobStore, type BlobRef } from "./blobs"
+export { Sequencer } from "./sequencer"
+
+// Default export: configured Hono app
+export default app
+```
+
+#### Configuration
+
+All configuration is via environment variables and secrets:
+
+**Required environment variables:**
+- `PDS_HOSTNAME` - Public hostname (set in wrangler.jsonc)
+
+**Required secrets:**
+- `DID` - Account's DID (e.g., "did:web:pds.example.com")
+- `HANDLE` - Account's handle (e.g., "alice.pds.example.com")
+- `AUTH_TOKEN` - Bearer token for write operations
+- `SIGNING_KEY` - Private key for signing commits (secp256k1 JWK)
+- `SIGNING_KEY_PUBLIC` - Public key for DID document (multibase)
+
+**Resource bindings:**
+- `ACCOUNT` - DurableObjectNamespace binding
+- `BLOBS` - R2Bucket binding
+
+#### Deployment Workflow
+
+1. **Scaffold** (future: via `npm create @ascorbic/pds`)
+   - Creates project directory with re-export pattern
+   - Generates wrangler.jsonc with bindings
+   - Provides setup script for key generation
+
+2. **Setup** (via setup script)
+   - Interactive prompts for hostname and handle
+   - Generates secp256k1 keypair
+   - Creates DID (did:web based on hostname)
+   - Generates random AUTH_TOKEN
+   - Writes to `.dev.vars` for local dev
+
+3. **Local Development**
+   ```bash
+   wrangler dev
+   ```
+
+4. **Production Deployment**
+   ```bash
+   # Create R2 bucket
+   wrangler r2 bucket create pds-blobs
+
+   # Set secrets
+   wrangler secret put DID
+   wrangler secret put HANDLE
+   wrangler secret put AUTH_TOKEN
+   wrangler secret put SIGNING_KEY
+   wrangler secret put SIGNING_KEY_PUBLIC
+
+   # Deploy
+   wrangler deploy
+   ```
+
+#### Demo Structure
+
+```
+demos/pds/
+├── src/
+│   └── index.ts          # Re-exports @ascorbic/pds-worker
+├── wrangler.jsonc        # Worker config with bindings
+├── package.json          # Dependencies
+├── .env.example          # Template for required vars
+└── README.md            # Setup instructions
+```
+
+#### Future: create-pds CLI
+
+```bash
+npm create @ascorbic/pds my-pds
+cd my-pds
+npm install
+npm run dev
+```
+
+This will scaffold a complete deployment with:
+- Project structure
+- Generated keys and configuration
+- Pre-configured wrangler.jsonc
+- Setup instructions
+
+#### Rationale
+
+1. **Single-user PDS**: Not a multi-tenant platform - each deployment serves one account
+2. **Configuration via environment**: All customization is environment-based
+3. **No code needed**: Users shouldn't need to understand Hono/Workers/DOs to deploy
+4. **Future-proof**: Can add factory function later for customization without breaking changes
+
+---
+
 ## Reference Material
 
 - AT Protocol specs: https://atproto.com/specs

packages/pds/src/account-do.ts

@@ -15,6 +15,7 @@ import { AtUri } from "@atproto/syntax";
 import { encode as cborEncode } from "@atproto/lex-cbor";
 import { SqliteRepoStorage } from "./storage";
 import { Sequencer, type SeqEvent, type CommitData } from "./sequencer";
+import { BlobStore, type BlobRef } from "./blobs";
 
 /**
  * Account Durable Object - manages a single user's AT Protocol repository.
@@ -30,6 +31,7 @@ export class AccountDurableObject extends DurableObject<Env> {
 	private repo: Repo | null = null;
 	private keypair: Secp256k1Keypair | null = null;
 	private sequencer: Sequencer | null = null;
+	private blobStore: BlobStore | null = null;
 	private storageInitialized = false;
 	private repoInitialized = false;
 
@@ -43,6 +45,11 @@ export class AccountDurableObject extends DurableObject<Env> {
 		if (!env.DID) {
 			throw new Error("Missing required environment variable: DID");
 		}
+
+		// Initialize BlobStore if R2 bucket is available
+		if (env.BLOBS) {
+			this.blobStore = new BlobStore(env.BLOBS, env.DID);
+		}
 	}
 
 	/**
@@ -242,7 +249,6 @@ export class AccountDurableObject extends DurableObject<Env> {
 	}> {
 		const repo = await this.getRepo();
 		const keypair = await this.getKeypair();
-		const storage = await this.getStorage();
 
 		const actualRkey = rkey || TID.nextStr();
 		const createOp: RecordCreateOp = {
@@ -461,6 +467,37 @@ export class AccountDurableObject extends DurableObject<Env> {
 		return blocksToCarFile(root, blocks);
 	}
 
+	/**
+	 * RPC method: Upload a blob to R2
+	 */
+	async rpcUploadBlob(bytes: Uint8Array, mimeType: string): Promise<BlobRef> {
+		if (!this.blobStore) {
+			throw new Error("Blob storage not configured");
+		}
+
+		// Enforce size limit (5MB)
+		const MAX_BLOB_SIZE = 5 * 1024 * 1024;
+		if (bytes.length > MAX_BLOB_SIZE) {
+			throw new Error(
+				`Blob too large: ${bytes.length} bytes (max ${MAX_BLOB_SIZE})`,
+			);
+		}
+
+		return this.blobStore.putBlob(bytes, mimeType);
+	}
+
+	/**
+	 * RPC method: Get a blob from R2
+	 */
+	async rpcGetBlob(cidStr: string): Promise<R2ObjectBody | null> {
+		if (!this.blobStore) {
+			throw new Error("Blob storage not configured");
+		}
+
+		const cid = CID.parse(cidStr);
+		return this.blobStore.getBlob(cid);
+	}
+
 	/**
 	 * Encode a firehose frame (header + body CBOR).
 	 */
@@ -495,10 +532,7 @@ export class AccountDurableObject extends DurableObject<Env> {
 	/**
 	 * Backfill firehose events from a cursor.
 	 */
-	private async backfillFirehose(
-		ws: WebSocket,
-		cursor: number,
-	): Promise<void> {
+	private async backfillFirehose(ws: WebSocket, cursor: number): Promise<void> {
 		if (!this.sequencer) {
 			throw new Error("Sequencer not initialized");
 		}
@@ -594,7 +628,10 @@ export class AccountDurableObject extends DurableObject<Env> {
 	/**
 	 * WebSocket message handler (hibernation API).
 	 */
-	override webSocketMessage(_ws: WebSocket, _message: string | ArrayBuffer): void {
+	override webSocketMessage(
+		_ws: WebSocket,
+		_message: string | ArrayBuffer,
+	): void {
 		// Firehose is server-push only, ignore client messages
 	}
 

packages/pds/src/blobs.ts

@@ -0,0 +1,58 @@
+import { CID } from "@atproto/lex-data";
+import { cidForRawBytes } from "@atproto/lex-cbor";
+
+export interface BlobRef {
+	$type: "blob";
+	ref: { $link: string };
+	mimeType: string;
+	size: number;
+}
+
+/**
+ * BlobStore manages blob storage in R2.
+ * Blobs are stored with CID-based keys prefixed by the account's DID.
+ */
+export class BlobStore {
+	constructor(
+		private r2: R2Bucket,
+		private did: string,
+	) {}
+
+	/**
+	 * Upload a blob to R2 and return a BlobRef.
+	 */
+	async putBlob(bytes: Uint8Array, mimeType: string): Promise<BlobRef> {
+		// Compute CID using SHA-256 (RAW codec)
+		const cid = await cidForRawBytes(bytes);
+
+		// Store in R2 with DID prefix for isolation
+		const key = `${this.did}/${cid.toString()}`;
+		await this.r2.put(key, bytes, {
+			httpMetadata: { contentType: mimeType },
+		});
+
+		return {
+			$type: "blob",
+			ref: { $link: cid.toString() },
+			mimeType,
+			size: bytes.length,
+		};
+	}
+
+	/**
+	 * Retrieve a blob from R2 by CID.
+	 */
+	async getBlob(cid: CID): Promise<R2ObjectBody | null> {
+		const key = `${this.did}/${cid.toString()}`;
+		return this.r2.get(key);
+	}
+
+	/**
+	 * Check if a blob exists in R2.
+	 */
+	async hasBlob(cid: CID): Promise<boolean> {
+		const key = `${this.did}/${cid.toString()}`;
+		const head = await this.r2.head(key);
+		return head !== null;
+	}
+}