chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
# PDS Core Separation Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** Cleanly separate PDS business logic from Cloudflare Workers to enable running on Express/Node with SQLite, Postgres, filesystem blobs, etc.
**Architecture:** Hexagonal architecture (ports & adapters) with constructor injection. Core defines `StoragePort` and `BlobPort` interfaces. Platform adapters (Cloudflare, Node) wire concrete implementations. No DI framework - simple constructor injection.
**Tech Stack:** JavaScript (ES modules), Web Standards (Request/Response), better-sqlite3 (Node), Cloudflare D1/R2, pnpm workspaces
---
## Phase 1: Setup Monorepo Structure
### Task 1.1: Initialize pnpm workspace
**Files:**
- Create: `pnpm-workspace.yaml`
- Modify: `package.json`
**Step 1: Create workspace config**
```yaml
# pnpm-workspace.yaml
packages:
- 'packages/*'
```
**Step 2: Update root package.json**
Add to `package.json`:
```json
{
"private": true,
"workspaces": ["packages/*"]
}
```
**Step 3: Create packages directory**
Run: `mkdir -p packages`
**Step 4: Commit**
```bash
git add pnpm-workspace.yaml package.json
git commit -m "chore: setup pnpm workspace"
```
---
### Task 1.2: Create @pdsjs/core package skeleton
**Files:**
- Create: `packages/core/package.json`
- Create: `packages/core/index.js`
**Step 1: Create package.json**
```json
{
"name": "@pdsjs/core",
"version": "0.1.0",
"type": "module",
"main": "index.js",
"exports": {
".": "./index.js",
"./repo": "./repo.js",
"./auth": "./auth.js",
"./crypto": "./crypto.js"
}
}
```
**Step 2: Create empty index.js**
```javascript
// @pdsjs/core - PDS business logic
// Exports will be added as code is extracted
export {}
```
**Step 3: Commit**
```bash
git add packages/core/
git commit -m "chore: create @pdsjs/core package skeleton"
```
---
## Phase 2: Extract Pure Utilities to Core
### Task 2.1: Extract crypto utilities
**Files:**
- Create: `packages/core/crypto.js`
- Modify: `src/pds.js`
**Step 1: Run existing tests to establish baseline**
Run: `npm test`
Expected: All tests pass
**Step 2: Create crypto.js with extracted functions**
Copy these functions from `src/pds.js` to `packages/core/crypto.js`:
- `bytesToHex`
- `hexToBytes`
- `base32Encode`
- `base32Decode`
- `base64UrlEncode`
- `base64UrlDecode`
- `generateKeyPair`
- `importPrivateKey`
- `sign`
- `getPublicKeyFromPrivate`
- `compressPublicKey`
- `computeJwkThumbprint`
```javascript
// packages/core/crypto.js
/** P-256 curve order for low-S normalization */
const P256_ORDER = 0xffffffff00000000ffffffffffffffffbce6faada7179e84f3b9cac2fc632551n
/** Base32 alphabet (RFC 4648 lowercase) */
const BASE32_ALPHABET = 'abcdefghijklmnopqrstuvwxyz234567'
/**
* Convert bytes to hex string
* @param {Uint8Array} bytes
* @returns {string}
*/
export function bytesToHex(bytes) {
return Array.from(bytes, (b) => b.toString(16).padStart(2, '0')).join('')
}
/**
* Convert hex string to bytes
* @param {string} hex
* @returns {Uint8Array}
*/
export function hexToBytes(hex) {
const bytes = new Uint8Array(hex.length / 2)
for (let i = 0; i < bytes.length; i++) {
bytes[i] = Number.parseInt(hex.substr(i * 2, 2), 16)
}
return bytes
}
/**
* Encode bytes to base32 (lowercase, no padding)
* @param {Uint8Array} bytes
* @returns {string}
*/
export function base32Encode(bytes) {
let bits = 0
let value = 0
let output = ''
for (const byte of bytes) {
value = (value << 8) | byte
bits += 8
while (bits >= 5) {
output += BASE32_ALPHABET[(value >>> (bits - 5)) & 31]
bits -= 5
}
}
if (bits > 0) {
output += BASE32_ALPHABET[(value << (5 - bits)) & 31]
}
return output
}
/**
* Decode base32 string to bytes
* @param {string} str
* @returns {Uint8Array}
*/
export function base32Decode(str) {
str = str.toLowerCase()
const output = []
let bits = 0
let value = 0
for (const char of str) {
const idx = BASE32_ALPHABET.indexOf(char)
if (idx === -1) continue
value = (value << 5) | idx
bits += 5
if (bits >= 8) {
output.push((value >>> (bits - 8)) & 255)
bits -= 8
}
}
return new Uint8Array(output)
}
/**
* Encode bytes to base64url (no padding)
* @param {Uint8Array} bytes
* @returns {string}
*/
export function base64UrlEncode(bytes) {
const binString = Array.from(bytes, (byte) => String.fromCodePoint(byte)).join('')
return btoa(binString).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
}
/**
* Decode base64url string to bytes
* @param {string} str
* @returns {Uint8Array}
*/
export function base64UrlDecode(str) {
const base64 = str.replace(/-/g, '+').replace(/_/g, '/')
const binString = atob(base64)
return Uint8Array.from(binString, (c) => c.codePointAt(0))
}
/**
* Generate P-256 key pair
* @returns {Promise<{privateKey: Uint8Array, publicKey: Uint8Array}>}
*/
export async function generateKeyPair() {
const keyPair = await crypto.subtle.generateKey({ name: 'ECDSA', namedCurve: 'P-256' }, true, ['sign'])
const privateJwk = await crypto.subtle.exportKey('jwk', keyPair.privateKey)
const privateKey = base64UrlDecode(privateJwk.d)
const publicKey = compressPublicKey(base64UrlDecode(privateJwk.x), base64UrlDecode(privateJwk.y))
return { privateKey, publicKey }
}
/**
* Import raw P-256 private key bytes
* @param {Uint8Array} privateKeyBytes
* @returns {Promise<CryptoKey>}
*/
export async function importPrivateKey(privateKeyBytes) {
if (!(privateKeyBytes instanceof Uint8Array)) {
throw new Error('Invalid private key: expected Uint8Array')
}
if (privateKeyBytes.length !== 32) {
throw new Error(`Invalid private key length: expected 32 bytes, got ${privateKeyBytes.length}`)
}
const publicKey = await getPublicKeyFromPrivate(privateKeyBytes)
const x = publicKey.slice(1, 33)
const y = publicKey.slice(33)
const jwk = {
kty: 'EC',
crv: 'P-256',
x: base64UrlEncode(x),
y: base64UrlEncode(y),
d: base64UrlEncode(privateKeyBytes),
}
return crypto.subtle.importKey('jwk', jwk, { name: 'ECDSA', namedCurve: 'P-256' }, false, ['sign'])
}
/**
* Sign data with P-256 key (low-S normalized)
* @param {CryptoKey} key
* @param {Uint8Array} data
* @returns {Promise<Uint8Array>}
*/
export async function sign(key, data) {
const sig = await crypto.subtle.sign({ name: 'ECDSA', hash: 'SHA-256' }, key, data)
const sigBytes = new Uint8Array(sig)
let s = 0n
for (let i = 32; i < 64; i++) {
s = (s << 8n) | BigInt(sigBytes[i])
}
const halfOrder = P256_ORDER >> 1n
if (s > halfOrder) {
s = P256_ORDER - s
for (let i = 63; i >= 32; i--) {
sigBytes[i] = Number(s & 0xffn)
s >>= 8n
}
}
return sigBytes
}
/**
* Derive uncompressed public key from private key
* @param {Uint8Array} privateKey
* @returns {Promise<Uint8Array>}
*/
export async function getPublicKeyFromPrivate(privateKey) {
const keyPair = await crypto.subtle.generateKey({ name: 'ECDSA', namedCurve: 'P-256' }, true, ['sign'])
const privateJwk = await crypto.subtle.exportKey('jwk', keyPair.privateKey)
privateJwk.d = base64UrlEncode(privateKey)
const importedKey = await crypto.subtle.importKey('jwk', privateJwk, { name: 'ECDSA', namedCurve: 'P-256' }, true, ['sign'])
const publicJwk = await crypto.subtle.exportKey('jwk', importedKey)
const x = base64UrlDecode(publicJwk.x)
const y = base64UrlDecode(publicJwk.y)
const uncompressed = new Uint8Array(65)
uncompressed[0] = 0x04
uncompressed.set(x, 1)
uncompressed.set(y, 33)
return uncompressed
}
/**
* Compress public key (SEC1 format)
* @param {Uint8Array} x
* @param {Uint8Array} y
* @returns {Uint8Array}
*/
export function compressPublicKey(x, y) {
const prefix = (y[y.length - 1] & 1) === 0 ? 0x02 : 0x03
const compressed = new Uint8Array(33)
compressed[0] = prefix
compressed.set(x, 1)
return compressed
}
/**
* Compute JWK thumbprint (RFC 7638)
* @param {Object} jwk
* @returns {Promise<string>}
*/
export async function computeJwkThumbprint(jwk) {
const normalized = JSON.stringify({ crv: jwk.crv, kty: jwk.kty, x: jwk.x, y: jwk.y })
const hash = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(normalized))
return base64UrlEncode(new Uint8Array(hash))
}
```
**Step 3: Update src/pds.js to import from core**
At the top of `src/pds.js`, add:
```javascript
import {
bytesToHex,
hexToBytes,
base32Encode,
base32Decode,
base64UrlEncode,
base64UrlDecode,
generateKeyPair,
importPrivateKey,
sign,
getPublicKeyFromPrivate,
compressPublicKey,
computeJwkThumbprint,
} from '../packages/core/crypto.js'
```
Remove the local implementations of these functions from `src/pds.js`.
**Step 4: Run tests to verify extraction**
Run: `npm test`
Expected: All tests pass
**Step 5: Commit**
```bash
git add packages/core/crypto.js src/pds.js
git commit -m "refactor: extract crypto utilities to @pdsjs/core"
```
---
### Task 2.2: Extract CBOR/repo utilities
**Files:**
- Create: `packages/core/repo.js`
- Modify: `src/pds.js`
**Step 1: Create repo.js with extracted functions**
Copy these functions from `src/pds.js` to `packages/core/repo.js`:
- `varint`
- `encodeHead`
- `cborEncode`
- `cborDecode`
- `sha256`
- `createCid`
- `createBlobCid`
- `cidToString`
- `cidFromString`
- `createTid`
- `getKeyDepth`
- `buildCarFile`
- `sniffMimeType`
- `findBlobRefs`
```javascript
// packages/core/repo.js
import { base32Encode, base32Decode } from './crypto.js'
/** CBOR codec for DAG-CBOR */
const DAGCBOR_CODEC = 0x71
/** Raw codec for blobs */
const RAW_CODEC = 0x55
/** SHA-256 multihash code */
const SHA256_CODE = 0x12
/** Base32-sort alphabet for TIDs */
const BASE32_SORT = '234567abcdefghijklmnopqrstuvwxyz'
/** Monotonic TID state */
let lastTimestamp = 0
let clockId = Math.floor(Math.random() * 1024)
/**
* Encode unsigned varint
* @param {number} n
* @returns {Uint8Array}
*/
export function varint(n) {
const bytes = []
while (n >= 0x80) {
bytes.push((n & 0x7f) | 0x80)
n >>>= 7
}
bytes.push(n)
return new Uint8Array(bytes)
}
/**
* Encode CBOR header byte
* @param {number} majorType
* @param {number} value
* @returns {Uint8Array}
*/
export function encodeHead(majorType, value) {
const major = majorType << 5
if (value < 24) return new Uint8Array([major | value])
if (value < 256) return new Uint8Array([major | 24, value])
if (value < 65536) return new Uint8Array([major | 25, value >> 8, value & 0xff])
if (value < 4294967296) {
return new Uint8Array([major | 26, (value >>> 24) & 0xff, (value >>> 16) & 0xff, (value >>> 8) & 0xff, value & 0xff])
}
throw new Error('Value too large for CBOR encoding')
}
/**
* Encode value as DAG-CBOR
* @param {*} value
* @returns {Uint8Array}
*/
export function cborEncode(value) {
if (value === null) return new Uint8Array([0xf6])
if (value === true) return new Uint8Array([0xf5])
if (value === false) return new Uint8Array([0xf4])
if (typeof value === 'number') {
if (Number.isInteger(value)) {
if (value >= 0) {
return encodeHead(0, value)
}
return encodeHead(1, -1 - value)
}
const buf = new ArrayBuffer(9)
const view = new DataView(buf)
view.setUint8(0, 0xfb)
view.setFloat64(1, value)
return new Uint8Array(buf)
}
if (typeof value === 'string') {
const bytes = new TextEncoder().encode(value)
return new Uint8Array([...encodeHead(3, bytes.length), ...bytes])
}
if (value instanceof Uint8Array) {
return new Uint8Array([...encodeHead(2, value.length), ...value])
}
if (Array.isArray(value)) {
const items = value.map(cborEncode)
const totalLen = items.reduce((sum, item) => sum + item.length, 0)
const result = new Uint8Array(encodeHead(4, value.length).length + totalLen)
let offset = 0
const header = encodeHead(4, value.length)
result.set(header, offset)
offset += header.length
for (const item of items) {
result.set(item, offset)
offset += item.length
}
return result
}
if (typeof value === 'object') {
// Handle CID links
if (value.$link) {
const cidBytes = cidFromString(value.$link)
const tagged = new Uint8Array([0xd8, 0x2a, ...encodeHead(2, cidBytes.length), ...cidBytes])
return tagged
}
// Sort keys for deterministic encoding
const keys = Object.keys(value).sort((a, b) => {
const aBytes = new TextEncoder().encode(a)
const bBytes = new TextEncoder().encode(b)
const minLen = Math.min(aBytes.length, bBytes.length)
for (let i = 0; i < minLen; i++) {
if (aBytes[i] !== bBytes[i]) return aBytes[i] - bBytes[i]
}
return aBytes.length - bBytes.length
})
const pairs = []
for (const key of keys) {
pairs.push(cborEncode(key), cborEncode(value[key]))
}
const totalLen = pairs.reduce((sum, item) => sum + item.length, 0)
const result = new Uint8Array(encodeHead(5, keys.length).length + totalLen)
let offset = 0
const header = encodeHead(5, keys.length)
result.set(header, offset)
offset += header.length
for (const pair of pairs) {
result.set(pair, offset)
offset += pair.length
}
return result
}
throw new Error(`Cannot encode ${typeof value}`)
}
/**
* Decode DAG-CBOR bytes
* @param {Uint8Array} bytes
* @returns {*}
*/
export function cborDecode(bytes) {
let offset = 0
function read() {
const initial = bytes[offset++]
const major = initial >> 5
const additional = initial & 0x1f
let value
if (additional < 24) {
value = additional
} else if (additional === 24) {
value = bytes[offset++]
} else if (additional === 25) {
value = (bytes[offset++] << 8) | bytes[offset++]
} else if (additional === 26) {
value = (bytes[offset++] * 16777216) + (bytes[offset++] << 16) | (bytes[offset++] << 8) | bytes[offset++]
} else if (additional === 27) {
value = 0
for (let i = 0; i < 8; i++) {
value = value * 256 + bytes[offset++]
}
}
switch (major) {
case 0: return value
case 1: return -1 - value
case 2: {
const slice = bytes.slice(offset, offset + value)
offset += value
return slice
}
case 3: {
const slice = bytes.slice(offset, offset + value)
offset += value
return new TextDecoder().decode(slice)
}
case 4: {
const arr = []
for (let i = 0; i < value; i++) arr.push(read())
return arr
}
case 5: {
const obj = {}
for (let i = 0; i < value; i++) {
const key = read()
obj[key] = read()
}
return obj
}
case 6: {
if (value === 42) {
const cidBytes = read()
return { $link: cidToString(cidBytes) }
}
return read()
}
case 7: {
if (additional === 20) return false
if (additional === 21) return true
if (additional === 22) return null
if (additional === 27) {
const buf = new ArrayBuffer(8)
const view = new DataView(buf)
for (let i = 0; i < 8; i++) {
view.setUint8(i, bytes[offset++])
}
return view.getFloat64(0)
}
}
}
}
return read()
}
/**
* Compute SHA-256 hash
* @param {Uint8Array} data
* @returns {Promise<Uint8Array>}
*/
export async function sha256(data) {
const hash = await crypto.subtle.digest('SHA-256', data)
return new Uint8Array(hash)
}
/**
* Create CID for DAG-CBOR data
* @param {Uint8Array} data
* @returns {Promise<Uint8Array>}
*/
export async function createCid(data) {
const hash = await sha256(data)
const cid = new Uint8Array(36)
cid[0] = 0x01 // CIDv1
cid[1] = DAGCBOR_CODEC
cid[2] = SHA256_CODE
cid[3] = 0x20 // 32 bytes
cid.set(hash, 4)
return cid
}
/**
* Create CID for raw blob data
* @param {Uint8Array} data
* @returns {Promise<Uint8Array>}
*/
export async function createBlobCid(data) {
const hash = await sha256(data)
const cid = new Uint8Array(36)
cid[0] = 0x01
cid[1] = RAW_CODEC
cid[2] = SHA256_CODE
cid[3] = 0x20
cid.set(hash, 4)
return cid
}
/**
* Convert CID bytes to base32 string
* @param {Uint8Array} cid
* @returns {string}
*/
export function cidToString(cid) {
return 'b' + base32Encode(cid)
}
/**
* Convert base32 CID string to bytes
* @param {string} str
* @returns {Uint8Array}
*/
export function cidFromString(str) {
if (str[0] !== 'b') throw new Error('Invalid CID: must start with b')
return base32Decode(str.slice(1))
}
/**
* Create monotonic TID
* @returns {string}
*/
export function createTid() {
let timestamp = Date.now() * 1000
if (timestamp <= lastTimestamp) {
timestamp = lastTimestamp + 1
}
lastTimestamp = timestamp
clockId = (clockId + 1) & 0x3ff
const combined = (BigInt(timestamp) << 10n) | BigInt(clockId)
let result = ''
let val = combined
for (let i = 0; i < 13; i++) {
result = BASE32_SORT[Number(val & 31n)] + result
val >>= 5n
}
return result
}
/**
* Compute MST key depth from SHA-256 leading zeros
* @param {string} key
* @returns {Promise<number>}
*/
export async function getKeyDepth(key) {
const hash = await sha256(new TextEncoder().encode(key))
let depth = 0
for (const byte of hash) {
if (byte === 0) {
depth += 8
} else {
let b = byte
while ((b & 0x80) === 0) {
depth++
b <<= 1
}
break
}
}
return depth
}
/**
* Build CAR file from root CID and blocks
* @param {string} rootCid
* @param {Array<{cid: string, data: Uint8Array}>} blocks
* @returns {Uint8Array}
*/
export function buildCarFile(rootCid, blocks) {
const rootBytes = cidFromString(rootCid)
const header = cborEncode({ version: 1, roots: [{ $link: rootCid }] })
const headerLen = varint(header.length)
let totalSize = headerLen.length + header.length
const encodedBlocks = blocks.map(({ cid, data }) => {
const cidBytes = cidFromString(cid)
const blockLen = varint(cidBytes.length + data.length)
totalSize += blockLen.length + cidBytes.length + data.length
return { cidBytes, data, blockLen }
})
const car = new Uint8Array(totalSize)
let offset = 0
car.set(headerLen, offset)
offset += headerLen.length
car.set(header, offset)
offset += header.length
for (const { cidBytes, data, blockLen } of encodedBlocks) {
car.set(blockLen, offset)
offset += blockLen.length
car.set(cidBytes, offset)
offset += cidBytes.length
car.set(data, offset)
offset += data.length
}
return car
}
/**
* Detect MIME type from magic bytes
* @param {Uint8Array} bytes
* @returns {string|null}
*/
export function sniffMimeType(bytes) {
if (bytes[0] === 0xff && bytes[1] === 0xd8 && bytes[2] === 0xff) return 'image/jpeg'
if (bytes[0] === 0x89 && bytes[1] === 0x50 && bytes[2] === 0x4e && bytes[3] === 0x47) return 'image/png'
if (bytes[0] === 0x47 && bytes[1] === 0x49 && bytes[2] === 0x46) return 'image/gif'
if (bytes[0] === 0x52 && bytes[1] === 0x49 && bytes[2] === 0x46 && bytes[3] === 0x46 &&
bytes[8] === 0x57 && bytes[9] === 0x45 && bytes[10] === 0x42 && bytes[11] === 0x50) return 'image/webp'
if (bytes[4] === 0x66 && bytes[5] === 0x74 && bytes[6] === 0x79 && bytes[7] === 0x70) {
const brand = String.fromCharCode(bytes[8], bytes[9], bytes[10], bytes[11])
if (brand === 'avif') return 'image/avif'
if (brand === 'heic' || brand === 'heix' || brand === 'mif1') return 'image/heic'
return 'video/mp4'
}
return null
}
/**
* Find all blob CID references in a record
* @param {*} value
* @returns {string[]}
*/
export function findBlobRefs(value) {
const refs = []
function walk(v) {
if (v === null || typeof v !== 'object') return
if (v.$type === 'blob' && v.ref?.$link) {
refs.push(v.ref.$link)
return
}
if (Array.isArray(v)) {
for (const item of v) walk(item)
} else {
for (const val of Object.values(v)) walk(val)
}
}
walk(value)
return refs
}
```
**Step 2: Update src/pds.js to import from core**
Add to imports:
```javascript
import {
varint,
encodeHead,
cborEncode,
cborDecode,
sha256,
createCid,
createBlobCid,
cidToString,
cidFromString,
createTid,
getKeyDepth,
buildCarFile,
sniffMimeType,
findBlobRefs,
} from '../packages/core/repo.js'
```
Remove local implementations.
**Step 3: Run tests to verify extraction**
Run: `npm test`
Expected: All tests pass
**Step 4: Commit**
```bash
git add packages/core/repo.js src/pds.js
git commit -m "refactor: extract repo utilities to @pdsjs/core"
```
---
### Task 2.3: Extract auth utilities
**Files:**
- Create: `packages/core/auth.js`
- Modify: `src/pds.js`
**Step 1: Create auth.js with extracted functions**
Copy these functions from `src/pds.js`:
- `createAccessJwt`
- `createRefreshJwt`
- `verifyAccessJwt`
- `verifyRefreshJwt`
- `isLoopbackClient`
- `getLoopbackClientMetadata`
- `validateClientMetadata`
- `parseAtprotoProxyHeader`
- `getKnownServiceUrl`
- `parseRepoScope`
- `parseBlobScope`
- `matchesMime`
- `parseScopesForDisplay`
- `ScopePermissions` class
```javascript
// packages/core/auth.js
import { base64UrlEncode, base64UrlDecode } from './crypto.js'
const BSKY_APPVIEW_URL = 'https://api.bsky.app'
/**
* Create access JWT
* @param {string} did
* @param {string} secret
* @param {number} expiresInSeconds
* @returns {Promise<string>}
*/
export async function createAccessJwt(did, secret, expiresInSeconds = 3600) {
const header = { typ: 'at+jwt', alg: 'HS256' }
const now = Math.floor(Date.now() / 1000)
const payload = {
scope: 'com.atproto.access',
sub: did,
aud: did,
iat: now,
exp: now + expiresInSeconds,
}
return signJwt(header, payload, secret)
}
/**
* Create refresh JWT
* @param {string} did
* @param {string} secret
* @param {number} expiresInSeconds
* @returns {Promise<string>}
*/
export async function createRefreshJwt(did, secret, expiresInSeconds = 7776000) {
const header = { typ: 'refresh+jwt', alg: 'HS256' }
const now = Math.floor(Date.now() / 1000)
const jti = crypto.randomUUID()
const payload = {
scope: 'com.atproto.refresh',
sub: did,
aud: did,
iat: now,
exp: now + expiresInSeconds,
jti,
}
return signJwt(header, payload, secret)
}
async function signJwt(header, payload, secret) {
const enc = new TextEncoder()
const headerB64 = base64UrlEncode(enc.encode(JSON.stringify(header)))
const payloadB64 = base64UrlEncode(enc.encode(JSON.stringify(payload)))
const data = enc.encode(`${headerB64}.${payloadB64}`)
const key = await crypto.subtle.importKey('raw', enc.encode(secret), { name: 'HMAC', hash: 'SHA-256' }, false, ['sign'])
const sig = await crypto.subtle.sign('HMAC', key, data)
const sigB64 = base64UrlEncode(new Uint8Array(sig))
return `${headerB64}.${payloadB64}.${sigB64}`
}
/**
* Verify access JWT
* @param {string} jwt
* @param {string} secret
* @returns {Promise<Object>}
*/
export async function verifyAccessJwt(jwt, secret) {
const payload = await verifyJwt(jwt, secret)
if (payload.scope !== 'com.atproto.access') {
throw new Error('Invalid token type: expected access token')
}
return payload
}
/**
* Verify refresh JWT
* @param {string} jwt
* @param {string} secret
* @returns {Promise<Object>}
*/
export async function verifyRefreshJwt(jwt, secret) {
const payload = await verifyJwt(jwt, secret)
if (payload.scope !== 'com.atproto.refresh') {
throw new Error('Invalid token type: expected refresh token')
}
return payload
}
async function verifyJwt(jwt, secret) {
const parts = jwt.split('.')
if (parts.length !== 3) throw new Error('Invalid JWT format')
const [headerB64, payloadB64, sigB64] = parts
const enc = new TextEncoder()
const data = enc.encode(`${headerB64}.${payloadB64}`)
const sig = base64UrlDecode(sigB64)
const key = await crypto.subtle.importKey('raw', enc.encode(secret), { name: 'HMAC', hash: 'SHA-256' }, false, ['verify'])
const valid = await crypto.subtle.verify('HMAC', key, sig, data)
if (!valid) throw new Error('Invalid signature')
const payload = JSON.parse(new TextDecoder().decode(base64UrlDecode(payloadB64)))
const now = Math.floor(Date.now() / 1000)
if (payload.exp && payload.exp < now) throw new Error('Token expired')
return payload
}
/**
* Check if client ID is loopback
* @param {string} clientId
* @returns {boolean}
*/
export function isLoopbackClient(clientId) {
try {
const url = new URL(clientId)
return url.hostname === 'localhost' || url.hostname === '127.0.0.1'
} catch {
return false
}
}
/**
* Get permissive metadata for loopback clients
* @param {string} clientId
* @returns {Object}
*/
export function getLoopbackClientMetadata(clientId) {
return {
client_id: clientId,
client_name: 'Loopback Client',
redirect_uris: [`${clientId}/callback`, `${new URL(clientId).origin}/callback`],
grant_types: ['authorization_code', 'refresh_token'],
response_types: ['code'],
scope: 'atproto transition:generic',
token_endpoint_auth_method: 'none',
application_type: 'native',
dpop_bound_access_tokens: true,
}
}
/**
* Validate OAuth client metadata
* @param {Object} metadata
* @param {string} expectedClientId
*/
export function validateClientMetadata(metadata, expectedClientId) {
if (metadata.client_id !== expectedClientId) {
throw new Error(`client_id mismatch: expected ${expectedClientId}, got ${metadata.client_id}`)
}
}
/**
* Parse atproto-proxy header
* @param {string|null} header
* @returns {{did: string, serviceId: string}|null}
*/
export function parseAtprotoProxyHeader(header) {
if (!header) return null
const hashIdx = header.indexOf('#')
if (hashIdx === -1 || hashIdx === 0 || hashIdx === header.length - 1) return null
return {
did: header.slice(0, hashIdx),
serviceId: header.slice(hashIdx + 1),
}
}
/**
* Get URL for known service
* @param {string} did
* @param {string} serviceId
* @returns {string|null}
*/
export function getKnownServiceUrl(did, serviceId) {
if (did === 'did:web:api.bsky.app' && serviceId === 'bsky_appview') {
return BSKY_APPVIEW_URL
}
return null
}
/**
* Parse repo scope string
* @param {string} scope
* @returns {{collection: string, actions: string[]}|null}
*/
export function parseRepoScope(scope) {
if (!scope.startsWith('repo')) return null
let collection = '*'
let actions = []
if (scope.startsWith('repo:')) {
const rest = scope.slice(5)
const qIdx = rest.indexOf('?')
if (qIdx === -1) {
collection = rest
actions = ['create', 'update', 'delete']
} else {
collection = rest.slice(0, qIdx)
const params = new URLSearchParams(rest.slice(qIdx + 1))
actions = [...new Set(params.getAll('action'))]
}
} else if (scope.startsWith('repo?')) {
const params = new URLSearchParams(scope.slice(5))
collection = params.get('collection') || '*'
actions = [...new Set(params.getAll('action'))]
} else {
return null
}
if (!collection || actions.length === 0) return null
return { collection, actions }
}
/**
* Parse blob scope string
* @param {string} scope
* @returns {{accept: string[]}|null}
*/
export function parseBlobScope(scope) {
if (!scope.startsWith('blob:')) return null
const mimes = scope.slice(5).split(',')
return { accept: mimes }
}
/**
* Check if MIME type matches pattern
* @param {string} pattern
* @param {string} mime
* @returns {boolean}
*/
export function matchesMime(pattern, mime) {
pattern = pattern.toLowerCase()
mime = mime.toLowerCase()
if (pattern === '*/*') return true
if (pattern.endsWith('/*')) {
const type = pattern.slice(0, -2)
return mime.startsWith(type + '/')
}
return pattern === mime
}
/**
* Parse scopes for display on consent page
* @param {string} scopeString
* @returns {Object}
*/
export function parseScopesForDisplay(scopeString) {
const scopes = (scopeString || '').split(' ').filter(Boolean)
const result = {
hasAtproto: scopes.includes('atproto'),
hasTransitionGeneric: scopes.includes('transition:generic'),
repoPermissions: new Map(),
blobPermissions: [],
}
for (const scope of scopes) {
const repo = parseRepoScope(scope)
if (repo) {
const existing = result.repoPermissions.get(repo.collection) || { create: false, update: false, delete: false }
for (const action of repo.actions) {
existing[action] = true
}
result.repoPermissions.set(repo.collection, existing)
}
const blob = parseBlobScope(scope)
if (blob) {
result.blobPermissions.push(...blob.accept)
}
}
return result
}
/**
* Scope permission checker
*/
export class ScopePermissions {
constructor(scopeString) {
this.scopes = (scopeString || '').split(' ').filter(Boolean)
this.hasAtproto = this.scopes.includes('atproto')
this.hasTransitionGeneric = this.scopes.includes('transition:generic')
this.repoScopes = this.scopes.map(parseRepoScope).filter(Boolean)
this.blobScopes = this.scopes.map(parseBlobScope).filter(Boolean)
}
allowsRepo(collection, action) {
if (this.hasAtproto || this.hasTransitionGeneric) return true
return this.repoScopes.some((s) => (s.collection === '*' || s.collection === collection) && s.actions.includes(action))
}
allowsBlob(mime) {
if (this.hasAtproto || this.hasTransitionGeneric) return true
return this.blobScopes.some((s) => s.accept.some((pattern) => matchesMime(pattern, mime)))
}
assertRepo(collection, action) {
if (!this.allowsRepo(collection, action)) {
throw new ScopeMissingError(`Missing required scope for ${action} on ${collection}`)
}
}
assertBlob(mime) {
if (!this.allowsBlob(mime)) {
throw new ScopeMissingError(`Missing required scope for blob type ${mime}`)
}
}
}
export class ScopeMissingError extends Error {
constructor(message) {
super(message)
this.name = 'ScopeMissingError'
}
}
```
**Step 2: Update src/pds.js to import from core**
Add imports and remove local implementations.
**Step 3: Run tests**
Run: `npm test`
Expected: All tests pass
**Step 4: Commit**
```bash
git add packages/core/auth.js src/pds.js
git commit -m "refactor: extract auth utilities to @pdsjs/core"
```
---
### Task 2.4: Update core index.js to re-export all utilities
**Files:**
- Modify: `packages/core/index.js`
**Step 1: Update index.js**
```javascript
// @pdsjs/core - PDS business logic
// Re-export all utilities for backwards compatibility
export * from './crypto.js'
export * from './repo.js'
export * from './auth.js'
```
**Step 2: Run tests**
Run: `npm test`
Expected: All tests pass
**Step 3: Commit**
```bash
git add packages/core/index.js
git commit -m "refactor: re-export utilities from @pdsjs/core index"
```
---
## Phase 3: Define Storage and Blob Ports
### Task 3.1: Define StoragePort and BlobPort interfaces
**Files:**
- Create: `packages/core/ports.js`
- Modify: `packages/core/index.js`
**Step 1: Create ports.js with JSDoc interface definitions**
```javascript
// packages/core/ports.js
/**
* @typedef {Object} StoragePort
* @property {(uri: string) => Promise<{cid: string, value: Uint8Array}|null>} getRecord
* @property {(collection: string, cursor: string|null, limit: number) => Promise<{records: Array, cursor: string|null}>} listRecords
* @property {(uri: string, cid: string, collection: string, rkey: string, value: Uint8Array) => Promise<void>} putRecord
* @property {(uri: string) => Promise<void>} deleteRecord
* @property {(cid: string) => Promise<Uint8Array|null>} getBlock
* @property {(cid: string, data: Uint8Array) => Promise<void>} putBlock
* @property {() => Promise<{seq: number, cid: string, rev: string}|null>} getLatestCommit
* @property {(seq: number, cid: string, rev: string, prev: string|null) => Promise<void>} putCommit
* @property {(cursor: number|null, limit: number) => Promise<{events: Array, cursor: number|null}>} getEvents
* @property {(seq: number, did: string, commitCid: string, evt: Uint8Array) => Promise<void>} putEvent
* @property {(cid: string) => Promise<{mimeType: string, size: number, createdAt: number}|null>} getBlob
* @property {(cid: string, mimeType: string, size: number) => Promise<void>} putBlob
* @property {(cid: string) => Promise<void>} deleteBlob
* @property {(cursor: string|null, limit: number) => Promise<{cids: string[], cursor: string|null}>} listBlobs
* @property {(olderThan: number) => Promise<string[]>} getOrphanedBlobs
* @property {(blobCid: string, recordUri: string) => Promise<void>} linkBlobToRecord
* @property {(recordUri: string) => Promise<void>} unlinkBlobsFromRecord
* @property {() => Promise<string|null>} getDid
* @property {(did: string) => Promise<void>} setDid
* @property {() => Promise<Uint8Array|null>} getPrivateKey
* @property {(key: Uint8Array) => Promise<void>} setPrivateKey
* @property {() => Promise<string|null>} getHandle
* @property {(handle: string) => Promise<void>} setHandle
*/
/**
* @typedef {Object} BlobPort
* @property {(did: string, cid: string) => Promise<{data: Uint8Array, mimeType: string}|null>} get
* @property {(did: string, cid: string, data: Uint8Array, mimeType: string) => Promise<void>} put
* @property {(did: string, cid: string) => Promise<void>} delete
*/
// Export empty object - types are defined via JSDoc
export {}
```
**Step 2: Add ports export to index.js**
```javascript
// @pdsjs/core - PDS business logic
export * from './crypto.js'
export * from './repo.js'
export * from './auth.js'
// Note: ports.js exports types only via JSDoc
```
**Step 3: Commit**
```bash
git add packages/core/ports.js packages/core/index.js
git commit -m "feat: define StoragePort and BlobPort interfaces"
```
---
## Phase 4: Create Storage Adapters
### Task 4.1: Create @pdsjs/storage-sqlite package
**Files:**
- Create: `packages/storage-sqlite/package.json`
- Create: `packages/storage-sqlite/index.js`
**Step 1: Create package.json**
```json
{
"name": "@pdsjs/storage-sqlite",
"version": "0.1.0",
"type": "module",
"main": "index.js",
"exports": {
".": "./index.js"
},
"peerDependencies": {
"better-sqlite3": ">=9.0.0"
},
"peerDependenciesMeta": {
"better-sqlite3": {
"optional": true
}
}
}
```
**Step 2: Create index.js with D1 adapter**
```javascript
// @pdsjs/storage-sqlite - SQLite storage adapters
/**
* Create storage adapter for Cloudflare D1 (Durable Object SQLite)
* @param {Object} sql - D1 SQL interface from state.storage.sql
* @returns {import('@pdsjs/core').StoragePort}
*/
export function createD1Storage(sql) {
return {
async getRecord(uri) {
const row = sql.exec('SELECT cid, value FROM records WHERE uri = ?', uri).one()
return row ? { cid: row.cid, value: row.value } : null
},
async listRecords(collection, cursor, limit) {
const rows = cursor
? sql.exec('SELECT uri, cid, rkey, value FROM records WHERE collection = ? AND rkey > ? ORDER BY rkey LIMIT ?', collection, cursor, limit).toArray()
: sql.exec('SELECT uri, cid, rkey, value FROM records WHERE collection = ? ORDER BY rkey LIMIT ?', collection, limit).toArray()
return {
records: rows,
cursor: rows.length === limit ? rows[rows.length - 1].rkey : null,
}
},
async putRecord(uri, cid, collection, rkey, value) {
sql.exec('INSERT OR REPLACE INTO records (uri, cid, collection, rkey, value) VALUES (?, ?, ?, ?, ?)', uri, cid, collection, rkey, value)
},
async deleteRecord(uri) {
sql.exec('DELETE FROM record_blobs WHERE record_uri = ?', uri)
sql.exec('DELETE FROM records WHERE uri = ?', uri)
},
async getBlock(cid) {
const row = sql.exec('SELECT data FROM blocks WHERE cid = ?', cid).one()
return row ? row.data : null
},
async putBlock(cid, data) {
sql.exec('INSERT OR IGNORE INTO blocks (cid, data) VALUES (?, ?)', cid, data)
},
async getLatestCommit() {
const row = sql.exec('SELECT seq, cid, rev FROM commits ORDER BY seq DESC LIMIT 1').one()
return row || null
},
async putCommit(seq, cid, rev, prev) {
sql.exec('INSERT INTO commits (seq, cid, rev, prev) VALUES (?, ?, ?, ?)', seq, cid, rev, prev)
},
async getEvents(cursor, limit) {
const rows = cursor
? sql.exec('SELECT seq, did, commit_cid, evt FROM seq_events WHERE seq > ? ORDER BY seq LIMIT ?', cursor, limit).toArray()
: sql.exec('SELECT seq, did, commit_cid, evt FROM seq_events ORDER BY seq LIMIT ?', limit).toArray()
return {
events: rows,
cursor: rows.length > 0 ? rows[rows.length - 1].seq : cursor,
}
},
async putEvent(seq, did, commitCid, evt) {
sql.exec('INSERT INTO seq_events (seq, did, commit_cid, evt) VALUES (?, ?, ?, ?)', seq, did, commitCid, evt)
},
async getBlob(cid) {
const row = sql.exec('SELECT mime_type, size, created_at FROM blobs WHERE cid = ?', cid).one()
return row ? { mimeType: row.mime_type, size: row.size, createdAt: row.created_at } : null
},
async putBlob(cid, mimeType, size) {
sql.exec('INSERT OR REPLACE INTO blobs (cid, mime_type, size, created_at) VALUES (?, ?, ?, ?)', cid, mimeType, size, Date.now())
},
async deleteBlob(cid) {
sql.exec('DELETE FROM blobs WHERE cid = ?', cid)
},
async listBlobs(cursor, limit) {
const rows = cursor
? sql.exec('SELECT cid FROM blobs WHERE cid > ? ORDER BY cid LIMIT ?', cursor, limit).toArray()
: sql.exec('SELECT cid FROM blobs ORDER BY cid LIMIT ?', limit).toArray()
return {
cids: rows.map((r) => r.cid),
cursor: rows.length === limit ? rows[rows.length - 1].cid : null,
}
},
async getOrphanedBlobs(olderThan) {
const rows = sql.exec(
'SELECT b.cid FROM blobs b LEFT JOIN record_blobs rb ON b.cid = rb.blob_cid WHERE rb.blob_cid IS NULL AND b.created_at < ?',
olderThan
).toArray()
return rows.map((r) => r.cid)
},
async linkBlobToRecord(blobCid, recordUri) {
sql.exec('INSERT OR IGNORE INTO record_blobs (blob_cid, record_uri) VALUES (?, ?)', blobCid, recordUri)
},
async unlinkBlobsFromRecord(recordUri) {
sql.exec('DELETE FROM record_blobs WHERE record_uri = ?', recordUri)
},
async getDid() {
const row = sql.exec("SELECT value FROM metadata WHERE key = 'did'").one()
return row ? row.value : null
},
async setDid(did) {
sql.exec("INSERT OR REPLACE INTO metadata (key, value) VALUES ('did', ?)", did)
},
async getPrivateKey() {
const row = sql.exec("SELECT value FROM metadata WHERE key = 'privateKey'").one()
return row ? new Uint8Array(row.value) : null
},
async setPrivateKey(key) {
sql.exec("INSERT OR REPLACE INTO metadata (key, value) VALUES ('privateKey', ?)", key)
},
async getHandle() {
const row = sql.exec("SELECT value FROM metadata WHERE key = 'handle'").one()
return row ? row.value : null
},
async setHandle(handle) {
sql.exec("INSERT OR REPLACE INTO metadata (key, value) VALUES ('handle', ?)", handle)
},
}
}
/**
* Create storage adapter for better-sqlite3 (Node.js)
* @param {import('better-sqlite3').Database} db
* @returns {import('@pdsjs/core').StoragePort}
*/
export function createBetterSqliteStorage(db) {
// Initialize schema
db.exec(`
CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value BLOB);
CREATE TABLE IF NOT EXISTS blocks (cid TEXT PRIMARY KEY, data BLOB);
CREATE TABLE IF NOT EXISTS records (uri TEXT PRIMARY KEY, cid TEXT, collection TEXT, rkey TEXT, value BLOB);
CREATE TABLE IF NOT EXISTS commits (seq INTEGER PRIMARY KEY, cid TEXT, rev TEXT, prev TEXT);
CREATE TABLE IF NOT EXISTS seq_events (seq INTEGER PRIMARY KEY, did TEXT, commit_cid TEXT, evt BLOB);
CREATE TABLE IF NOT EXISTS blobs (cid TEXT PRIMARY KEY, mime_type TEXT, size INTEGER, created_at INTEGER);
CREATE TABLE IF NOT EXISTS record_blobs (blob_cid TEXT, record_uri TEXT, PRIMARY KEY (blob_cid, record_uri));
CREATE INDEX IF NOT EXISTS idx_records_collection ON records(collection, rkey);
`)
const stmts = {
getRecord: db.prepare('SELECT cid, value FROM records WHERE uri = ?'),
listRecords: db.prepare('SELECT uri, cid, rkey, value FROM records WHERE collection = ? ORDER BY rkey LIMIT ?'),
listRecordsAfter: db.prepare('SELECT uri, cid, rkey, value FROM records WHERE collection = ? AND rkey > ? ORDER BY rkey LIMIT ?'),
putRecord: db.prepare('INSERT OR REPLACE INTO records (uri, cid, collection, rkey, value) VALUES (?, ?, ?, ?, ?)'),
deleteRecord: db.prepare('DELETE FROM records WHERE uri = ?'),
deleteRecordBlobs: db.prepare('DELETE FROM record_blobs WHERE record_uri = ?'),
getBlock: db.prepare('SELECT data FROM blocks WHERE cid = ?'),
putBlock: db.prepare('INSERT OR IGNORE INTO blocks (cid, data) VALUES (?, ?)'),
getLatestCommit: db.prepare('SELECT seq, cid, rev FROM commits ORDER BY seq DESC LIMIT 1'),
putCommit: db.prepare('INSERT INTO commits (seq, cid, rev, prev) VALUES (?, ?, ?, ?)'),
getEvents: db.prepare('SELECT seq, did, commit_cid, evt FROM seq_events ORDER BY seq LIMIT ?'),
getEventsAfter: db.prepare('SELECT seq, did, commit_cid, evt FROM seq_events WHERE seq > ? ORDER BY seq LIMIT ?'),
putEvent: db.prepare('INSERT INTO seq_events (seq, did, commit_cid, evt) VALUES (?, ?, ?, ?)'),
getBlob: db.prepare('SELECT mime_type, size, created_at FROM blobs WHERE cid = ?'),
putBlob: db.prepare('INSERT OR REPLACE INTO blobs (cid, mime_type, size, created_at) VALUES (?, ?, ?, ?)'),
deleteBlob: db.prepare('DELETE FROM blobs WHERE cid = ?'),
listBlobs: db.prepare('SELECT cid FROM blobs ORDER BY cid LIMIT ?'),
listBlobsAfter: db.prepare('SELECT cid FROM blobs WHERE cid > ? ORDER BY cid LIMIT ?'),
getOrphanedBlobs: db.prepare('SELECT b.cid FROM blobs b LEFT JOIN record_blobs rb ON b.cid = rb.blob_cid WHERE rb.blob_cid IS NULL AND b.created_at < ?'),
linkBlobToRecord: db.prepare('INSERT OR IGNORE INTO record_blobs (blob_cid, record_uri) VALUES (?, ?)'),
unlinkBlobsFromRecord: db.prepare('DELETE FROM record_blobs WHERE record_uri = ?'),
getMetadata: db.prepare('SELECT value FROM metadata WHERE key = ?'),
setMetadata: db.prepare('INSERT OR REPLACE INTO metadata (key, value) VALUES (?, ?)'),
}
return {
async getRecord(uri) {
const row = stmts.getRecord.get(uri)
return row ? { cid: row.cid, value: row.value } : null
},
async listRecords(collection, cursor, limit) {
const rows = cursor
? stmts.listRecordsAfter.all(collection, cursor, limit)
: stmts.listRecords.all(collection, limit)
return {
records: rows,
cursor: rows.length === limit ? rows[rows.length - 1].rkey : null,
}
},
async putRecord(uri, cid, collection, rkey, value) {
stmts.putRecord.run(uri, cid, collection, rkey, value)
},
async deleteRecord(uri) {
stmts.deleteRecordBlobs.run(uri)
stmts.deleteRecord.run(uri)
},
async getBlock(cid) {
const row = stmts.getBlock.get(cid)
return row ? row.data : null
},
async putBlock(cid, data) {
stmts.putBlock.run(cid, data)
},
async getLatestCommit() {
return stmts.getLatestCommit.get() || null
},
async putCommit(seq, cid, rev, prev) {
stmts.putCommit.run(seq, cid, rev, prev)
},
async getEvents(cursor, limit) {
const rows = cursor
? stmts.getEventsAfter.all(cursor, limit)
: stmts.getEvents.all(limit)
return {
events: rows,
cursor: rows.length > 0 ? rows[rows.length - 1].seq : cursor,
}
},
async putEvent(seq, did, commitCid, evt) {
stmts.putEvent.run(seq, did, commitCid, evt)
},
async getBlob(cid) {
const row = stmts.getBlob.get(cid)
return row ? { mimeType: row.mime_type, size: row.size, createdAt: row.created_at } : null
},
async putBlob(cid, mimeType, size) {
stmts.putBlob.run(cid, mimeType, size, Date.now())
},
async deleteBlob(cid) {
stmts.deleteBlob.run(cid)
},
async listBlobs(cursor, limit) {
const rows = cursor
? stmts.listBlobsAfter.all(cursor, limit)
: stmts.listBlobs.all(limit)
return {
cids: rows.map((r) => r.cid),
cursor: rows.length === limit ? rows[rows.length - 1].cid : null,
}
},
async getOrphanedBlobs(olderThan) {
return stmts.getOrphanedBlobs.all(olderThan).map((r) => r.cid)
},
async linkBlobToRecord(blobCid, recordUri) {
stmts.linkBlobToRecord.run(blobCid, recordUri)
},
async unlinkBlobsFromRecord(recordUri) {
stmts.unlinkBlobsFromRecord.run(recordUri)
},
async getDid() {
const row = stmts.getMetadata.get('did')
return row ? row.value : null
},
async setDid(did) {
stmts.setMetadata.run('did', did)
},
async getPrivateKey() {
const row = stmts.getMetadata.get('privateKey')
return row ? new Uint8Array(row.value) : null
},
async setPrivateKey(key) {
stmts.setMetadata.run('privateKey', key)
},
async getHandle() {
const row = stmts.getMetadata.get('handle')
return row ? row.value : null
},
async setHandle(handle) {
stmts.setMetadata.run('handle', handle)
},
}
}
```
**Step 3: Commit**
```bash
git add packages/storage-sqlite/
git commit -m "feat: create @pdsjs/storage-sqlite with D1 and better-sqlite3 adapters"
```
---
## Phase 5: Create Blob Adapters
### Task 5.1: Create @pdsjs/blobs-fs package
**Files:**
- Create: `packages/blobs-fs/package.json`
- Create: `packages/blobs-fs/index.js`
**Step 1: Create package.json**
```json
{
"name": "@pdsjs/blobs-fs",
"version": "0.1.0",
"type": "module",
"main": "index.js"
}
```
**Step 2: Create index.js**
```javascript
// @pdsjs/blobs-fs - Filesystem blob storage
import { readFile, writeFile, unlink, mkdir } from 'node:fs/promises'
import { join, dirname } from 'node:path'
/**
* Create filesystem blob adapter
* @param {string} baseDir - Base directory for blob storage
* @returns {import('@pdsjs/core').BlobPort}
*/
export function createFsBlobs(baseDir) {
function blobPath(did, cid) {
// Use first 2 chars of CID for sharding
const shard = cid.slice(1, 3)
return join(baseDir, did, shard, cid)
}
return {
async get(did, cid) {
try {
const path = blobPath(did, cid)
const data = await readFile(path)
const metaPath = path + '.meta'
const meta = JSON.parse(await readFile(metaPath, 'utf8'))
return { data: new Uint8Array(data), mimeType: meta.mimeType }
} catch (e) {
if (e.code === 'ENOENT') return null
throw e
}
},
async put(did, cid, data, mimeType) {
const path = blobPath(did, cid)
await mkdir(dirname(path), { recursive: true })
await writeFile(path, data)
await writeFile(path + '.meta', JSON.stringify({ mimeType }))
},
async delete(did, cid) {
const path = blobPath(did, cid)
try {
await unlink(path)
await unlink(path + '.meta')
} catch (e) {
if (e.code !== 'ENOENT') throw e
}
},
}
}
```
**Step 3: Commit**
```bash
git add packages/blobs-fs/
git commit -m "feat: create @pdsjs/blobs-fs filesystem blob adapter"
```
---
### Task 5.2: Create @pdsjs/blobs-s3 package
**Files:**
- Create: `packages/blobs-s3/package.json`
- Create: `packages/blobs-s3/index.js`
**Step 1: Create package.json**
```json
{
"name": "@pdsjs/blobs-s3",
"version": "0.1.0",
"type": "module",
"main": "index.js"
}
```
**Step 2: Create index.js**
```javascript
// @pdsjs/blobs-s3 - S3-compatible blob storage (R2, MinIO, AWS S3)
/**
* Create R2 blob adapter (Cloudflare Workers)
* @param {R2Bucket} bucket - R2 bucket binding
* @returns {import('@pdsjs/core').BlobPort}
*/
export function createR2Blobs(bucket) {
return {
async get(did, cid) {
const obj = await bucket.get(`${did}/${cid}`)
if (!obj) return null
const data = new Uint8Array(await obj.arrayBuffer())
const mimeType = obj.customMetadata?.mime_type || 'application/octet-stream'
return { data, mimeType }
},
async put(did, cid, data, mimeType) {
await bucket.put(`${did}/${cid}`, data, {
customMetadata: { mime_type: mimeType },
})
},
async delete(did, cid) {
await bucket.delete(`${did}/${cid}`)
},
}
}
/**
* Create S3-compatible blob adapter (AWS S3, MinIO)
* @param {Object} options
* @param {Object} options.client - S3 client instance
* @param {string} options.bucket - Bucket name
* @returns {import('@pdsjs/core').BlobPort}
*/
export function createS3Blobs({ client, bucket }) {
return {
async get(did, cid) {
try {
const response = await client.send(new client.GetObjectCommand({
Bucket: bucket,
Key: `${did}/${cid}`,
}))
const data = new Uint8Array(await response.Body.transformToByteArray())
const mimeType = response.Metadata?.mime_type || 'application/octet-stream'
return { data, mimeType }
} catch (e) {
if (e.name === 'NoSuchKey') return null
throw e
}
},
async put(did, cid, data, mimeType) {
await client.send(new client.PutObjectCommand({
Bucket: bucket,
Key: `${did}/${cid}`,
Body: data,
ContentType: mimeType,
Metadata: { mime_type: mimeType },
}))
},
async delete(did, cid) {
await client.send(new client.DeleteObjectCommand({
Bucket: bucket,
Key: `${did}/${cid}`,
}))
},
}
}
```
**Step 3: Commit**
```bash
git add packages/blobs-s3/
git commit -m "feat: create @pdsjs/blobs-s3 for R2 and S3-compatible storage"
```
---
## Phase 6: Refactor Core PDS Class
### Task 6.1: Create PDS class that accepts ports
**Files:**
- Create: `packages/core/pds.js`
**Step 1: Create pds.js with port-based PDS class**
This is the main refactoring task. The `PersonalDataServer` class will:
- Accept `storage` and `blobs` ports via constructor
- Move all handler methods from `src/pds.js`
- Remove direct SQL and R2 calls
- Use `this.storage.*` and `this.blobs.*` instead
```javascript
// packages/core/pds.js
import { cborEncode, cborDecode, createCid, createBlobCid, cidToString, cidFromString, createTid, getKeyDepth, buildCarFile, sniffMimeType, findBlobRefs } from './repo.js'
import { importPrivateKey, sign, generateKeyPair, base64UrlEncode, computeJwkThumbprint } from './crypto.js'
import { ScopePermissions, createAccessJwt, createRefreshJwt, verifyAccessJwt, verifyRefreshJwt } from './auth.js'
/**
* @typedef {import('./ports.js').StoragePort} StoragePort
* @typedef {import('./ports.js').BlobPort} BlobPort
*/
/**
* Personal Data Server - core business logic
*/
export class PersonalDataServer {
/**
* @param {Object} ports
* @param {StoragePort} ports.storage
* @param {BlobPort} ports.blobs
*/
constructor({ storage, blobs }) {
this.storage = storage
this.blobs = blobs
}
/**
* Handle incoming request
* @param {Request} request
* @returns {Promise<Response>}
*/
async fetch(request) {
const url = new URL(request.url)
const route = pdsRoutes[url.pathname]
if (!route) {
return new Response('Not Found', { status: 404 })
}
if (route.method && request.method !== route.method) {
return new Response('Method Not Allowed', { status: 405 })
}
try {
return await route.handler(this, request, url)
} catch (error) {
console.error('Handler error:', error)
return Response.json({ error: 'InternalError', message: error.message }, { status: 500 })
}
}
// ... All handler methods will be moved here from src/pds.js
// Example structure:
async handleCreateRecord(request) {
const { repo, collection, rkey, record } = await request.json()
// Implementation using this.storage.putRecord(), etc.
}
async handleGetRecord(url) {
const repo = url.searchParams.get('repo')
const collection = url.searchParams.get('collection')
const rkey = url.searchParams.get('rkey')
// Implementation using this.storage.getRecord()
}
// ... etc for all 40+ handlers
}
// Route table
const pdsRoutes = {
'/xrpc/com.atproto.repo.createRecord': {
method: 'POST',
handler: (pds, req, url) => pds.handleCreateRecord(req),
},
'/xrpc/com.atproto.repo.getRecord': {
handler: (pds, req, url) => pds.handleGetRecord(url),
},
// ... all other routes
}
```
**Note:** This task involves significant code movement. The full implementation requires:
1. Moving all handler methods from `src/pds.js` to `packages/core/pds.js`
2. Replacing `this.sql.exec(...)` calls with `await this.storage.*()`
3. Replacing `this.env.BLOBS.*` calls with `await this.blobs.*()`
4. Updating the route table
**Step 2: Run tests after full migration**
Run: `npm test`
Expected: All tests pass
**Step 3: Commit**
```bash
git add packages/core/pds.js
git commit -m "refactor: create port-based PersonalDataServer class"
```
---
## Phase 7: Create Platform Adapters
### Task 7.1: Create @pdsjs/cloudflare adapter
**Files:**
- Create: `packages/cloudflare/package.json`
- Create: `packages/cloudflare/index.js`
**Step 1: Create package.json**
```json
{
"name": "@pdsjs/cloudflare",
"version": "0.1.0",
"type": "module",
"main": "index.js",
"dependencies": {
"@pdsjs/core": "workspace:*",
"@pdsjs/storage-sqlite": "workspace:*",
"@pdsjs/blobs-s3": "workspace:*"
}
}
```
**Step 2: Create index.js**
```javascript
// @pdsjs/cloudflare - Cloudflare Workers + Durable Objects adapter
import { PersonalDataServer } from '@pdsjs/core'
import { createD1Storage } from '@pdsjs/storage-sqlite'
import { createR2Blobs } from '@pdsjs/blobs-s3'
/**
* Durable Object class for per-user PDS instances
*/
export class PDSDurableObject {
constructor(state, env) {
this.state = state
this.env = env
// Initialize storage adapter
const storage = createD1Storage(state.storage.sql)
// Initialize blob adapter (optional)
const blobs = env.BLOBS ? createR2Blobs(env.BLOBS) : null
// Create PDS instance
this.pds = new PersonalDataServer({ storage, blobs })
// Initialize schema on first access
this.initialized = this.initSchema()
}
async initSchema() {
this.state.storage.sql.exec(`
CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value BLOB);
CREATE TABLE IF NOT EXISTS blocks (cid TEXT PRIMARY KEY, data BLOB);
CREATE TABLE IF NOT EXISTS records (uri TEXT PRIMARY KEY, cid TEXT, collection TEXT, rkey TEXT, value BLOB);
CREATE TABLE IF NOT EXISTS commits (seq INTEGER PRIMARY KEY, cid TEXT, rev TEXT, prev TEXT);
CREATE TABLE IF NOT EXISTS seq_events (seq INTEGER PRIMARY KEY, did TEXT, commit_cid TEXT, evt BLOB);
CREATE TABLE IF NOT EXISTS blobs (cid TEXT PRIMARY KEY, mime_type TEXT, size INTEGER, created_at INTEGER);
CREATE TABLE IF NOT EXISTS record_blobs (blob_cid TEXT, record_uri TEXT, PRIMARY KEY (blob_cid, record_uri));
CREATE INDEX IF NOT EXISTS idx_records_collection ON records(collection, rkey);
`)
}
async fetch(request) {
await this.initialized
return this.pds.fetch(request)
}
async alarm() {
// Blob cleanup job
await this.pds.runBlobCleanup()
}
}
/**
* Worker entry point - routes requests to appropriate DO instance
*/
export default {
async fetch(request, env) {
const url = new URL(request.url)
// Extract DID from request (auth header, query param, or path)
const did = extractDid(request, url)
if (!did) {
// Use default instance for unauthenticated requests
const stub = env.PDS.get(env.PDS.idFromName('default'))
return stub.fetch(request)
}
// Route to user's DO instance
const stub = env.PDS.get(env.PDS.idFromName(did))
return stub.fetch(request)
},
}
function extractDid(request, url) {
// Check query param
const didParam = url.searchParams.get('did') || url.searchParams.get('repo')
if (didParam?.startsWith('did:')) return didParam
// Check auth header for DID
const auth = request.headers.get('authorization')
if (auth) {
// Extract from JWT sub claim (simplified - real impl would verify)
try {
const token = auth.replace(/^(Bearer|DPoP)\s+/i, '')
const [, payloadB64] = token.split('.')
const payload = JSON.parse(atob(payloadB64.replace(/-/g, '+').replace(/_/g, '/')))
if (payload.sub?.startsWith('did:')) return payload.sub
} catch {}
}
return null
}
```
**Step 3: Commit**
```bash
git add packages/cloudflare/
git commit -m "feat: create @pdsjs/cloudflare adapter"
```
---
### Task 7.2: Create @pdsjs/node adapter
**Files:**
- Create: `packages/node/package.json`
- Create: `packages/node/index.js`
**Step 1: Create package.json**
```json
{
"name": "@pdsjs/node",
"version": "0.1.0",
"type": "module",
"main": "index.js",
"dependencies": {
"@pdsjs/core": "workspace:*",
"@pdsjs/storage-sqlite": "workspace:*",
"@pdsjs/blobs-fs": "workspace:*"
},
"peerDependencies": {
"better-sqlite3": ">=9.0.0"
}
}
```
**Step 2: Create index.js**
```javascript
// @pdsjs/node - Node.js HTTP server adapter
import { createServer as createHttpServer } from 'node:http'
import { PersonalDataServer } from '@pdsjs/core'
import { createBetterSqliteStorage } from '@pdsjs/storage-sqlite'
import { createFsBlobs } from '@pdsjs/blobs-fs'
import Database from 'better-sqlite3'
/**
* Create Node.js PDS server
* @param {Object} options
* @param {string} options.dbPath - Path to SQLite database
* @param {string} options.blobsDir - Directory for blob storage
* @param {number} [options.port=3000] - Server port
* @returns {Object}
*/
export function createServer({ dbPath, blobsDir, port = 3000 }) {
const db = new Database(dbPath)
const storage = createBetterSqliteStorage(db)
const blobs = createFsBlobs(blobsDir)
const pds = new PersonalDataServer({ storage, blobs })
const server = createHttpServer(async (req, res) => {
try {
const request = nodeToWebRequest(req)
const response = await pds.fetch(request)
await webToNodeResponse(response, res)
} catch (error) {
console.error('Server error:', error)
res.statusCode = 500
res.end(JSON.stringify({ error: 'InternalError', message: error.message }))
}
})
// Start blob cleanup interval
const cleanupInterval = setInterval(() => pds.runBlobCleanup(), 60 * 60 * 1000)
return {
pds,
server,
listen() {
return new Promise((resolve) => {
server.listen(port, () => {
console.log(`PDS listening on http://localhost:${port}`)
resolve(server)
})
})
},
close() {
clearInterval(cleanupInterval)
db.close()
return new Promise((resolve) => server.close(resolve))
},
}
}
/**
* Convert Node.js IncomingMessage to Web Request
*/
function nodeToWebRequest(req) {
const protocol = req.socket.encrypted ? 'https' : 'http'
const url = new URL(req.url, `${protocol}://${req.headers.host}`)
const headers = new Headers()
for (const [key, value] of Object.entries(req.headers)) {
if (value) headers.set(key, Array.isArray(value) ? value.join(', ') : value)
}
const hasBody = req.method !== 'GET' && req.method !== 'HEAD'
return new Request(url.toString(), {
method: req.method,
headers,
body: hasBody ? req : null,
duplex: hasBody ? 'half' : undefined,
})
}
/**
* Write Web Response to Node.js ServerResponse
*/
async function webToNodeResponse(response, res) {
res.statusCode = response.status
res.statusMessage = response.statusText
for (const [key, value] of response.headers) {
res.setHeader(key, value)
}
if (response.body) {
const reader = response.body.getReader()
while (true) {
const { done, value } = await reader.read()
if (done) break
res.write(value)
}
}
res.end()
}
```
**Step 3: Commit**
```bash
git add packages/node/
git commit -m "feat: create @pdsjs/node HTTP server adapter"
```
---
## Phase 8: Testing
### Task 8.1: Create shared test scenarios
**Files:**
- Create: `test/helpers/scenarios.js`
**Step 1: Extract reusable test scenarios**
```javascript
// test/helpers/scenarios.js
// Shared test scenarios for both Cloudflare and Node E2E tests
export const TEST_DID = 'did:plc:test'
export const TEST_PASSWORD = 'test-password'
/**
* Test basic server endpoints
*/
export async function testServerEndpoints(BASE) {
const tests = []
// Root returns something
const rootRes = await fetch(`${BASE}/`)
tests.push({ name: 'root returns ok', pass: rootRes.ok })
// describeServer returns DID
const descRes = await fetch(`${BASE}/xrpc/com.atproto.server.describeServer`)
const descData = await descRes.json()
tests.push({ name: 'describeServer returns did', pass: !!descData.did })
return tests
}
/**
* Test authentication flow
*/
export async function testAuthentication(BASE, did, password) {
const tests = []
// createSession
const sessionRes = await fetch(`${BASE}/xrpc/com.atproto.server.createSession`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ identifier: did, password }),
})
const sessionData = await sessionRes.json()
tests.push({ name: 'createSession returns tokens', pass: !!sessionData.accessJwt })
return { tests, token: sessionData.accessJwt }
}
/**
* Test record operations
*/
export async function testRecordOperations(BASE, did, token) {
const tests = []
// createRecord
const createRes = await fetch(`${BASE}/xrpc/com.atproto.repo.createRecord`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`,
},
body: JSON.stringify({
repo: did,
collection: 'app.bsky.feed.post',
record: { text: 'test', createdAt: new Date().toISOString() },
}),
})
const createData = await createRes.json()
tests.push({ name: 'createRecord returns uri', pass: !!createData.uri })
const rkey = createData.uri?.split('/').pop()
// getRecord
const getRes = await fetch(
`${BASE}/xrpc/com.atproto.repo.getRecord?repo=${did}&collection=app.bsky.feed.post&rkey=${rkey}`
)
const getData = await getRes.json()
tests.push({ name: 'getRecord returns value', pass: !!getData.value })
// deleteRecord
const deleteRes = await fetch(`${BASE}/xrpc/com.atproto.repo.deleteRecord`, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`,
},
body: JSON.stringify({ repo: did, collection: 'app.bsky.feed.post', rkey }),
})
tests.push({ name: 'deleteRecord succeeds', pass: deleteRes.ok })
return tests
}
```
**Step 2: Commit**
```bash
git add test/helpers/scenarios.js
git commit -m "test: create shared test scenarios"
```
---
### Task 8.2: Create Node E2E test file
**Files:**
- Create: `test/e2e-node.test.js`
**Step 1: Create e2e-node.test.js**
```javascript
// test/e2e-node.test.js
// E2E tests for Node.js PDS adapter
import { describe, it, expect, beforeAll, afterAll } from 'vitest'
import { createServer } from '@pdsjs/node'
import { randomBytes } from 'node:crypto'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
const PORT = 8788
const BASE = `http://localhost:${PORT}`
const DID = `did:plc:test${randomBytes(8).toString('hex')}`
const PASSWORD = 'test-password'
let server
let tempDir
let token
describe('Node E2E Tests', () => {
beforeAll(async () => {
// Create temp directory
tempDir = await mkdtemp(join(tmpdir(), 'pds-test-'))
// Start server
server = createServer({
dbPath: join(tempDir, 'pds.db'),
blobsDir: join(tempDir, 'blobs'),
port: PORT,
})
await server.listen()
// Initialize PDS
// ... initialization logic
})
afterAll(async () => {
await server?.close()
await rm(tempDir, { recursive: true, force: true })
})
describe('Server endpoints', () => {
it('root returns ok', async () => {
const res = await fetch(`${BASE}/`)
expect(res.ok).toBe(true)
})
it('describeServer returns DID', async () => {
const res = await fetch(`${BASE}/xrpc/com.atproto.server.describeServer`)
const data = await res.json()
expect(data.did).toBeTruthy()
})
})
// ... same test structure as e2e.test.js
})
```
**Step 2: Commit**
```bash
git add test/e2e-node.test.js
git commit -m "test: create Node E2E test file"
```
---
## Phase 9: Cleanup
### Task 9.1: Remove old src/pds.js
**Files:**
- Delete: `src/pds.js`
- Modify: `wrangler.toml` to point to new entry
**Step 1: Update wrangler.toml**
```toml
main = "packages/cloudflare/index.js"
```
**Step 2: Delete src/pds.js**
Run: `rm src/pds.js`
**Step 3: Run all tests**
Run: `npm test`
Expected: All tests pass
**Step 4: Commit**
```bash
git add -A
git commit -m "refactor: complete PDS core separation"
```
---
## Summary
This plan separates the 5,560-line PDS into:
| Package | Purpose | Size (est.) |
|---------|---------|-------------|
| `@pdsjs/core` | Business logic, handlers | ~3,500 lines |
| `@pdsjs/storage-sqlite` | D1 + better-sqlite3 adapters | ~300 lines |
| `@pdsjs/blobs-fs` | Filesystem blobs | ~50 lines |
| `@pdsjs/blobs-s3` | R2/S3 blobs | ~60 lines |
| `@pdsjs/cloudflare` | Workers + DO adapter | ~150 lines |
| `@pdsjs/node` | Node HTTP adapter | ~100 lines |
Both Cloudflare and Node deployments run identical E2E tests, proving the abstraction works.