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

# pnpm-workspace.yaml
packages:
  - 'packages/*'

Step 2: Update root package.json

Add to package.json:

{
  "private": true,
  "workspaces": ["packages/*"]
}

Step 3: Create packages directory

Run: mkdir -p packages

Step 4: Commit

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

{
  "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

// @pdsjs/core - PDS business logic
// Exports will be added as code is extracted

export {}

Step 3: Commit

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

// 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:

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

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

// 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:

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

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

// 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

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

// @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

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

// 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

// @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

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

{
  "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

// @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

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

{
  "name": "@pdsjs/blobs-fs",
  "version": "0.1.0",
  "type": "module",
  "main": "index.js"
}

Step 2: Create index.js

// @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

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

{
  "name": "@pdsjs/blobs-s3",
  "version": "0.1.0",
  "type": "module",
  "main": "index.js"
}

Step 2: Create index.js

// @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

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

// 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:

Moving all handler methods from src/pds.js to packages/core/pds.js
Replacing this.sql.exec(...) calls with await this.storage.*()
Replacing this.env.BLOBS.* calls with await this.blobs.*()
Updating the route table

Step 2: Run tests after full migration

Run: npm test Expected: All tests pass

Step 3: Commit

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

{
  "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

// @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

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

{
  "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

// @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

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

// 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

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

// 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

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

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

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.

chadtmiller.com/pds.js Switch repository Loading…

PDS Core Separation Implementation Plan

Phase 1: Setup Monorepo Structure

Task 1.1: Initialize pnpm workspace

Task 1.2: Create @pdsjs/core package skeleton

Phase 2: Extract Pure Utilities to Core

Task 2.1: Extract crypto utilities

Task 2.2: Extract CBOR/repo utilities

Task 2.3: Extract auth utilities

Task 2.4: Update core index.js to re-export all utilities

Phase 3: Define Storage and Blob Ports

Task 3.1: Define StoragePort and BlobPort interfaces

Phase 4: Create Storage Adapters

Task 4.1: Create @pdsjs/storage-sqlite package

Phase 5: Create Blob Adapters

Task 5.1: Create @pdsjs/blobs-fs package

Task 5.2: Create @pdsjs/blobs-s3 package

Phase 6: Refactor Core PDS Class

Task 6.1: Create PDS class that accepts ports

Phase 7: Create Platform Adapters

Task 7.1: Create @pdsjs/cloudflare adapter

Task 7.2: Create @pdsjs/node adapter

Phase 8: Testing

Task 8.1: Create shared test scenarios

Task 8.2: Create Node E2E test file

Phase 9: Cleanup

Task 9.1: Remove old src/pds.js

Summary

chadtmiller.com/pds.js

Switch repository

Loading…