pds.js / src / pds.js

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║                                                                              ║
// ║   ██████╗ ██████╗ ███████╗   Personal Data Server                            ║
// ║   ██╔══██╗██╔══██╗██╔════╝   for AT Protocol                                 ║
// ║   ██████╔╝██║  ██║███████╗                                                   ║
// ║   ██╔═══╝ ██║  ██║╚════██║                                                   ║
// ║   ██║     ██████╔╝███████║                                                   ║
// ║   ╚═╝     ╚═════╝ ╚══════╝                                                   ║
// ║                                                                              ║
// ╠══════════════════════════════════════════════════════════════════════════════╣
// ║                                                                              ║
// ║   A single-file ATProto PDS for Cloudflare Workers + Durable Objects         ║
// ║                                                                              ║
// ║   Features:                                                                  ║
// ║     • CBOR/DAG-CBOR encoding for content-addressed data                      ║
// ║     • CID generation (CIDv1 with dag-cbor + sha-256)                         ║
// ║     • Merkle Search Tree (MST) for repository structure                      ║
// ║     • P-256 signing with low-S normalization                                 ║
// ║     • JWT authentication (access, refresh, service tokens)                   ║
// ║     • CAR file building for repo sync                                        ║
// ║     • R2 blob storage with MIME detection                                    ║
// ║     • SQLite persistence via Durable Objects                                 ║
// ║                                                                              ║
// ║   @see https://atproto.com                                                   ║
// ║                                                                              ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ TYPES & CONSTANTS                                                            ║
// ║ Environment bindings, SQL row types, protocol constants                      ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

// CBOR primitive markers (RFC 8949)
const CBOR_FALSE = 0xf4;
const CBOR_TRUE = 0xf5;
const CBOR_NULL = 0xf6;

// DAG-CBOR CID link tag
const CBOR_TAG_CID = 42;

// CID codec constants
const CODEC_DAG_CBOR = 0x71;
const CODEC_RAW = 0x55;

// TID generation constants
const TID_CHARS = '234567abcdefghijklmnopqrstuvwxyz';
let lastTimestamp = 0;
const clockId = Math.floor(Math.random() * 1024);

// P-256 curve order N (for low-S signature normalization)
const P256_N = BigInt(
  '0xFFFFFFFF00000000FFFFFFFFFFFFFFFFBCE6FAADA7179E84F3B9CAC2FC632551',
);
const P256_N_DIV_2 = P256_N / 2n;

// Crawler notification throttle
const CRAWL_NOTIFY_THRESHOLD = 20 * 60 * 1000; // 20 minutes (matches official PDS)
let lastCrawlNotify = 0;

/**
 * Cloudflare Workers environment bindings
 * @typedef {Object} Env
 * @property {string} JWT_SECRET - Secret for signing/verifying session JWTs
 * @property {string} [RELAY_HOST] - Relay host to notify of repo updates (e.g., bsky.network)
 * @property {string} [APPVIEW_URL] - AppView URL for proxying app.bsky.* requests
 * @property {string} [APPVIEW_DID] - AppView DID for service auth
 * @property {string} [PDS_PASSWORD] - Password for createSession authentication
 * @property {DurableObjectNamespace} PDS - Durable Object namespace for PDS instances
 * @property {R2Bucket} [BLOB_BUCKET] - R2 bucket for blob storage (legacy name)
 * @property {R2Bucket} [BLOBS] - R2 bucket for blob storage
 */

/**
 * Row from the `blocks` table - stores raw CBOR-encoded data blocks
 * @typedef {Object} BlockRow
 * @property {string} cid - Content ID (CIDv1 base32lower)
 * @property {ArrayBuffer} data - Raw block data (CBOR-encoded)
 */

/**
 * Row from the `records` table - indexes AT Protocol records
 * @typedef {Object} RecordRow
 * @property {string} uri - AT URI (at://did/collection/rkey)
 * @property {string} cid - Content ID of the record block
 * @property {string} collection - Collection NSID (e.g., app.bsky.feed.post)
 * @property {string} rkey - Record key within collection
 * @property {ArrayBuffer} value - CBOR-encoded record value
 */

/**
 * Row from the `commits` table - tracks repo commit history
 * @typedef {Object} CommitRow
 * @property {string} cid - Content ID of the signed commit block
 * @property {string} rev - Revision TID for ordering
 * @property {string|null} prev - Previous commit CID (null for first commit)
 */

/**
 * Row from the `seq_events` table - stores firehose events for subscribeRepos
 * @typedef {Object} SeqEventRow
 * @property {number} seq - Sequence number for cursor-based pagination
 * @property {string} did - DID of the repo that changed
 * @property {string} commit_cid - CID of the commit
 * @property {ArrayBuffer|Uint8Array} evt - CBOR-encoded event with ops, blocks, rev, time
 */

/**
 * Row from the `blob` table - tracks uploaded blob metadata
 * @typedef {Object} BlobRow
 * @property {string} cid - Content ID of the blob (raw codec)
 * @property {string} mimeType - MIME type (sniffed or from Content-Type header)
 * @property {number} size - Size in bytes
 * @property {string} createdAt - ISO timestamp of upload
 */

/**
 * Decoded JWT payload for session tokens
 * @typedef {Object} JwtPayload
 * @property {string} [scope] - Token scope (e.g., "com.atproto.access")
 * @property {string} sub - Subject DID (the authenticated user)
 * @property {string} [aud] - Audience (for refresh tokens, should match sub)
 * @property {number} [iat] - Issued-at timestamp (Unix seconds)
 * @property {number} [exp] - Expiration timestamp (Unix seconds)
 * @property {string} [jti] - Unique token identifier
 */

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ UTILITIES                                                                    ║
// ║ Error responses, byte conversion, base encoding                              ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * @param {string} error - Error code
 * @param {string} message - Error message
 * @param {number} status - HTTP status code
 * @returns {Response}
 */
function errorResponse(error, message, status) {
  return Response.json({ error, message }, { status });
}

/**
 * Convert bytes to hexadecimal string
 * @param {Uint8Array} bytes - Bytes to convert
 * @returns {string} Hex string
 */
export function bytesToHex(bytes) {
  return Array.from(bytes)
    .map((b) => b.toString(16).padStart(2, '0'))
    .join('');
}

/**
 * Convert hexadecimal string to bytes
 * @param {string} hex - Hex string
 * @returns {Uint8Array} Decoded bytes
 */
export function hexToBytes(hex) {
  const bytes = new Uint8Array(hex.length / 2);
  for (let i = 0; i < hex.length; i += 2) {
    bytes[i / 2] = parseInt(hex.substr(i, 2), 16);
  }
  return bytes;
}

/**
 * @param {Uint8Array} bytes
 * @returns {bigint}
 */
function bytesToBigInt(bytes) {
  let result = 0n;
  for (const byte of bytes) {
    result = (result << 8n) | BigInt(byte);
  }
  return result;
}

/**
 * @param {bigint} n
 * @param {number} length
 * @returns {Uint8Array}
 */
function bigIntToBytes(n, length) {
  const bytes = new Uint8Array(length);
  for (let i = length - 1; i >= 0; i--) {
    bytes[i] = Number(n & 0xffn);
    n >>= 8n;
  }
  return bytes;
}

/**
 * Encode bytes as base32lower string
 * @param {Uint8Array} bytes - Bytes to encode
 * @returns {string} Base32lower-encoded string
 */
export function base32Encode(bytes) {
  const alphabet = 'abcdefghijklmnopqrstuvwxyz234567';
  let result = '';
  let bits = 0;
  let value = 0;

  for (const byte of bytes) {
    value = (value << 8) | byte;
    bits += 8;
    while (bits >= 5) {
      bits -= 5;
      result += alphabet[(value >> bits) & 31];
    }
  }

  if (bits > 0) {
    result += alphabet[(value << (5 - bits)) & 31];
  }

  return result;
}

/**
 * Decode base32lower string to bytes
 * @param {string} str - Base32lower-encoded string
 * @returns {Uint8Array} Decoded bytes
 */
export function base32Decode(str) {
  const alphabet = 'abcdefghijklmnopqrstuvwxyz234567';
  let bits = 0;
  let value = 0;
  const output = [];

  for (const char of str) {
    const idx = alphabet.indexOf(char);
    if (idx === -1) continue;
    value = (value << 5) | idx;
    bits += 5;
    if (bits >= 8) {
      bits -= 8;
      output.push((value >> bits) & 0xff);
    }
  }

  return new Uint8Array(output);
}

/**
 * Encode bytes as base64url string (no padding)
 * @param {Uint8Array} bytes - Bytes to encode
 * @returns {string} Base64url-encoded string
 */
export function base64UrlEncode(bytes) {
  let binary = '';
  for (const byte of bytes) {
    binary += String.fromCharCode(byte);
  }
  const base64 = btoa(binary);
  return base64.replace(/\+/g, '-').replace(/\//g, '_').replace(/=/g, '');
}

/**
 * Decode base64url string to bytes
 * @param {string} str - Base64url-encoded string
 * @returns {Uint8Array} Decoded bytes
 */
export function base64UrlDecode(str) {
  const base64 = str.replace(/-/g, '+').replace(/_/g, '/');
  const pad = base64.length % 4;
  const padded = pad ? base64 + '='.repeat(4 - pad) : base64;
  const binary = atob(padded);
  const bytes = new Uint8Array(binary.length);
  for (let i = 0; i < binary.length; i++) {
    bytes[i] = binary.charCodeAt(i);
  }
  return bytes;
}

/**
 * Encode integer as unsigned varint
 * @param {number} n - Non-negative integer
 * @returns {Uint8Array} Varint-encoded bytes
 */
export function varint(n) {
  const bytes = [];
  while (n >= 0x80) {
    bytes.push((n & 0x7f) | 0x80);
    n >>>= 7;
  }
  bytes.push(n);
  return new Uint8Array(bytes);
}

// === CID WRAPPER ===
// Explicit CID type for DAG-CBOR encoding (avoids fragile heuristic detection)

class CID {
  /** @param {Uint8Array} bytes */
  constructor(bytes) {
    if (!(bytes instanceof Uint8Array)) {
      throw new Error('CID must be constructed with Uint8Array');
    }
    this.bytes = bytes;
  }
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ CBOR ENCODING                                                                ║
// ║ RFC 8949 CBOR and DAG-CBOR for content-addressed data                        ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * Encode CBOR type header (major type + length)
 * @param {number[]} parts - Array to push bytes to
 * @param {number} majorType - CBOR major type (0-7)
 * @param {number} length - Value or length to encode
 */
function encodeHead(parts, majorType, length) {
  const mt = majorType << 5;
  if (length < 24) {
    parts.push(mt | length);
  } else if (length < 256) {
    parts.push(mt | 24, length);
  } else if (length < 65536) {
    parts.push(mt | 25, length >> 8, length & 0xff);
  } else if (length < 4294967296) {
    // Use Math.floor instead of bitshift to avoid 32-bit signed integer overflow
    parts.push(
      mt | 26,
      Math.floor(length / 0x1000000) & 0xff,
      Math.floor(length / 0x10000) & 0xff,
      Math.floor(length / 0x100) & 0xff,
      length & 0xff,
    );
  }
}

/**
 * Encode a value as CBOR bytes (RFC 8949 deterministic encoding)
 * @param {*} value - Value to encode (null, boolean, number, string, Uint8Array, array, or object)
 * @returns {Uint8Array} CBOR-encoded bytes
 */
export function cborEncode(value) {
  /** @type {number[]} */
  const parts = [];

  /** @param {*} val */
  function encode(val) {
    if (val === null) {
      parts.push(CBOR_NULL);
    } else if (val === true) {
      parts.push(CBOR_TRUE);
    } else if (val === false) {
      parts.push(CBOR_FALSE);
    } else if (typeof val === 'number') {
      encodeInteger(val);
    } else if (typeof val === 'string') {
      const bytes = new TextEncoder().encode(val);
      encodeHead(parts, 3, bytes.length); // major type 3 = text string
      parts.push(...bytes);
    } else if (val instanceof Uint8Array) {
      encodeHead(parts, 2, val.length); // major type 2 = byte string
      parts.push(...val);
    } else if (Array.isArray(val)) {
      encodeHead(parts, 4, val.length); // major type 4 = array
      for (const item of val) encode(item);
    } else if (typeof val === 'object') {
      // Sort keys for deterministic encoding
      const keys = Object.keys(val).sort();
      encodeHead(parts, 5, keys.length); // major type 5 = map
      for (const key of keys) {
        encode(key);
        encode(val[key]);
      }
    }
  }

  /** @param {number} n */
  function encodeInteger(n) {
    if (n >= 0) {
      encodeHead(parts, 0, n); // major type 0 = unsigned int
    } else {
      encodeHead(parts, 1, -n - 1); // major type 1 = negative int
    }
  }

  encode(value);
  return new Uint8Array(parts);
}

/**
 * DAG-CBOR encoder that handles CIDs with tag 42
 * @param {*} value
 * @returns {Uint8Array}
 */
function cborEncodeDagCbor(value) {
  /** @type {number[]} */
  const parts = [];

  /** @param {*} val */
  function encode(val) {
    if (val === null) {
      parts.push(CBOR_NULL);
    } else if (val === true) {
      parts.push(CBOR_TRUE);
    } else if (val === false) {
      parts.push(CBOR_FALSE);
    } else if (typeof val === 'number') {
      if (Number.isInteger(val) && val >= 0) {
        encodeHead(parts, 0, val);
      } else if (Number.isInteger(val) && val < 0) {
        encodeHead(parts, 1, -val - 1);
      }
    } else if (typeof val === 'string') {
      const bytes = new TextEncoder().encode(val);
      encodeHead(parts, 3, bytes.length);
      parts.push(...bytes);
    } else if (val instanceof CID) {
      // CID links in DAG-CBOR use tag 42 + 0x00 multibase prefix
      // The 0x00 prefix indicates "identity" multibase (raw bytes)
      parts.push(0xd8, CBOR_TAG_CID);
      encodeHead(parts, 2, val.bytes.length + 1); // +1 for 0x00 prefix
      parts.push(0x00);
      parts.push(...val.bytes);
    } else if (val instanceof Uint8Array) {
      // Regular byte string
      encodeHead(parts, 2, val.length);
      parts.push(...val);
    } else if (Array.isArray(val)) {
      encodeHead(parts, 4, val.length);
      for (const item of val) encode(item);
    } else if (typeof val === 'object') {
      // DAG-CBOR: sort keys by length first, then lexicographically
      // (differs from standard CBOR which sorts lexicographically only)
      const keys = Object.keys(val).filter((k) => val[k] !== undefined);
      keys.sort((a, b) => {
        if (a.length !== b.length) return a.length - b.length;
        return a < b ? -1 : a > b ? 1 : 0;
      });
      encodeHead(parts, 5, keys.length);
      for (const key of keys) {
        const keyBytes = new TextEncoder().encode(key);
        encodeHead(parts, 3, keyBytes.length);
        parts.push(...keyBytes);
        encode(val[key]);
      }
    }
  }

  encode(value);
  return new Uint8Array(parts);
}

/**
 * Decode CBOR bytes to a JavaScript value
 * @param {Uint8Array} bytes - CBOR-encoded bytes
 * @returns {*} Decoded value
 */
export function cborDecode(bytes) {
  let offset = 0;

  /** @returns {*} */
  function read() {
    const initial = bytes[offset++];
    const major = initial >> 5;
    const info = initial & 0x1f;

    let length = info;
    if (info === 24) length = bytes[offset++];
    else if (info === 25) {
      length = (bytes[offset++] << 8) | bytes[offset++];
    } else if (info === 26) {
      // Use multiplication instead of bitshift to avoid 32-bit signed integer overflow
      length =
        bytes[offset++] * 0x1000000 +
        bytes[offset++] * 0x10000 +
        bytes[offset++] * 0x100 +
        bytes[offset++];
    }

    switch (major) {
      case 0:
        return length; // unsigned int
      case 1:
        return -1 - length; // negative int
      case 2: {
        // byte string
        const data = bytes.slice(offset, offset + length);
        offset += length;
        return data;
      }
      case 3: {
        // text string
        const data = new TextDecoder().decode(
          bytes.slice(offset, offset + length),
        );
        offset += length;
        return data;
      }
      case 4: {
        // array
        const arr = [];
        for (let i = 0; i < length; i++) arr.push(read());
        return arr;
      }
      case 5: {
        // map
        /** @type {Record<string, *>} */
        const obj = {};
        for (let i = 0; i < length; i++) {
          const key = /** @type {string} */ (read());
          obj[key] = read();
        }
        return obj;
      }
      case 6: {
        // tag
        // length is the tag number
        const taggedValue = read();
        if (length === CBOR_TAG_CID) {
          // CID link: byte string with 0x00 multibase prefix, return raw CID bytes
          return taggedValue.slice(1); // strip 0x00 prefix
        }
        return taggedValue;
      }
      case 7: {
        // special
        if (info === 20) return false;
        if (info === 21) return true;
        if (info === 22) return null;
        return undefined;
      }
    }
  }

  return read();
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ CONTENT IDENTIFIERS                                                          ║
// ║ CIDs (content hashes) and TIDs (timestamp IDs)                               ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * Create a CIDv1 with SHA-256 hash
 * @param {Uint8Array} bytes - Content to hash
 * @param {number} codec - Codec identifier (0x71 for dag-cbor, 0x55 for raw)
 * @returns {Promise<Uint8Array>} CID bytes (36 bytes: version + codec + multihash)
 */
async function createCidWithCodec(bytes, codec) {
  const hash = await crypto.subtle.digest(
    'SHA-256',
    /** @type {BufferSource} */ (bytes),
  );
  const hashBytes = new Uint8Array(hash);

  // CIDv1: version(1) + codec + multihash(sha256)
  // Multihash: hash-type(0x12) + length(0x20=32) + digest
  const cid = new Uint8Array(2 + 2 + 32);
  cid[0] = 0x01; // CIDv1
  cid[1] = codec;
  cid[2] = 0x12; // sha-256
  cid[3] = 0x20; // 32 bytes
  cid.set(hashBytes, 4);

  return cid;
}

/**
 * Create CID for DAG-CBOR encoded data (records, commits)
 * @param {Uint8Array} bytes - DAG-CBOR encoded content
 * @returns {Promise<Uint8Array>} CID bytes
 */
export async function createCid(bytes) {
  return createCidWithCodec(bytes, CODEC_DAG_CBOR);
}

/**
 * Create CID for raw blob data (images, videos)
 * @param {Uint8Array} bytes - Raw binary content
 * @returns {Promise<Uint8Array>} CID bytes
 */
export async function createBlobCid(bytes) {
  return createCidWithCodec(bytes, CODEC_RAW);
}

/**
 * Convert CID bytes to base32lower string representation
 * @param {Uint8Array} cid - CID bytes
 * @returns {string} Base32lower-encoded CID with 'b' prefix
 */
export function cidToString(cid) {
  // base32lower encoding for CIDv1
  return `b${base32Encode(cid)}`;
}

/**
 * Convert base32lower CID string to raw bytes
 * @param {string} cidStr - CID string with 'b' prefix
 * @returns {Uint8Array} CID bytes
 */
export function cidToBytes(cidStr) {
  // Decode base32lower CID string to bytes
  if (!cidStr.startsWith('b')) throw new Error('expected base32lower CID');
  return base32Decode(cidStr.slice(1));
}

/**
 * Generate a timestamp-based ID (TID) for record keys
 * Monotonic within a process, sortable by time
 * @returns {string} 13-character base32-sort encoded TID
 */
export function createTid() {
  let timestamp = Date.now() * 1000; // microseconds

  // Ensure monotonic
  if (timestamp <= lastTimestamp) {
    timestamp = lastTimestamp + 1;
  }
  lastTimestamp = timestamp;

  // 13 chars: 11 for timestamp (64 bits but only ~53 used), 2 for clock ID
  let tid = '';

  // Encode timestamp (high bits first for sortability)
  let ts = timestamp;
  for (let i = 0; i < 11; i++) {
    tid = TID_CHARS[ts & 31] + tid;
    ts = Math.floor(ts / 32);
  }

  // Append clock ID (2 chars)
  tid += TID_CHARS[(clockId >> 5) & 31];
  tid += TID_CHARS[clockId & 31];

  return tid;
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ CRYPTOGRAPHY                                                                 ║
// ║ P-256 signing with low-S normalization, key management                       ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * @param {BufferSource} data
 * @returns {Promise<Uint8Array>}
 */
async function sha256(data) {
  const hash = await crypto.subtle.digest('SHA-256', data);
  return new Uint8Array(hash);
}

/**
 * Import a raw P-256 private key for signing
 * @param {Uint8Array} privateKeyBytes - 32-byte raw private key
 * @returns {Promise<CryptoKey>} Web Crypto key handle
 */
export async function importPrivateKey(privateKeyBytes) {
  // Validate private key length (P-256 requires exactly 32 bytes)
  if (
    !(privateKeyBytes instanceof Uint8Array) ||
    privateKeyBytes.length !== 32
  ) {
    throw new Error(
      `Invalid private key: expected 32 bytes, got ${privateKeyBytes?.length ?? 'non-Uint8Array'}`,
    );
  }

  // PKCS#8 wrapper for raw P-256 private key
  const pkcs8Prefix = new Uint8Array([
    0x30, 0x41, 0x02, 0x01, 0x00, 0x30, 0x13, 0x06, 0x07, 0x2a, 0x86, 0x48,
    0xce, 0x3d, 0x02, 0x01, 0x06, 0x08, 0x2a, 0x86, 0x48, 0xce, 0x3d, 0x03,
    0x01, 0x07, 0x04, 0x27, 0x30, 0x25, 0x02, 0x01, 0x01, 0x04, 0x20,
  ]);

  const pkcs8 = new Uint8Array(pkcs8Prefix.length + 32);
  pkcs8.set(pkcs8Prefix);
  pkcs8.set(privateKeyBytes, pkcs8Prefix.length);

  return crypto.subtle.importKey(
    'pkcs8',
    /** @type {BufferSource} */ (pkcs8),
    { name: 'ECDSA', namedCurve: 'P-256' },
    false,
    ['sign'],
  );
}

/**
 * Sign data with ECDSA P-256, returning low-S normalized signature
 * @param {CryptoKey} privateKey - Web Crypto key from importPrivateKey
 * @param {Uint8Array} data - Data to sign
 * @returns {Promise<Uint8Array>} 64-byte signature (r || s)
 */
export async function sign(privateKey, data) {
  const signature = await crypto.subtle.sign(
    { name: 'ECDSA', hash: 'SHA-256' },
    privateKey,
    /** @type {BufferSource} */ (data),
  );
  const sig = new Uint8Array(signature);

  const r = sig.slice(0, 32);
  const s = sig.slice(32, 64);
  const sBigInt = bytesToBigInt(s);

  // Low-S normalization: Bitcoin/ATProto require S <= N/2 to prevent
  // signature malleability (two valid signatures for same message)
  if (sBigInt > P256_N_DIV_2) {
    const newS = P256_N - sBigInt;
    const newSBytes = bigIntToBytes(newS, 32);
    const normalized = new Uint8Array(64);
    normalized.set(r, 0);
    normalized.set(newSBytes, 32);
    return normalized;
  }

  return sig;
}

/**
 * Generate a new P-256 key pair
 * @returns {Promise<{privateKey: Uint8Array, publicKey: Uint8Array}>} 32-byte private key, 33-byte compressed public key
 */
export async function generateKeyPair() {
  const keyPair = await crypto.subtle.generateKey(
    { name: 'ECDSA', namedCurve: 'P-256' },
    true,
    ['sign', 'verify'],
  );

  // Export private key as raw bytes
  const privateJwk = await crypto.subtle.exportKey('jwk', keyPair.privateKey);
  const privateBytes = base64UrlDecode(/** @type {string} */ (privateJwk.d));

  // Export public key as compressed point
  const publicRaw = await crypto.subtle.exportKey('raw', keyPair.publicKey);
  const publicBytes = new Uint8Array(publicRaw);
  const compressed = compressPublicKey(publicBytes);

  return { privateKey: privateBytes, publicKey: compressed };
}

/**
 * @param {Uint8Array} uncompressed
 * @returns {Uint8Array}
 */
function compressPublicKey(uncompressed) {
  // uncompressed is 65 bytes: 0x04 + x(32) + y(32)
  // compressed is 33 bytes: prefix(02 or 03) + x(32)
  const x = uncompressed.slice(1, 33);
  const y = uncompressed.slice(33, 65);
  const prefix = (y[31] & 1) === 0 ? 0x02 : 0x03;
  const compressed = new Uint8Array(33);
  compressed[0] = prefix;
  compressed.set(x, 1);
  return compressed;
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ AUTHENTICATION                                                               ║
// ║ JWT creation/verification for sessions and service auth                      ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * Create HMAC-SHA256 signature for JWT
 * @param {string} data - Data to sign (header.payload)
 * @param {string} secret - Secret key
 * @returns {Promise<string>} Base64url-encoded signature
 */
async function hmacSign(data, secret) {
  const key = await crypto.subtle.importKey(
    'raw',
    /** @type {BufferSource} */ (new TextEncoder().encode(secret)),
    { name: 'HMAC', hash: 'SHA-256' },
    false,
    ['sign'],
  );
  const sig = await crypto.subtle.sign(
    'HMAC',
    key,
    /** @type {BufferSource} */ (new TextEncoder().encode(data)),
  );
  return base64UrlEncode(new Uint8Array(sig));
}

/**
 * Create an access JWT for ATProto
 * @param {string} did - User's DID (subject and audience)
 * @param {string} secret - JWT signing secret
 * @param {number} [expiresIn=7200] - Expiration in seconds (default 2 hours)
 * @returns {Promise<string>} Signed JWT
 */
export async function createAccessJwt(did, secret, expiresIn = 7200) {
  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 + expiresIn,
  };

  const headerB64 = base64UrlEncode(
    new TextEncoder().encode(JSON.stringify(header)),
  );
  const payloadB64 = base64UrlEncode(
    new TextEncoder().encode(JSON.stringify(payload)),
  );
  const signature = await hmacSign(`${headerB64}.${payloadB64}`, secret);

  return `${headerB64}.${payloadB64}.${signature}`;
}

/**
 * Create a refresh JWT for ATProto
 * @param {string} did - User's DID (subject and audience)
 * @param {string} secret - JWT signing secret
 * @param {number} [expiresIn=86400] - Expiration in seconds (default 24 hours)
 * @returns {Promise<string>} Signed JWT
 */
export async function createRefreshJwt(did, secret, expiresIn = 86400) {
  const header = { typ: 'refresh+jwt', alg: 'HS256' };
  const now = Math.floor(Date.now() / 1000);
  // Generate random jti (token ID)
  const jtiBytes = new Uint8Array(32);
  crypto.getRandomValues(jtiBytes);
  const jti = base64UrlEncode(jtiBytes);

  const payload = {
    scope: 'com.atproto.refresh',
    sub: did,
    aud: did,
    jti,
    iat: now,
    exp: now + expiresIn,
  };

  const headerB64 = base64UrlEncode(
    new TextEncoder().encode(JSON.stringify(header)),
  );
  const payloadB64 = base64UrlEncode(
    new TextEncoder().encode(JSON.stringify(payload)),
  );
  const signature = await hmacSign(`${headerB64}.${payloadB64}`, secret);

  return `${headerB64}.${payloadB64}.${signature}`;
}

/**
 * Verify and decode a JWT (shared logic)
 * @param {string} jwt - JWT string to verify
 * @param {string} secret - JWT signing secret
 * @param {string} expectedType - Expected token type (e.g., 'at+jwt', 'refresh+jwt')
 * @returns {Promise<{header: {typ: string, alg: string}, payload: JwtPayload}>} Decoded header and payload
 * @throws {Error} If token is invalid, expired, or wrong type
 */
async function verifyJwt(jwt, secret, expectedType) {
  const parts = jwt.split('.');
  if (parts.length !== 3) {
    throw new Error('Invalid JWT format');
  }

  const [headerB64, payloadB64, signatureB64] = parts;

  // Verify signature
  const expectedSig = await hmacSign(`${headerB64}.${payloadB64}`, secret);
  if (signatureB64 !== expectedSig) {
    throw new Error('Invalid signature');
  }

  // Decode header and payload
  const header = JSON.parse(
    new TextDecoder().decode(base64UrlDecode(headerB64)),
  );
  const payload = JSON.parse(
    new TextDecoder().decode(base64UrlDecode(payloadB64)),
  );

  // Check token type
  if (header.typ !== expectedType) {
    throw new Error(`Invalid token type: expected ${expectedType}`);
  }

  // Check expiration
  const now = Math.floor(Date.now() / 1000);
  if (payload.exp && payload.exp < now) {
    throw new Error('Token expired');
  }

  return { header, payload };
}

/**
 * Verify and decode an access JWT
 * @param {string} jwt - JWT string to verify
 * @param {string} secret - JWT signing secret
 * @returns {Promise<JwtPayload>} Decoded payload
 * @throws {Error} If token is invalid, expired, or wrong type
 */
export async function verifyAccessJwt(jwt, secret) {
  const { payload } = await verifyJwt(jwt, secret, 'at+jwt');
  return payload;
}

/**
 * Verify and decode a refresh JWT
 * @param {string} jwt - JWT string to verify
 * @param {string} secret - JWT signing secret
 * @returns {Promise<JwtPayload>} Decoded payload
 * @throws {Error} If token is invalid, expired, or wrong type
 */
export async function verifyRefreshJwt(jwt, secret) {
  const { payload } = await verifyJwt(jwt, secret, 'refresh+jwt');

  // Validate audience matches subject (token intended for this user)
  if (payload.aud && payload.aud !== payload.sub) {
    throw new Error('Invalid audience');
  }

  return payload;
}

/**
 * Create a service auth JWT signed with ES256 (P-256)
 * Used for proxying requests to AppView
 * @param {Object} params - JWT parameters
 * @param {string} params.iss - Issuer DID (PDS DID)
 * @param {string} params.aud - Audience DID (AppView DID)
 * @param {string|null} params.lxm - Lexicon method being called
 * @param {CryptoKey} params.signingKey - P-256 private key from importPrivateKey
 * @returns {Promise<string>} Signed JWT
 */
export async function createServiceJwt({ iss, aud, lxm, signingKey }) {
  const header = { typ: 'JWT', alg: 'ES256' };
  const now = Math.floor(Date.now() / 1000);

  // Generate random jti
  const jtiBytes = new Uint8Array(16);
  crypto.getRandomValues(jtiBytes);
  const jti = bytesToHex(jtiBytes);

  /** @type {{ iss: string, aud: string, exp: number, iat: number, jti: string, lxm?: string }} */
  const payload = {
    iss,
    aud,
    exp: now + 60, // 1 minute expiration
    iat: now,
    jti,
  };
  if (lxm) payload.lxm = lxm;

  const headerB64 = base64UrlEncode(
    new TextEncoder().encode(JSON.stringify(header)),
  );
  const payloadB64 = base64UrlEncode(
    new TextEncoder().encode(JSON.stringify(payload)),
  );
  const toSign = new TextEncoder().encode(`${headerB64}.${payloadB64}`);

  const sig = await sign(signingKey, toSign);
  const sigB64 = base64UrlEncode(sig);

  return `${headerB64}.${payloadB64}.${sigB64}`;
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ MERKLE SEARCH TREE                                                           ║
// ║ MST for ATProto repository structure                                         ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

// Cache for key depths (SHA-256 is expensive)
const keyDepthCache = new Map();

/**
 * Get MST tree depth for a key based on leading zeros in SHA-256 hash
 * @param {string} key - Record key (collection/rkey)
 * @returns {Promise<number>} Tree depth (leading zeros / 2)
 */
export async function getKeyDepth(key) {
  // Count leading zeros in SHA-256 hash, divide by 2
  if (keyDepthCache.has(key)) return keyDepthCache.get(key);

  const keyBytes = new TextEncoder().encode(key);
  const hash = await sha256(keyBytes);

  let zeros = 0;
  for (const byte of hash) {
    if (byte === 0) {
      zeros += 8;
    } else {
      // Count leading zeros in this byte
      for (let i = 7; i >= 0; i--) {
        if ((byte >> i) & 1) break;
        zeros++;
      }
      break;
    }
  }

  // MST depth = leading zeros in SHA-256 hash / 2
  // This creates a probabilistic tree where ~50% of keys are at depth 0,
  // ~25% at depth 1, etc., giving O(log n) lookups
  const depth = Math.floor(zeros / 2);
  keyDepthCache.set(key, depth);
  return depth;
}

/**
 * Compute common prefix length between two byte arrays
 * @param {Uint8Array} a
 * @param {Uint8Array} b
 * @returns {number}
 */
function commonPrefixLen(a, b) {
  const minLen = Math.min(a.length, b.length);
  for (let i = 0; i < minLen; i++) {
    if (a[i] !== b[i]) return i;
  }
  return minLen;
}

class MST {
  /** @param {SqlStorage} sql */
  constructor(sql) {
    this.sql = sql;
  }

  async computeRoot() {
    const records = this.sql
      .exec(`
      SELECT collection, rkey, cid FROM records ORDER BY collection, rkey
    `)
      .toArray();

    if (records.length === 0) {
      return null;
    }

    // Build entries with pre-computed depths (heights)
    // In ATProto MST, "height" determines which layer a key belongs to
    // Layer 0 is at the BOTTOM, root is at the highest layer
    const entries = [];
    let maxDepth = 0;
    for (const r of records) {
      const key = `${r.collection}/${r.rkey}`;
      const depth = await getKeyDepth(key);
      maxDepth = Math.max(maxDepth, depth);
      entries.push({
        key,
        keyBytes: new TextEncoder().encode(key),
        cid: /** @type {string} */ (r.cid),
        depth,
      });
    }

    // Start building from the root (highest layer) going down to layer 0
    return this.buildTree(entries, maxDepth);
  }

  /**
   * @param {Array<{key: string, keyBytes: Uint8Array, cid: string, depth: number}>} entries
   * @param {number} layer
   * @returns {Promise<string|null>}
   */
  async buildTree(entries, layer) {
    if (entries.length === 0) return null;

    // Separate entries for this layer vs lower layers (subtrees)
    // Keys with depth == layer stay at this node
    // Keys with depth < layer go into subtrees (going down toward layer 0)
    /** @type {Array<{type: 'subtree', cid: string|null} | {type: 'entry', entry: {key: string, keyBytes: Uint8Array, cid: string, depth: number}}>} */
    const thisLayer = [];
    /** @type {Array<{key: string, keyBytes: Uint8Array, cid: string, depth: number}>} */
    let leftSubtree = [];

    for (const entry of entries) {
      if (entry.depth < layer) {
        // This entry belongs to a lower layer - accumulate for subtree
        leftSubtree.push(entry);
      } else {
        // This entry belongs at current layer (depth == layer)
        // Process accumulated left subtree first
        if (leftSubtree.length > 0) {
          const leftCid = await this.buildTree(leftSubtree, layer - 1);
          thisLayer.push({ type: 'subtree', cid: leftCid });
          leftSubtree = [];
        }
        thisLayer.push({ type: 'entry', entry });
      }
    }

    // Handle remaining left subtree
    if (leftSubtree.length > 0) {
      const leftCid = await this.buildTree(leftSubtree, layer - 1);
      thisLayer.push({ type: 'subtree', cid: leftCid });
    }

    // Build node with proper ATProto format
    /** @type {{ e: Array<{p: number, k: Uint8Array, v: CID, t: CID|null}>, l?: CID|null }} */
    const node = { e: [] };
    /** @type {string|null} */
    let leftCid = null;
    let prevKeyBytes = new Uint8Array(0);

    for (let i = 0; i < thisLayer.length; i++) {
      const item = thisLayer[i];

      if (item.type === 'subtree') {
        if (node.e.length === 0) {
          leftCid = item.cid;
        } else {
          // Attach to previous entry's 't' field
          if (item.cid !== null) {
            node.e[node.e.length - 1].t = new CID(cidToBytes(item.cid));
          }
        }
      } else {
        // Entry - compute prefix compression
        const keyBytes = item.entry.keyBytes;
        const prefixLen = commonPrefixLen(prevKeyBytes, keyBytes);
        const keySuffix = keyBytes.slice(prefixLen);

        // ATProto requires t field to be present (can be null)
        const e = {
          p: prefixLen,
          k: keySuffix,
          v: new CID(cidToBytes(item.entry.cid)),
          t: null, // Will be updated if there's a subtree
        };

        node.e.push(e);
        prevKeyBytes = /** @type {Uint8Array<ArrayBuffer>} */ (keyBytes);
      }
    }

    // ATProto requires l field to be present (can be null)
    node.l = leftCid ? new CID(cidToBytes(leftCid)) : null;

    // Encode node with proper MST CBOR format
    const nodeBytes = cborEncodeDagCbor(node);
    const nodeCid = await createCid(nodeBytes);
    const cidStr = cidToString(nodeCid);

    this.sql.exec(
      `INSERT OR REPLACE INTO blocks (cid, data) VALUES (?, ?)`,
      cidStr,
      nodeBytes,
    );

    return cidStr;
  }
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ CAR FILES                                                                    ║
// ║ Content Addressable aRchive format for repo sync                             ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * Build a CAR (Content Addressable aRchive) file
 * @param {string} rootCid - Root CID string
 * @param {Array<{cid: string, data: Uint8Array}>} blocks - Blocks to include
 * @returns {Uint8Array} CAR file bytes
 */
export function buildCarFile(rootCid, blocks) {
  const parts = [];

  // Header: { version: 1, roots: [rootCid] }
  const rootCidBytes = cidToBytes(rootCid);
  const header = cborEncodeDagCbor({
    version: 1,
    roots: [new CID(rootCidBytes)],
  });
  parts.push(varint(header.length));
  parts.push(header);

  // Blocks: varint(len) + cid + data
  for (const block of blocks) {
    const cidBytes = cidToBytes(block.cid);
    const blockLen = cidBytes.length + block.data.length;
    parts.push(varint(blockLen));
    parts.push(cidBytes);
    parts.push(block.data);
  }

  // Concatenate all parts
  const totalLen = parts.reduce((sum, p) => sum + p.length, 0);
  const car = new Uint8Array(totalLen);
  let offset = 0;
  for (const part of parts) {
    car.set(part, offset);
    offset += part.length;
  }

  return car;
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ BLOB HANDLING                                                                ║
// ║ MIME detection, blob reference scanning                                      ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * Sniff MIME type from file magic bytes
 * @param {Uint8Array|ArrayBuffer} bytes - File bytes (only first 12 needed)
 * @returns {string|null} Detected MIME type or null if unknown
 */
export function sniffMimeType(bytes) {
  const arr = new Uint8Array(bytes.slice(0, 12));

  // JPEG: FF D8 FF
  if (arr[0] === 0xff && arr[1] === 0xd8 && arr[2] === 0xff) {
    return 'image/jpeg';
  }

  // PNG: 89 50 4E 47 0D 0A 1A 0A
  if (
    arr[0] === 0x89 &&
    arr[1] === 0x50 &&
    arr[2] === 0x4e &&
    arr[3] === 0x47 &&
    arr[4] === 0x0d &&
    arr[5] === 0x0a &&
    arr[6] === 0x1a &&
    arr[7] === 0x0a
  ) {
    return 'image/png';
  }

  // GIF: 47 49 46 38 (GIF8)
  if (
    arr[0] === 0x47 &&
    arr[1] === 0x49 &&
    arr[2] === 0x46 &&
    arr[3] === 0x38
  ) {
    return 'image/gif';
  }

  // WebP: RIFF....WEBP
  if (
    arr[0] === 0x52 &&
    arr[1] === 0x49 &&
    arr[2] === 0x46 &&
    arr[3] === 0x46 &&
    arr[8] === 0x57 &&
    arr[9] === 0x45 &&
    arr[10] === 0x42 &&
    arr[11] === 0x50
  ) {
    return 'image/webp';
  }

  // ISOBMFF container: ....ftyp at byte 4 (MP4, AVIF, HEIC, etc.)
  if (
    arr[4] === 0x66 &&
    arr[5] === 0x74 &&
    arr[6] === 0x79 &&
    arr[7] === 0x70
  ) {
    // Check brand code at bytes 8-11
    const brand = String.fromCharCode(arr[8], arr[9], arr[10], arr[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 {*} obj - Record value to scan
 * @param {string[]} refs - Accumulator array (internal)
 * @returns {string[]} Array of blob CID strings
 */
export function findBlobRefs(obj, refs = []) {
  if (!obj || typeof obj !== 'object') {
    return refs;
  }

  // Check if this object is a blob ref
  if (obj.$type === 'blob' && obj.ref?.$link) {
    refs.push(obj.ref.$link);
  }

  // Recurse into arrays and objects
  if (Array.isArray(obj)) {
    for (const item of obj) {
      findBlobRefs(item, refs);
    }
  } else {
    for (const value of Object.values(obj)) {
      findBlobRefs(value, refs);
    }
  }

  return refs;
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ RELAY NOTIFICATION                                                           ║
// ║ Notify relays to crawl after repo updates                                    ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * Notify relays to come crawl us after writes (like official PDS)
 * @param {{ RELAY_HOST?: string }} env
 * @param {string} hostname
 */
async function notifyCrawlers(env, hostname) {
  const now = Date.now();
  if (now - lastCrawlNotify < CRAWL_NOTIFY_THRESHOLD) {
    return; // Throttle notifications
  }

  const relayHost = env.RELAY_HOST;
  if (!relayHost) return;

  lastCrawlNotify = now;

  // Fire and forget - don't block writes on relay notification
  fetch(`${relayHost}/xrpc/com.atproto.sync.requestCrawl`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ hostname }),
  }).catch(() => {
    // Silently ignore relay notification failures
  });
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ ROUTING                                                                      ║
// ║ XRPC endpoint definitions                                                    ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

/**
 * Route handler function type
 * @callback RouteHandler
 * @param {PersonalDataServer} pds - PDS instance
 * @param {Request} request - HTTP request
 * @param {URL} url - Parsed URL
 * @returns {Response | Promise<Response>} HTTP response
 */

/**
 * Route definition for the PDS router
 * @typedef {Object} Route
 * @property {string} [method] - Required HTTP method (default: any)
 * @property {RouteHandler} handler - Handler function
 */

/** @type {Record<string, Route>} */
const pdsRoutes = {
  '/.well-known/atproto-did': {
    handler: (pds, _req, _url) => pds.handleAtprotoDid(),
  },
  '/init': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleInit(req),
  },
  '/status': {
    handler: (pds, _req, _url) => pds.handleStatus(),
  },
  '/forward-event': {
    handler: (pds, req, _url) => pds.handleForwardEvent(req),
  },
  '/register-did': {
    handler: (pds, req, _url) => pds.handleRegisterDid(req),
  },
  '/get-registered-dids': {
    handler: (pds, _req, _url) => pds.handleGetRegisteredDids(),
  },
  '/register-handle': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleRegisterHandle(req),
  },
  '/resolve-handle': {
    handler: (pds, _req, url) => pds.handleResolveHandle(url),
  },
  '/repo-info': {
    handler: (pds, _req, _url) => pds.handleRepoInfo(),
  },
  '/xrpc/com.atproto.server.describeServer': {
    handler: (pds, req, _url) => pds.handleDescribeServer(req),
  },
  '/xrpc/com.atproto.server.createSession': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleCreateSession(req),
  },
  '/xrpc/com.atproto.server.getSession': {
    handler: (pds, req, _url) => pds.handleGetSession(req),
  },
  '/xrpc/com.atproto.server.refreshSession': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleRefreshSession(req),
  },
  '/xrpc/app.bsky.actor.getPreferences': {
    handler: (pds, req, _url) => pds.handleGetPreferences(req),
  },
  '/xrpc/app.bsky.actor.putPreferences': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handlePutPreferences(req),
  },
  '/xrpc/com.atproto.sync.listRepos': {
    handler: (pds, _req, _url) => pds.handleListRepos(),
  },
  '/xrpc/com.atproto.repo.createRecord': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleCreateRecord(req),
  },
  '/xrpc/com.atproto.repo.deleteRecord': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleDeleteRecord(req),
  },
  '/xrpc/com.atproto.repo.putRecord': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handlePutRecord(req),
  },
  '/xrpc/com.atproto.repo.applyWrites': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleApplyWrites(req),
  },
  '/xrpc/com.atproto.repo.getRecord': {
    handler: (pds, _req, url) => pds.handleGetRecord(url),
  },
  '/xrpc/com.atproto.repo.describeRepo': {
    handler: (pds, _req, _url) => pds.handleDescribeRepo(),
  },
  '/xrpc/com.atproto.repo.listRecords': {
    handler: (pds, _req, url) => pds.handleListRecords(url),
  },
  '/xrpc/com.atproto.repo.uploadBlob': {
    method: 'POST',
    handler: (pds, req, _url) => pds.handleUploadBlob(req),
  },
  '/xrpc/com.atproto.sync.getLatestCommit': {
    handler: (pds, _req, _url) => pds.handleGetLatestCommit(),
  },
  '/xrpc/com.atproto.sync.getRepoStatus': {
    handler: (pds, _req, _url) => pds.handleGetRepoStatus(),
  },
  '/xrpc/com.atproto.sync.getRepo': {
    handler: (pds, _req, _url) => pds.handleGetRepo(),
  },
  '/xrpc/com.atproto.sync.getRecord': {
    handler: (pds, _req, url) => pds.handleSyncGetRecord(url),
  },
  '/xrpc/com.atproto.sync.getBlob': {
    handler: (pds, _req, url) => pds.handleGetBlob(url),
  },
  '/xrpc/com.atproto.sync.listBlobs': {
    handler: (pds, _req, url) => pds.handleListBlobs(url),
  },
  '/xrpc/com.atproto.sync.subscribeRepos': {
    handler: (pds, req, url) => pds.handleSubscribeRepos(req, url),
  },
};

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ PERSONAL DATA SERVER                                                         ║
// ║ Durable Object class implementing ATProto PDS                                ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

export class PersonalDataServer {
  /** @type {string | undefined} */
  _did;

  /**
   * @param {DurableObjectState} state
   * @param {Env} env
   */
  constructor(state, env) {
    this.state = state;
    this.sql = state.storage.sql;
    this.env = env;

    // Initialize schema
    this.sql.exec(`
      CREATE TABLE IF NOT EXISTS blocks (
        cid TEXT PRIMARY KEY,
        data BLOB NOT NULL
      );

      CREATE TABLE IF NOT EXISTS records (
        uri TEXT PRIMARY KEY,
        cid TEXT NOT NULL,
        collection TEXT NOT NULL,
        rkey TEXT NOT NULL,
        value BLOB NOT NULL
      );

      CREATE TABLE IF NOT EXISTS commits (
        seq INTEGER PRIMARY KEY AUTOINCREMENT,
        cid TEXT NOT NULL,
        rev TEXT NOT NULL,
        prev TEXT
      );

      CREATE TABLE IF NOT EXISTS seq_events (
        seq INTEGER PRIMARY KEY AUTOINCREMENT,
        did TEXT NOT NULL,
        commit_cid TEXT NOT NULL,
        evt BLOB NOT NULL
      );

      CREATE TABLE IF NOT EXISTS blob (
        cid TEXT PRIMARY KEY,
        mimeType TEXT NOT NULL,
        size INTEGER NOT NULL,
        createdAt TEXT NOT NULL
      );

      CREATE TABLE IF NOT EXISTS record_blob (
        blobCid TEXT NOT NULL,
        recordUri TEXT NOT NULL,
        PRIMARY KEY (blobCid, recordUri)
      );

      CREATE INDEX IF NOT EXISTS idx_record_blob_uri ON record_blob(recordUri);

      CREATE INDEX IF NOT EXISTS idx_records_collection ON records(collection, rkey);
    `);
  }

  /**
   * @param {string} did
   * @param {string} privateKeyHex
   * @param {string|null} [handle]
   */
  async initIdentity(did, privateKeyHex, handle = null) {
    await this.state.storage.put('did', did);
    await this.state.storage.put('privateKey', privateKeyHex);
    if (handle) {
      await this.state.storage.put('handle', handle);
    }

    // Schedule blob cleanup alarm (runs daily)
    const currentAlarm = await this.state.storage.getAlarm();
    if (!currentAlarm) {
      await this.state.storage.setAlarm(Date.now() + 24 * 60 * 60 * 1000);
    }
  }

  async getDid() {
    if (!this._did) {
      this._did = await this.state.storage.get('did');
    }
    return this._did;
  }

  async getHandle() {
    return this.state.storage.get('handle');
  }

  async getSigningKey() {
    const hex = await this.state.storage.get('privateKey');
    if (!hex) return null;
    return importPrivateKey(hexToBytes(/** @type {string} */ (hex)));
  }

  /**
   * Collect MST node blocks for a given root CID
   * @param {string} rootCidStr
   * @returns {Array<{cid: string, data: Uint8Array}>}
   */
  collectMstBlocks(rootCidStr) {
    /** @type {Array<{cid: string, data: Uint8Array}>} */
    const blocks = [];
    const visited = new Set();

    /** @param {string} cidStr */
    const collect = (cidStr) => {
      if (visited.has(cidStr)) return;
      visited.add(cidStr);

      const rows = /** @type {BlockRow[]} */ (
        this.sql.exec(`SELECT data FROM blocks WHERE cid = ?`, cidStr).toArray()
      );
      if (rows.length === 0) return;

      const data = new Uint8Array(rows[0].data);
      blocks.push({ cid: cidStr, data }); // Keep as string, buildCarFile will convert

      // Decode and follow child CIDs (MST nodes have 'l' and 'e' with 't' subtrees)
      try {
        const node = cborDecode(data);
        if (node.l) collect(cidToString(node.l));
        if (node.e) {
          for (const entry of node.e) {
            if (entry.t) collect(cidToString(entry.t));
          }
        }
      } catch (_e) {
        // Not an MST node, ignore
      }
    };

    collect(rootCidStr);
    return blocks;
  }

  /**
   * @param {string} collection
   * @param {Record<string, *>} record
   * @param {string|null} [rkey]
   * @returns {Promise<{uri: string, cid: string, commit: string}>}
   */
  async createRecord(collection, record, rkey = null) {
    const did = await this.getDid();
    if (!did) throw new Error('PDS not initialized');

    rkey = rkey || createTid();
    const uri = `at://${did}/${collection}/${rkey}`;

    // Encode and hash record (must use DAG-CBOR for proper key ordering)
    const recordBytes = cborEncodeDagCbor(record);
    const recordCid = await createCid(recordBytes);
    const recordCidStr = cidToString(recordCid);

    // Store block
    this.sql.exec(
      `INSERT OR REPLACE INTO blocks (cid, data) VALUES (?, ?)`,
      recordCidStr,
      recordBytes,
    );

    // Store record index
    this.sql.exec(
      `INSERT OR REPLACE INTO records (uri, cid, collection, rkey, value) VALUES (?, ?, ?, ?, ?)`,
      uri,
      recordCidStr,
      collection,
      rkey,
      recordBytes,
    );

    // Associate blobs with this record (delete old associations first for updates)
    this.sql.exec('DELETE FROM record_blob WHERE recordUri = ?', uri);

    const blobRefs = findBlobRefs(record);
    for (const blobCid of blobRefs) {
      // Verify blob exists
      const blobExists = this.sql
        .exec('SELECT cid FROM blob WHERE cid = ?', blobCid)
        .toArray();

      if (blobExists.length === 0) {
        throw new Error(`BlobNotFound: ${blobCid}`);
      }

      // Create association
      this.sql.exec(
        'INSERT INTO record_blob (blobCid, recordUri) VALUES (?, ?)',
        blobCid,
        uri,
      );
    }

    // Rebuild MST
    const mst = new MST(this.sql);
    const dataRoot = await mst.computeRoot();

    // Get previous commit
    const prevCommits = this.sql
      .exec(`SELECT cid, rev FROM commits ORDER BY seq DESC LIMIT 1`)
      .toArray();
    const prevCommit = prevCommits.length > 0 ? prevCommits[0] : null;

    // Create commit
    const rev = createTid();
    // Build commit with CIDs wrapped in CID class (for dag-cbor tag 42 encoding)
    const commit = {
      did,
      version: 3,
      data: new CID(cidToBytes(/** @type {string} */ (dataRoot))), // CID wrapped for explicit encoding
      rev,
      prev: prevCommit?.cid
        ? new CID(cidToBytes(/** @type {string} */ (prevCommit.cid)))
        : null,
    };

    // Sign commit (using dag-cbor encoder for CIDs)
    const commitBytes = cborEncodeDagCbor(commit);
    const signingKey = await this.getSigningKey();
    if (!signingKey) throw new Error('No signing key');
    const sig = await sign(signingKey, commitBytes);

    const signedCommit = { ...commit, sig };
    const signedBytes = cborEncodeDagCbor(signedCommit);
    const commitCid = await createCid(signedBytes);
    const commitCidStr = cidToString(commitCid);

    // Store commit block
    this.sql.exec(
      `INSERT OR REPLACE INTO blocks (cid, data) VALUES (?, ?)`,
      commitCidStr,
      signedBytes,
    );

    // Store commit reference
    this.sql.exec(
      `INSERT INTO commits (cid, rev, prev) VALUES (?, ?, ?)`,
      commitCidStr,
      rev,
      prevCommit?.cid || null,
    );

    // Update head and rev for listRepos
    await this.state.storage.put('head', commitCidStr);
    await this.state.storage.put('rev', rev);

    // Collect blocks for the event (record + commit + MST nodes)
    // Build a mini CAR with just the new blocks - use string CIDs
    const newBlocks = [];
    // Add record block
    newBlocks.push({ cid: recordCidStr, data: recordBytes });
    // Add commit block
    newBlocks.push({ cid: commitCidStr, data: signedBytes });
    // Add MST node blocks (get all blocks referenced by commit.data)
    const mstBlocks = this.collectMstBlocks(/** @type {string} */ (dataRoot));
    newBlocks.push(...mstBlocks);

    // Sequence event with blocks - store complete event data including rev and time
    // blocks must be a full CAR file with header (roots = [commitCid])
    const eventTime = new Date().toISOString();
    const evt = cborEncode({
      ops: [
        { action: 'create', path: `${collection}/${rkey}`, cid: recordCidStr },
      ],
      blocks: buildCarFile(commitCidStr, newBlocks), // Full CAR with header
      rev, // Store the actual commit revision
      time: eventTime, // Store the actual event time
    });
    this.sql.exec(
      `INSERT INTO seq_events (did, commit_cid, evt) VALUES (?, ?, ?)`,
      did,
      commitCidStr,
      evt,
    );

    // Broadcast to subscribers (both local and via default DO for relay)
    const evtRows = /** @type {SeqEventRow[]} */ (
      this.sql
        .exec(`SELECT * FROM seq_events ORDER BY seq DESC LIMIT 1`)
        .toArray()
    );
    if (evtRows.length > 0) {
      this.broadcastEvent(evtRows[0]);
      // Also forward to default DO for relay subscribers
      if (this.env?.PDS) {
        const defaultId = this.env.PDS.idFromName('default');
        const defaultPds = this.env.PDS.get(defaultId);
        // Convert ArrayBuffer to array for JSON serialization
        const row = evtRows[0];
        const evtArray = Array.from(new Uint8Array(row.evt));
        // Fire and forget but log errors
        defaultPds
          .fetch(
            new Request('http://internal/forward-event', {
              method: 'POST',
              body: JSON.stringify({ ...row, evt: evtArray }),
            }),
          )
          .catch(() => {}); // Ignore forward errors
      }
    }

    return { uri, cid: recordCidStr, commit: commitCidStr };
  }

  /**
   * @param {string} collection
   * @param {string} rkey
   */
  async deleteRecord(collection, rkey) {
    const did = await this.getDid();
    if (!did) throw new Error('PDS not initialized');

    const uri = `at://${did}/${collection}/${rkey}`;

    // Check if record exists
    const existing = this.sql
      .exec(`SELECT cid FROM records WHERE uri = ?`, uri)
      .toArray();
    if (existing.length === 0) {
      return { error: 'RecordNotFound', message: 'record not found' };
    }

    // Delete from records table
    this.sql.exec(`DELETE FROM records WHERE uri = ?`, uri);

    // Get blobs associated with this record
    const associatedBlobs = this.sql
      .exec('SELECT blobCid FROM record_blob WHERE recordUri = ?', uri)
      .toArray();

    // Remove associations for this record
    this.sql.exec('DELETE FROM record_blob WHERE recordUri = ?', uri);

    // Check each blob for orphan status and delete if unreferenced
    for (const { blobCid } of associatedBlobs) {
      const stillReferenced = this.sql
        .exec('SELECT 1 FROM record_blob WHERE blobCid = ? LIMIT 1', blobCid)
        .toArray();

      if (stillReferenced.length === 0) {
        // Blob is orphaned, delete from R2 and database
        await this.env?.BLOBS?.delete(`${did}/${blobCid}`);
        this.sql.exec('DELETE FROM blob WHERE cid = ?', blobCid);
      }
    }

    // Rebuild MST
    const mst = new MST(this.sql);
    const dataRoot = await mst.computeRoot();

    // Get previous commit
    const prevCommits = this.sql
      .exec(`SELECT cid, rev FROM commits ORDER BY seq DESC LIMIT 1`)
      .toArray();
    const prevCommit = prevCommits.length > 0 ? prevCommits[0] : null;

    // Create commit
    const rev = createTid();
    const commit = {
      did,
      version: 3,
      data: dataRoot
        ? new CID(cidToBytes(/** @type {string} */ (dataRoot)))
        : null,
      rev,
      prev: prevCommit?.cid
        ? new CID(cidToBytes(/** @type {string} */ (prevCommit.cid)))
        : null,
    };

    // Sign commit
    const commitBytes = cborEncodeDagCbor(commit);
    const signingKey = await this.getSigningKey();
    if (!signingKey) throw new Error('No signing key');
    const sig = await sign(signingKey, commitBytes);

    const signedCommit = { ...commit, sig };
    const signedBytes = cborEncodeDagCbor(signedCommit);
    const commitCid = await createCid(signedBytes);
    const commitCidStr = cidToString(commitCid);

    // Store commit block
    this.sql.exec(
      `INSERT OR REPLACE INTO blocks (cid, data) VALUES (?, ?)`,
      commitCidStr,
      signedBytes,
    );

    // Store commit reference
    this.sql.exec(
      `INSERT INTO commits (cid, rev, prev) VALUES (?, ?, ?)`,
      commitCidStr,
      rev,
      prevCommit?.cid || null,
    );

    // Update head and rev
    await this.state.storage.put('head', commitCidStr);
    await this.state.storage.put('rev', rev);

    // Collect blocks for the event (commit + MST nodes, no record block)
    const newBlocks = [];
    newBlocks.push({ cid: commitCidStr, data: signedBytes });
    if (dataRoot) {
      const mstBlocks = this.collectMstBlocks(/** @type {string} */ (dataRoot));
      newBlocks.push(...mstBlocks);
    }

    // Sequence event with delete action
    const eventTime = new Date().toISOString();
    const evt = cborEncode({
      ops: [{ action: 'delete', path: `${collection}/${rkey}`, cid: null }],
      blocks: buildCarFile(commitCidStr, newBlocks),
      rev,
      time: eventTime,
    });
    this.sql.exec(
      `INSERT INTO seq_events (did, commit_cid, evt) VALUES (?, ?, ?)`,
      did,
      commitCidStr,
      evt,
    );

    // Broadcast to subscribers
    const evtRows = /** @type {SeqEventRow[]} */ (
      this.sql
        .exec(`SELECT * FROM seq_events ORDER BY seq DESC LIMIT 1`)
        .toArray()
    );
    if (evtRows.length > 0) {
      this.broadcastEvent(evtRows[0]);
      // Forward to default DO for relay subscribers
      if (this.env?.PDS) {
        const defaultId = this.env.PDS.idFromName('default');
        const defaultPds = this.env.PDS.get(defaultId);
        const row = evtRows[0];
        const evtArray = Array.from(new Uint8Array(row.evt));
        defaultPds
          .fetch(
            new Request('http://internal/forward-event', {
              method: 'POST',
              body: JSON.stringify({ ...row, evt: evtArray }),
            }),
          )
          .catch(() => {}); // Ignore forward errors
      }
    }

    return { ok: true };
  }

  /**
   * @param {SeqEventRow} evt
   * @returns {Uint8Array}
   */
  formatEvent(evt) {
    // AT Protocol frame format: header + body
    // Use DAG-CBOR encoding for body (CIDs need tag 42 + 0x00 prefix)
    const header = cborEncode({ op: 1, t: '#commit' });

    // Decode stored event to get ops, blocks, rev, and time
    const evtData = cborDecode(new Uint8Array(evt.evt));
    /** @type {Array<{action: string, path: string, cid: CID|null}>} */
    const ops = evtData.ops.map(
      (/** @type {{action: string, path: string, cid?: string}} */ op) => ({
        ...op,
        cid: op.cid ? new CID(cidToBytes(op.cid)) : null, // Wrap in CID class for tag 42 encoding
      }),
    );
    // Get blocks from stored event (already in CAR format)
    const blocks = evtData.blocks || new Uint8Array(0);

    const body = cborEncodeDagCbor({
      seq: evt.seq,
      rebase: false,
      tooBig: false,
      repo: evt.did,
      commit: new CID(cidToBytes(evt.commit_cid)), // Wrap in CID class for tag 42 encoding
      rev: evtData.rev, // Use stored rev from commit creation
      since: null,
      blocks: blocks instanceof Uint8Array ? blocks : new Uint8Array(blocks),
      ops,
      blobs: [],
      time: evtData.time, // Use stored time from event creation
    });

    // Concatenate header + body
    const frame = new Uint8Array(header.length + body.length);
    frame.set(header);
    frame.set(body, header.length);
    return frame;
  }

  /**
   * @param {WebSocket} ws
   * @param {string | ArrayBuffer} message
   */
  async webSocketMessage(ws, message) {
    // Handle ping
    if (message === 'ping') ws.send('pong');
  }

  /**
   * @param {WebSocket} _ws
   * @param {number} _code
   * @param {string} _reason
   */
  async webSocketClose(_ws, _code, _reason) {
    // Durable Object will hibernate when no connections remain
  }

  /**
   * @param {SeqEventRow} evt
   */
  broadcastEvent(evt) {
    const frame = this.formatEvent(evt);
    for (const ws of this.state.getWebSockets()) {
      try {
        ws.send(frame);
      } catch (_e) {
        // Client disconnected
      }
    }
  }

  async handleAtprotoDid() {
    let did = await this.getDid();
    if (!did) {
      /** @type {string[]} */
      const registeredDids =
        (await this.state.storage.get('registeredDids')) || [];
      did = registeredDids[0];
    }
    if (!did) {
      return new Response('User not found', { status: 404 });
    }
    return new Response(/** @type {string} */ (did), {
      headers: { 'Content-Type': 'text/plain' },
    });
  }

  /** @param {Request} request */
  async handleInit(request) {
    const body = await request.json();
    if (!body.did || !body.privateKey) {
      return errorResponse('InvalidRequest', 'missing did or privateKey', 400);
    }
    await this.initIdentity(body.did, body.privateKey, body.handle || null);
    return Response.json({
      ok: true,
      did: body.did,
      handle: body.handle || null,
    });
  }

  async handleStatus() {
    const did = await this.getDid();
    return Response.json({ initialized: !!did, did: did || null });
  }

  /** @param {Request} request */
  async handleForwardEvent(request) {
    const evt = await request.json();
    const numSockets = [...this.state.getWebSockets()].length;
    this.broadcastEvent({
      seq: evt.seq,
      did: evt.did,
      commit_cid: evt.commit_cid,
      evt: new Uint8Array(Object.values(evt.evt)),
    });
    return Response.json({ ok: true, sockets: numSockets });
  }

  /** @param {Request} request */
  async handleRegisterDid(request) {
    const body = await request.json();
    /** @type {string[]} */
    const registeredDids =
      (await this.state.storage.get('registeredDids')) || [];
    if (!registeredDids.includes(body.did)) {
      registeredDids.push(body.did);
      await this.state.storage.put('registeredDids', registeredDids);
    }
    return Response.json({ ok: true });
  }

  async handleGetRegisteredDids() {
    const registeredDids =
      (await this.state.storage.get('registeredDids')) || [];
    return Response.json({ dids: registeredDids });
  }

  /** @param {Request} request */
  async handleRegisterHandle(request) {
    const body = await request.json();
    const { handle, did } = body;
    if (!handle || !did) {
      return errorResponse('InvalidRequest', 'missing handle or did', 400);
    }
    /** @type {Record<string, string>} */
    const handleMap = (await this.state.storage.get('handleMap')) || {};
    handleMap[handle] = did;
    await this.state.storage.put('handleMap', handleMap);
    return Response.json({ ok: true });
  }

  /** @param {URL} url */
  async handleResolveHandle(url) {
    const handle = url.searchParams.get('handle');
    if (!handle) {
      return errorResponse('InvalidRequest', 'missing handle', 400);
    }
    /** @type {Record<string, string>} */
    const handleMap = (await this.state.storage.get('handleMap')) || {};
    const did = handleMap[handle];
    if (!did) {
      return errorResponse('NotFound', 'handle not found', 404);
    }
    return Response.json({ did });
  }

  async handleRepoInfo() {
    const head = await this.state.storage.get('head');
    const rev = await this.state.storage.get('rev');
    return Response.json({ head: head || null, rev: rev || null });
  }

  /** @param {Request} request */
  handleDescribeServer(request) {
    const hostname = request.headers.get('x-hostname') || 'localhost';
    return Response.json({
      did: `did:web:${hostname}`,
      availableUserDomains: [`.${hostname}`],
      inviteCodeRequired: false,
      phoneVerificationRequired: false,
      links: {},
      contact: {},
    });
  }

  /** @param {Request} request */
  async handleCreateSession(request) {
    const body = await request.json();
    const { identifier, password } = body;

    if (!identifier || !password) {
      return errorResponse(
        'InvalidRequest',
        'Missing identifier or password',
        400,
      );
    }

    // Check password against env var
    const expectedPassword = this.env?.PDS_PASSWORD;
    if (!expectedPassword || password !== expectedPassword) {
      return errorResponse(
        'AuthRequired',
        'Invalid identifier or password',
        401,
      );
    }

    // Resolve identifier to DID
    let did = identifier;
    if (!identifier.startsWith('did:')) {
      // Try to resolve handle
      /** @type {Record<string, string>} */
      const handleMap = (await this.state.storage.get('handleMap')) || {};
      did = handleMap[identifier];
      if (!did) {
        return errorResponse('InvalidRequest', 'Unable to resolve handle', 400);
      }
    }

    // Get handle for response
    const handle = await this.getHandleForDid(did);

    // Create tokens
    const jwtSecret = this.env?.JWT_SECRET;
    if (!jwtSecret) {
      return errorResponse(
        'InternalServerError',
        'Server not configured for authentication',
        500,
      );
    }

    const accessJwt = await createAccessJwt(did, jwtSecret);
    const refreshJwt = await createRefreshJwt(did, jwtSecret);

    return Response.json({
      accessJwt,
      refreshJwt,
      handle: handle || did,
      did,
      active: true,
    });
  }

  /** @param {Request} request */
  async handleGetSession(request) {
    const authHeader = request.headers.get('Authorization');
    if (!authHeader || !authHeader.startsWith('Bearer ')) {
      return errorResponse(
        'AuthRequired',
        'Missing or invalid authorization header',
        401,
      );
    }

    const token = authHeader.slice(7); // Remove 'Bearer '
    const jwtSecret = this.env?.JWT_SECRET;
    if (!jwtSecret) {
      return errorResponse(
        'InternalServerError',
        'Server not configured for authentication',
        500,
      );
    }

    try {
      const payload = await verifyAccessJwt(token, jwtSecret);
      const did = payload.sub;
      const handle = await this.getHandleForDid(did);

      return Response.json({
        handle: handle || did,
        did,
        active: true,
      });
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return errorResponse('InvalidToken', message, 401);
    }
  }

  /** @param {Request} request */
  async handleRefreshSession(request) {
    const authHeader = request.headers.get('Authorization');
    if (!authHeader || !authHeader.startsWith('Bearer ')) {
      return errorResponse(
        'AuthRequired',
        'Missing or invalid authorization header',
        401,
      );
    }

    const token = authHeader.slice(7); // Remove 'Bearer '
    const jwtSecret = this.env?.JWT_SECRET;
    if (!jwtSecret) {
      return errorResponse(
        'InternalServerError',
        'Server not configured for authentication',
        500,
      );
    }

    try {
      const payload = await verifyRefreshJwt(token, jwtSecret);
      const did = payload.sub;
      const handle = await this.getHandleForDid(did);

      // Issue fresh tokens
      const accessJwt = await createAccessJwt(did, jwtSecret);
      const refreshJwt = await createRefreshJwt(did, jwtSecret);

      return Response.json({
        accessJwt,
        refreshJwt,
        handle: handle || did,
        did,
        active: true,
      });
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      if (message === 'Token expired') {
        return errorResponse('ExpiredToken', 'Refresh token has expired', 400);
      }
      return errorResponse('InvalidToken', message, 400);
    }
  }

  /** @param {Request} _request */
  async handleGetPreferences(_request) {
    // Preferences are stored per-user in their DO
    const preferences = (await this.state.storage.get('preferences')) || [];
    return Response.json({ preferences });
  }

  /** @param {Request} request */
  async handlePutPreferences(request) {
    const body = await request.json();
    const { preferences } = body;
    if (!Array.isArray(preferences)) {
      return errorResponse(
        'InvalidRequest',
        'preferences must be an array',
        400,
      );
    }
    await this.state.storage.put('preferences', preferences);
    return Response.json({});
  }

  /**
   * @param {string} did
   * @returns {Promise<string|null>}
   */
  async getHandleForDid(did) {
    // Check if this DID has a handle registered
    /** @type {Record<string, string>} */
    const handleMap = (await this.state.storage.get('handleMap')) || {};
    for (const [handle, mappedDid] of Object.entries(handleMap)) {
      if (mappedDid === did) return handle;
    }
    // Check instance's own handle
    const instanceDid = await this.getDid();
    if (instanceDid === did) {
      return /** @type {string|null} */ (
        await this.state.storage.get('handle')
      );
    }
    return null;
  }

  /**
   * @param {string} did
   * @param {string|null} lxm
   */
  async createServiceAuthForAppView(did, lxm) {
    const signingKey = await this.getSigningKey();
    if (!signingKey) throw new Error('No signing key available');
    return createServiceJwt({
      iss: did,
      aud: 'did:web:api.bsky.app',
      lxm,
      signingKey,
    });
  }

  /**
   * @param {Request} request
   * @param {string} userDid
   */
  async handleAppViewProxy(request, userDid) {
    const url = new URL(request.url);
    // Extract lexicon method from path: /xrpc/app.bsky.actor.getPreferences -> app.bsky.actor.getPreferences
    const lxm = url.pathname.replace('/xrpc/', '');

    // Create service auth JWT
    const serviceJwt = await this.createServiceAuthForAppView(userDid, lxm);

    // Build AppView URL
    const appViewUrl = new URL(
      url.pathname + url.search,
      'https://api.bsky.app',
    );

    // Forward request with service auth
    const headers = new Headers();
    headers.set('Authorization', `Bearer ${serviceJwt}`);
    headers.set(
      'Content-Type',
      request.headers.get('Content-Type') || 'application/json',
    );
    const acceptHeader = request.headers.get('Accept');
    if (acceptHeader) {
      headers.set('Accept', acceptHeader);
    }
    const acceptLangHeader = request.headers.get('Accept-Language');
    if (acceptLangHeader) {
      headers.set('Accept-Language', acceptLangHeader);
    }

    const proxyReq = new Request(appViewUrl.toString(), {
      method: request.method,
      headers,
      body:
        request.method !== 'GET' && request.method !== 'HEAD'
          ? request.body
          : undefined,
    });

    try {
      const response = await fetch(proxyReq);
      // Return the response with CORS headers
      const responseHeaders = new Headers(response.headers);
      responseHeaders.set('Access-Control-Allow-Origin', '*');
      return new Response(response.body, {
        status: response.status,
        statusText: response.statusText,
        headers: responseHeaders,
      });
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return errorResponse(
        'UpstreamFailure',
        `Failed to reach AppView: ${message}`,
        502,
      );
    }
  }

  async handleListRepos() {
    /** @type {string[]} */
    const registeredDids =
      (await this.state.storage.get('registeredDids')) || [];
    const did = await this.getDid();
    const repos = did
      ? [{ did, head: null, rev: null }]
      : registeredDids.map((/** @type {string} */ d) => ({
          did: d,
          head: null,
          rev: null,
        }));
    return Response.json({ repos });
  }

  /** @param {Request} request */
  async handleCreateRecord(request) {
    const body = await request.json();
    if (!body.collection || !body.record) {
      return errorResponse(
        'InvalidRequest',
        'missing collection or record',
        400,
      );
    }
    try {
      const result = await this.createRecord(
        body.collection,
        body.record,
        body.rkey,
      );
      const head = await this.state.storage.get('head');
      const rev = await this.state.storage.get('rev');
      return Response.json({
        uri: result.uri,
        cid: result.cid,
        commit: { cid: head, rev },
        validationStatus: 'valid',
      });
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return errorResponse('InternalError', message, 500);
    }
  }

  /** @param {Request} request */
  async handleDeleteRecord(request) {
    const body = await request.json();
    if (!body.collection || !body.rkey) {
      return errorResponse('InvalidRequest', 'missing collection or rkey', 400);
    }
    try {
      const result = await this.deleteRecord(body.collection, body.rkey);
      if (result.error) {
        return Response.json(result, { status: 404 });
      }
      return Response.json({});
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return errorResponse('InternalError', message, 500);
    }
  }

  /** @param {Request} request */
  async handlePutRecord(request) {
    const body = await request.json();
    if (!body.collection || !body.rkey || !body.record) {
      return errorResponse(
        'InvalidRequest',
        'missing collection, rkey, or record',
        400,
      );
    }
    try {
      // putRecord is like createRecord but with a specific rkey (upsert)
      const result = await this.createRecord(
        body.collection,
        body.record,
        body.rkey,
      );
      const head = await this.state.storage.get('head');
      const rev = await this.state.storage.get('rev');
      return Response.json({
        uri: result.uri,
        cid: result.cid,
        commit: { cid: head, rev },
        validationStatus: 'valid',
      });
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return errorResponse('InternalError', message, 500);
    }
  }

  /** @param {Request} request */
  async handleApplyWrites(request) {
    const body = await request.json();
    if (!body.writes || !Array.isArray(body.writes)) {
      return errorResponse('InvalidRequest', 'missing writes array', 400);
    }
    try {
      const results = [];
      for (const write of body.writes) {
        const type = write.$type;
        if (type === 'com.atproto.repo.applyWrites#create') {
          const result = await this.createRecord(
            write.collection,
            write.value,
            write.rkey,
          );
          results.push({
            $type: 'com.atproto.repo.applyWrites#createResult',
            uri: result.uri,
            cid: result.cid,
            validationStatus: 'valid',
          });
        } else if (type === 'com.atproto.repo.applyWrites#update') {
          const result = await this.createRecord(
            write.collection,
            write.value,
            write.rkey,
          );
          results.push({
            $type: 'com.atproto.repo.applyWrites#updateResult',
            uri: result.uri,
            cid: result.cid,
            validationStatus: 'valid',
          });
        } else if (type === 'com.atproto.repo.applyWrites#delete') {
          await this.deleteRecord(write.collection, write.rkey);
          results.push({
            $type: 'com.atproto.repo.applyWrites#deleteResult',
          });
        } else {
          return errorResponse(
            'InvalidRequest',
            `Unknown write operation type: ${type}`,
            400,
          );
        }
      }
      // Return commit info
      const head = await this.state.storage.get('head');
      const rev = await this.state.storage.get('rev');
      return Response.json({ commit: { cid: head, rev }, results });
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return errorResponse('InternalError', message, 500);
    }
  }

  /** @param {URL} url */
  async handleGetRecord(url) {
    const collection = url.searchParams.get('collection');
    const rkey = url.searchParams.get('rkey');
    if (!collection || !rkey) {
      return errorResponse('InvalidRequest', 'missing collection or rkey', 400);
    }
    const did = await this.getDid();
    const uri = `at://${did}/${collection}/${rkey}`;
    const rows = /** @type {RecordRow[]} */ (
      this.sql
        .exec(`SELECT cid, value FROM records WHERE uri = ?`, uri)
        .toArray()
    );
    if (rows.length === 0) {
      return errorResponse('RecordNotFound', 'record not found', 404);
    }
    const row = rows[0];
    const value = cborDecode(new Uint8Array(row.value));
    return Response.json({ uri, cid: row.cid, value });
  }

  async handleDescribeRepo() {
    const did = await this.getDid();
    if (!did) {
      return errorResponse('RepoNotFound', 'repo not found', 404);
    }
    const handle = await this.state.storage.get('handle');
    // Get unique collections
    const collections = this.sql
      .exec(`SELECT DISTINCT collection FROM records`)
      .toArray()
      .map((r) => r.collection);

    return Response.json({
      handle: handle || did,
      did,
      didDoc: {},
      collections,
      handleIsCorrect: !!handle,
    });
  }

  /** @param {URL} url */
  async handleListRecords(url) {
    const collection = url.searchParams.get('collection');
    if (!collection) {
      return errorResponse('InvalidRequest', 'missing collection', 400);
    }
    const limit = Math.min(
      parseInt(url.searchParams.get('limit') || '50', 10),
      100,
    );
    const reverse = url.searchParams.get('reverse') === 'true';
    const _cursor = url.searchParams.get('cursor');

    const _did = await this.getDid();
    const query = `SELECT uri, cid, value FROM records WHERE collection = ? ORDER BY rkey ${reverse ? 'DESC' : 'ASC'} LIMIT ?`;
    const params = [collection, limit + 1];

    const rows = /** @type {RecordRow[]} */ (
      this.sql.exec(query, ...params).toArray()
    );
    const hasMore = rows.length > limit;
    const records = rows.slice(0, limit).map((r) => ({
      uri: r.uri,
      cid: r.cid,
      value: cborDecode(new Uint8Array(r.value)),
    }));

    return Response.json({
      records,
      cursor: hasMore ? records[records.length - 1]?.uri : undefined,
    });
  }

  handleGetLatestCommit() {
    const commits = /** @type {CommitRow[]} */ (
      this.sql
        .exec(`SELECT cid, rev FROM commits ORDER BY seq DESC LIMIT 1`)
        .toArray()
    );
    if (commits.length === 0) {
      return errorResponse('RepoNotFound', 'repo not found', 404);
    }
    return Response.json({ cid: commits[0].cid, rev: commits[0].rev });
  }

  async handleGetRepoStatus() {
    const did = await this.getDid();
    const commits = /** @type {CommitRow[]} */ (
      this.sql
        .exec(`SELECT cid, rev FROM commits ORDER BY seq DESC LIMIT 1`)
        .toArray()
    );
    if (commits.length === 0 || !did) {
      return errorResponse('RepoNotFound', 'repo not found', 404);
    }
    return Response.json({
      did,
      active: true,
      status: 'active',
      rev: commits[0].rev,
    });
  }

  handleGetRepo() {
    const commits = /** @type {CommitRow[]} */ (
      this.sql
        .exec(`SELECT cid FROM commits ORDER BY seq DESC LIMIT 1`)
        .toArray()
    );
    if (commits.length === 0) {
      return errorResponse('RepoNotFound', 'repo not found', 404);
    }

    // Only include blocks reachable from the current commit
    const commitCid = commits[0].cid;
    const neededCids = new Set();

    // Helper to get block data
    /** @param {string} cid */
    const getBlock = (cid) => {
      const rows = /** @type {BlockRow[]} */ (
        this.sql.exec(`SELECT data FROM blocks WHERE cid = ?`, cid).toArray()
      );
      return rows.length > 0 ? new Uint8Array(rows[0].data) : null;
    };

    // Collect all reachable blocks starting from commit
    /** @param {string} cid */
    const collectBlocks = (cid) => {
      if (neededCids.has(cid)) return;
      neededCids.add(cid);

      const data = getBlock(cid);
      if (!data) return;

      // Decode CBOR to find CID references
      try {
        const decoded = cborDecode(data);
        if (decoded && typeof decoded === 'object') {
          // Commit object - follow 'data' (MST root)
          if (decoded.data instanceof Uint8Array) {
            collectBlocks(cidToString(decoded.data));
          }
          // MST node - follow 'l' and entries' 'v' and 't'
          if (decoded.l instanceof Uint8Array) {
            collectBlocks(cidToString(decoded.l));
          }
          if (Array.isArray(decoded.e)) {
            for (const entry of decoded.e) {
              if (entry.v instanceof Uint8Array) {
                collectBlocks(cidToString(entry.v));
              }
              if (entry.t instanceof Uint8Array) {
                collectBlocks(cidToString(entry.t));
              }
            }
          }
        }
      } catch (_e) {
        // Not a structured block, that's fine
      }
    };

    collectBlocks(commitCid);

    // Build CAR with only needed blocks
    const blocksForCar = [];
    for (const cid of neededCids) {
      const data = getBlock(cid);
      if (data) {
        blocksForCar.push({ cid, data });
      }
    }

    const car = buildCarFile(commitCid, blocksForCar);
    return new Response(/** @type {BodyInit} */ (car), {
      headers: { 'content-type': 'application/vnd.ipld.car' },
    });
  }

  /** @param {URL} url */
  async handleSyncGetRecord(url) {
    const collection = url.searchParams.get('collection');
    const rkey = url.searchParams.get('rkey');
    if (!collection || !rkey) {
      return errorResponse('InvalidRequest', 'missing collection or rkey', 400);
    }
    const did = await this.getDid();
    const uri = `at://${did}/${collection}/${rkey}`;
    const rows = /** @type {RecordRow[]} */ (
      this.sql.exec(`SELECT cid FROM records WHERE uri = ?`, uri).toArray()
    );
    if (rows.length === 0) {
      return errorResponse('RecordNotFound', 'record not found', 404);
    }
    const recordCid = rows[0].cid;

    // Get latest commit
    const commits = /** @type {CommitRow[]} */ (
      this.sql
        .exec(`SELECT cid FROM commits ORDER BY seq DESC LIMIT 1`)
        .toArray()
    );
    if (commits.length === 0) {
      return errorResponse('RepoNotFound', 'no commits', 404);
    }
    const commitCid = commits[0].cid;

    // Build proof chain: commit -> MST path -> record
    // Include commit block, all MST nodes on path to record, and record block
    /** @type {Array<{cid: string, data: Uint8Array}>} */
    const blocks = [];
    const included = new Set();

    /** @param {string} cidStr */
    const addBlock = (cidStr) => {
      if (included.has(cidStr)) return;
      included.add(cidStr);
      const blockRows = /** @type {BlockRow[]} */ (
        this.sql.exec(`SELECT data FROM blocks WHERE cid = ?`, cidStr).toArray()
      );
      if (blockRows.length > 0) {
        blocks.push({ cid: cidStr, data: new Uint8Array(blockRows[0].data) });
      }
    };

    // Add commit block
    addBlock(commitCid);

    // Get commit to find data root
    const commitRows = /** @type {BlockRow[]} */ (
      this.sql
        .exec(`SELECT data FROM blocks WHERE cid = ?`, commitCid)
        .toArray()
    );
    if (commitRows.length > 0) {
      const commit = cborDecode(new Uint8Array(commitRows[0].data));
      if (commit.data) {
        const dataRootCid = cidToString(commit.data);
        // Collect MST path blocks (this includes all MST nodes)
        const mstBlocks = this.collectMstBlocks(dataRootCid);
        for (const block of mstBlocks) {
          addBlock(block.cid);
        }
      }
    }

    // Add record block
    addBlock(recordCid);

    const car = buildCarFile(commitCid, blocks);
    return new Response(/** @type {BodyInit} */ (car), {
      headers: { 'content-type': 'application/vnd.ipld.car' },
    });
  }

  /** @param {Request} request */
  async handleUploadBlob(request) {
    // Require auth
    const authHeader = request.headers.get('Authorization');
    if (!authHeader || !authHeader.startsWith('Bearer ')) {
      return errorResponse(
        'AuthRequired',
        'Missing or invalid authorization header',
        401,
      );
    }

    const token = authHeader.slice(7);
    const jwtSecret = this.env?.JWT_SECRET;
    if (!jwtSecret) {
      return errorResponse(
        'InternalServerError',
        'Server not configured for authentication',
        500,
      );
    }

    try {
      await verifyAccessJwt(token, jwtSecret);
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return errorResponse('InvalidToken', message, 401);
    }

    const did = await this.getDid();
    if (!did) {
      return errorResponse('InvalidRequest', 'PDS not initialized', 400);
    }

    // Read body as ArrayBuffer
    const bodyBytes = await request.arrayBuffer();
    const size = bodyBytes.byteLength;

    // Check size limits
    if (size === 0) {
      return errorResponse(
        'InvalidRequest',
        'Empty blobs are not allowed',
        400,
      );
    }
    const MAX_BLOB_SIZE = 50 * 1024 * 1024;
    if (size > MAX_BLOB_SIZE) {
      return errorResponse(
        'BlobTooLarge',
        `Blob size ${size} exceeds maximum ${MAX_BLOB_SIZE}`,
        400,
      );
    }

    // Sniff MIME type, fall back to Content-Type header
    const contentType =
      request.headers.get('Content-Type') || 'application/octet-stream';
    const sniffed = sniffMimeType(bodyBytes);
    const mimeType = sniffed || contentType;

    // Compute CID using raw codec for blobs
    const cid = await createBlobCid(new Uint8Array(bodyBytes));
    const cidStr = cidToString(cid);

    // Upload to R2 (idempotent - same CID always has same content)
    const r2Key = `${did}/${cidStr}`;
    await this.env?.BLOBS?.put(r2Key, bodyBytes, {
      httpMetadata: { contentType: mimeType },
    });

    // Insert metadata (INSERT OR IGNORE handles concurrent uploads)
    const createdAt = new Date().toISOString();
    this.sql.exec(
      'INSERT OR IGNORE INTO blob (cid, mimeType, size, createdAt) VALUES (?, ?, ?, ?)',
      cidStr,
      mimeType,
      size,
      createdAt,
    );

    // Return BlobRef
    return Response.json({
      blob: {
        $type: 'blob',
        ref: { $link: cidStr },
        mimeType,
        size,
      },
    });
  }

  /** @param {URL} url */
  async handleGetBlob(url) {
    const did = url.searchParams.get('did');
    const cid = url.searchParams.get('cid');

    if (!did || !cid) {
      return errorResponse(
        'InvalidRequest',
        'missing did or cid parameter',
        400,
      );
    }

    // Validate CID format (CIDv1 base32lower: starts with 'b', 59 chars total)
    if (!/^b[a-z2-7]{58}$/.test(cid)) {
      return errorResponse('InvalidRequest', 'invalid CID format', 400);
    }

    // Verify DID matches this DO
    const myDid = await this.getDid();
    if (did !== myDid) {
      return errorResponse(
        'InvalidRequest',
        'DID does not match this repo',
        400,
      );
    }

    // Look up blob metadata
    const rows = this.sql
      .exec('SELECT mimeType, size FROM blob WHERE cid = ?', cid)
      .toArray();

    if (rows.length === 0) {
      return errorResponse('BlobNotFound', 'blob not found', 404);
    }

    const { mimeType, size } = rows[0];

    // Fetch from R2
    const r2Key = `${did}/${cid}`;
    const object = await this.env?.BLOBS?.get(r2Key);

    if (!object) {
      return errorResponse('BlobNotFound', 'blob not found in storage', 404);
    }

    // Return blob with security headers
    return new Response(object.body, {
      headers: {
        'Content-Type': /** @type {string} */ (mimeType),
        'Content-Length': String(size),
        'X-Content-Type-Options': 'nosniff',
        'Content-Security-Policy': "default-src 'none'; sandbox",
        'Cache-Control': 'public, max-age=31536000, immutable',
      },
    });
  }

  /** @param {URL} url */
  async handleListBlobs(url) {
    const did = url.searchParams.get('did');
    const cursor = url.searchParams.get('cursor');
    const limit = Math.min(Number(url.searchParams.get('limit')) || 500, 1000);

    if (!did) {
      return errorResponse('InvalidRequest', 'missing did parameter', 400);
    }

    // Verify DID matches this DO
    const myDid = await this.getDid();
    if (did !== myDid) {
      return errorResponse(
        'InvalidRequest',
        'DID does not match this repo',
        400,
      );
    }

    // Query blobs with pagination (cursor is createdAt::cid for uniqueness)
    let query = 'SELECT cid, createdAt FROM blob';
    const params = [];

    if (cursor) {
      const [cursorTime, cursorCid] = cursor.split('::');
      query += ' WHERE (createdAt > ? OR (createdAt = ? AND cid > ?))';
      params.push(cursorTime, cursorTime, cursorCid);
    }

    query += ' ORDER BY createdAt ASC, cid ASC LIMIT ?';
    params.push(limit + 1); // Fetch one extra to detect if there's more

    const rows = this.sql.exec(query, ...params).toArray();

    // Determine if there's a next page
    let nextCursor = null;
    if (rows.length > limit) {
      rows.pop(); // Remove the extra row
      const last = rows[rows.length - 1];
      nextCursor = `${last.createdAt}::${last.cid}`;
    }

    return Response.json({
      cids: rows.map((r) => r.cid),
      cursor: nextCursor,
    });
  }

  /**
   * @param {Request} request
   * @param {URL} url
   */
  handleSubscribeRepos(request, url) {
    const upgradeHeader = request.headers.get('Upgrade');
    if (upgradeHeader !== 'websocket') {
      return new Response('expected websocket', { status: 426 });
    }
    const { 0: client, 1: server } = new WebSocketPair();
    this.state.acceptWebSocket(server);
    const cursor = url.searchParams.get('cursor');
    if (cursor) {
      const events = /** @type {SeqEventRow[]} */ (
        this.sql
          .exec(
            `SELECT * FROM seq_events WHERE seq > ? ORDER BY seq`,
            parseInt(cursor, 10),
          )
          .toArray()
      );
      for (const evt of events) {
        server.send(this.formatEvent(evt));
      }
    }
    return new Response(null, { status: 101, webSocket: client });
  }

  /** @param {Request} request */
  async fetch(request) {
    const url = new URL(request.url);
    const route = pdsRoutes[url.pathname];

    // Check for local route first
    if (route) {
      if (route.method && request.method !== route.method) {
        return errorResponse('MethodNotAllowed', 'method not allowed', 405);
      }
      return route.handler(this, request, url);
    }

    // Handle app.bsky.* proxy requests (only if no local route)
    if (url.pathname.startsWith('/xrpc/app.bsky.')) {
      const userDid = request.headers.get('x-authed-did');
      if (!userDid) {
        return errorResponse('Unauthorized', 'Missing auth context', 401);
      }
      return this.handleAppViewProxy(request, userDid);
    }

    return errorResponse('NotFound', 'not found', 404);
  }

  async alarm() {
    await this.cleanupOrphanedBlobs();
    // Reschedule for next day
    await this.state.storage.setAlarm(Date.now() + 24 * 60 * 60 * 1000);
  }

  async cleanupOrphanedBlobs() {
    const did = await this.getDid();
    if (!did) return;

    // Find orphans: blobs not in record_blob, older than 24h
    const cutoff = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString();

    const orphans = this.sql
      .exec(
        `SELECT b.cid FROM blob b
         LEFT JOIN record_blob rb ON b.cid = rb.blobCid
         WHERE rb.blobCid IS NULL AND b.createdAt < ?`,
        cutoff,
      )
      .toArray();

    for (const { cid } of orphans) {
      await this.env?.BLOBS?.delete(`${did}/${cid}`);
      this.sql.exec('DELETE FROM blob WHERE cid = ?', cid);
    }
  }
}

// ╔══════════════════════════════════════════════════════════════════════════════╗
// ║ WORKERS ENTRY POINT                                                          ║
// ║ Request handling, CORS, auth middleware                                      ║
// ╚══════════════════════════════════════════════════════════════════════════════╝

const corsHeaders = {
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': 'GET, POST, OPTIONS',
  'Access-Control-Allow-Headers':
    'Content-Type, Authorization, atproto-accept-labelers, atproto-proxy, x-bsky-topics',
};

/**
 * @param {Response} response
 * @returns {Response}
 */
function addCorsHeaders(response) {
  const newHeaders = new Headers(response.headers);
  for (const [key, value] of Object.entries(corsHeaders)) {
    newHeaders.set(key, value);
  }
  return new Response(response.body, {
    status: response.status,
    statusText: response.statusText,
    headers: newHeaders,
  });
}

export default {
  /**
   * @param {Request} request
   * @param {Env} env
   */
  async fetch(request, env) {
    // Handle CORS preflight
    if (request.method === 'OPTIONS') {
      return new Response(null, { headers: corsHeaders });
    }

    const response = await handleRequest(request, env);
    // Don't wrap WebSocket upgrades - they need the webSocket property preserved
    if (response.status === 101) {
      return response;
    }
    return addCorsHeaders(response);
  },
};

/**
 * Extract subdomain from hostname (e.g., "alice" from "alice.foo.workers.dev")
 * @param {string} hostname
 * @returns {string|null}
 */
function getSubdomain(hostname) {
  const parts = hostname.split('.');
  // workers.dev domains: [subdomain?].[worker-name].[account].workers.dev
  // If more than 4 parts, first part(s) are user subdomain
  if (parts.length > 4 && parts.slice(-2).join('.') === 'workers.dev') {
    return parts.slice(0, -4).join('.');
  }
  // Custom domains: check if there's a subdomain before the base
  // For now, assume no subdomain on custom domains
  return null;
}

/**
 * Verify auth and return DID from token
 * @param {Request} request - HTTP request with Authorization header
 * @param {Env} env - Environment with JWT_SECRET
 * @returns {Promise<{did: string} | {error: Response}>} DID or error response
 */
async function requireAuth(request, env) {
  const authHeader = request.headers.get('Authorization');
  if (!authHeader || !authHeader.startsWith('Bearer ')) {
    return {
      error: Response.json(
        {
          error: 'AuthRequired',
          message: 'Authentication required',
        },
        { status: 401 },
      ),
    };
  }

  const token = authHeader.slice(7);
  const jwtSecret = env?.JWT_SECRET;
  if (!jwtSecret) {
    return {
      error: Response.json(
        {
          error: 'InternalServerError',
          message: 'Server not configured for authentication',
        },
        { status: 500 },
      ),
    };
  }

  try {
    const payload = await verifyAccessJwt(token, jwtSecret);
    return { did: payload.sub };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    return {
      error: Response.json(
        {
          error: 'InvalidToken',
          message: message,
        },
        { status: 401 },
      ),
    };
  }
}

/**
 * @param {Request} request
 * @param {Env} env
 */
async function handleAuthenticatedBlobUpload(request, env) {
  const auth = await requireAuth(request, env);
  if ('error' in auth) return auth.error;

  // Route to the user's DO based on their DID from the token
  const id = env.PDS.idFromName(auth.did);
  const pds = env.PDS.get(id);
  return pds.fetch(request);
}

/**
 * @param {Request} request
 * @param {Env} env
 */
async function handleAuthenticatedRepoWrite(request, env) {
  const auth = await requireAuth(request, env);
  if ('error' in auth) return auth.error;

  const body = await request.json();
  const repo = body.repo;
  if (!repo) {
    return errorResponse('InvalidRequest', 'missing repo param', 400);
  }

  if (auth.did !== repo) {
    return errorResponse('Forbidden', "Cannot modify another user's repo", 403);
  }

  const id = env.PDS.idFromName(repo);
  const pds = env.PDS.get(id);
  const response = await pds.fetch(
    new Request(request.url, {
      method: 'POST',
      headers: request.headers,
      body: JSON.stringify(body),
    }),
  );

  // Notify relay of updates on successful writes
  if (response.ok) {
    const url = new URL(request.url);
    notifyCrawlers(env, url.hostname);
  }

  return response;
}

/**
 * @param {Request} request
 * @param {Env} env
 */
async function handleRequest(request, env) {
  const url = new URL(request.url);
  const subdomain = getSubdomain(url.hostname);

  // Handle resolution via subdomain or bare domain
  if (url.pathname === '/.well-known/atproto-did') {
    // Look up handle -> DID in default DO
    // Use subdomain if present, otherwise try bare hostname as handle
    const handleToResolve = subdomain || url.hostname;
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    const resolveRes = await defaultPds.fetch(
      new Request(
        `http://internal/resolve-handle?handle=${encodeURIComponent(handleToResolve)}`,
      ),
    );
    if (!resolveRes.ok) {
      return new Response('Handle not found', { status: 404 });
    }
    const { did } = await resolveRes.json();
    return new Response(did, { headers: { 'Content-Type': 'text/plain' } });
  }

  // describeServer - works on bare domain
  if (url.pathname === '/xrpc/com.atproto.server.describeServer') {
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    const newReq = new Request(request.url, {
      method: request.method,
      headers: {
        ...Object.fromEntries(request.headers),
        'x-hostname': url.hostname,
      },
    });
    return defaultPds.fetch(newReq);
  }

  // createSession - handle on default DO (has handleMap for identifier resolution)
  if (url.pathname === '/xrpc/com.atproto.server.createSession') {
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    return defaultPds.fetch(request);
  }

  // getSession - route to default DO
  if (url.pathname === '/xrpc/com.atproto.server.getSession') {
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    return defaultPds.fetch(request);
  }

  // refreshSession - route to default DO
  if (url.pathname === '/xrpc/com.atproto.server.refreshSession') {
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    return defaultPds.fetch(request);
  }

  // Proxy app.bsky.* endpoints to Bluesky AppView
  if (url.pathname.startsWith('/xrpc/app.bsky.')) {
    // Authenticate the user first
    const auth = await requireAuth(request, env);
    if ('error' in auth) return auth.error;

    // Route to the user's DO instance to create service auth and proxy
    const id = env.PDS.idFromName(auth.did);
    const pds = env.PDS.get(id);
    return pds.fetch(
      new Request(request.url, {
        method: request.method,
        headers: {
          ...Object.fromEntries(request.headers),
          'x-authed-did': auth.did, // Pass the authenticated DID
        },
        body:
          request.method !== 'GET' && request.method !== 'HEAD'
            ? request.body
            : undefined,
      }),
    );
  }

  // Handle registration routes - go to default DO
  if (
    url.pathname === '/register-handle' ||
    url.pathname === '/resolve-handle'
  ) {
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    return defaultPds.fetch(request);
  }

  // resolveHandle XRPC endpoint
  if (url.pathname === '/xrpc/com.atproto.identity.resolveHandle') {
    const handle = url.searchParams.get('handle');
    if (!handle) {
      return errorResponse('InvalidRequest', 'missing handle param', 400);
    }
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    const resolveRes = await defaultPds.fetch(
      new Request(
        `http://internal/resolve-handle?handle=${encodeURIComponent(handle)}`,
      ),
    );
    if (!resolveRes.ok) {
      return errorResponse('InvalidRequest', 'Unable to resolve handle', 400);
    }
    const { did } = await resolveRes.json();
    return Response.json({ did });
  }

  // subscribeRepos WebSocket - route to default instance for firehose
  if (url.pathname === '/xrpc/com.atproto.sync.subscribeRepos') {
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    return defaultPds.fetch(request);
  }

  // listRepos needs to aggregate from all registered DIDs
  if (url.pathname === '/xrpc/com.atproto.sync.listRepos') {
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    const regRes = await defaultPds.fetch(
      new Request('http://internal/get-registered-dids'),
    );
    const { dids } = await regRes.json();

    const repos = [];
    for (const did of dids) {
      const id = env.PDS.idFromName(did);
      const pds = env.PDS.get(id);
      const infoRes = await pds.fetch(new Request('http://internal/repo-info'));
      const info = await infoRes.json();
      if (info.head) {
        repos.push({ did, head: info.head, rev: info.rev, active: true });
      }
    }
    return Response.json({ repos, cursor: undefined });
  }

  // Repo endpoints use ?repo= param instead of ?did=
  if (
    url.pathname === '/xrpc/com.atproto.repo.describeRepo' ||
    url.pathname === '/xrpc/com.atproto.repo.listRecords' ||
    url.pathname === '/xrpc/com.atproto.repo.getRecord'
  ) {
    const repo = url.searchParams.get('repo');
    if (!repo) {
      return errorResponse('InvalidRequest', 'missing repo param', 400);
    }
    const id = env.PDS.idFromName(repo);
    const pds = env.PDS.get(id);
    return pds.fetch(request);
  }

  // Sync endpoints use ?did= param
  if (
    url.pathname === '/xrpc/com.atproto.sync.getLatestCommit' ||
    url.pathname === '/xrpc/com.atproto.sync.getRepoStatus' ||
    url.pathname === '/xrpc/com.atproto.sync.getRepo' ||
    url.pathname === '/xrpc/com.atproto.sync.getRecord' ||
    url.pathname === '/xrpc/com.atproto.sync.getBlob' ||
    url.pathname === '/xrpc/com.atproto.sync.listBlobs'
  ) {
    const did = url.searchParams.get('did');
    if (!did) {
      return errorResponse('InvalidRequest', 'missing did param', 400);
    }
    const id = env.PDS.idFromName(did);
    const pds = env.PDS.get(id);
    return pds.fetch(request);
  }

  // Blob upload endpoint (binary body, uses DID from token)
  if (url.pathname === '/xrpc/com.atproto.repo.uploadBlob') {
    return handleAuthenticatedBlobUpload(request, env);
  }

  // Authenticated repo write endpoints
  const repoWriteEndpoints = [
    '/xrpc/com.atproto.repo.createRecord',
    '/xrpc/com.atproto.repo.deleteRecord',
    '/xrpc/com.atproto.repo.putRecord',
    '/xrpc/com.atproto.repo.applyWrites',
  ];
  if (repoWriteEndpoints.includes(url.pathname)) {
    return handleAuthenticatedRepoWrite(request, env);
  }

  // Health check endpoint
  if (url.pathname === '/xrpc/_health') {
    return Response.json({ version: '0.1.0' });
  }

  // Root path - ASCII art
  if (url.pathname === '/') {
    const ascii = `
 ██████╗  ██████╗  ███████╗          ██╗ ███████╗
 ██╔══██╗ ██╔══██╗ ██╔════╝          ██║ ██╔════╝
 ██████╔╝ ██║  ██║ ███████╗          ██║ ███████╗
 ██╔═══╝  ██║  ██║ ╚════██║     ██   ██║ ╚════██║
 ██║      ██████╔╝ ███████║ ██╗ ╚█████╔╝ ███████║
 ╚═╝      ╚═════╝  ╚══════╝ ╚═╝  ╚════╝  ╚══════╝

 ATProto PDS on Cloudflare Workers
`;
    return new Response(ascii, {
      headers: { 'Content-Type': 'text/plain; charset=utf-8' },
    });
  }

  // On init, register this DID with the default instance (requires ?did= param, no auth yet)
  if (url.pathname === '/init' && request.method === 'POST') {
    const did = url.searchParams.get('did');
    if (!did) {
      return errorResponse('InvalidRequest', 'missing did param', 400);
    }
    const body = await request.json();

    // Register with default instance for discovery
    const defaultId = env.PDS.idFromName('default');
    const defaultPds = env.PDS.get(defaultId);
    await defaultPds.fetch(
      new Request('http://internal/register-did', {
        method: 'POST',
        body: JSON.stringify({ did }),
      }),
    );

    // Register handle if provided
    if (body.handle) {
      await defaultPds.fetch(
        new Request('http://internal/register-handle', {
          method: 'POST',
          body: JSON.stringify({ did, handle: body.handle }),
        }),
      );
    }

    // Forward to the actual PDS instance
    const id = env.PDS.idFromName(did);
    const pds = env.PDS.get(id);
    return pds.fetch(
      new Request(request.url, {
        method: 'POST',
        headers: request.headers,
        body: JSON.stringify(body),
      }),
    );
  }

  // Unknown endpoint
  return errorResponse('NotFound', 'Endpoint not found', 404);
}