// @pdsjs/cloudflare - Cloudflare Workers + Durable Objects adapter

import { secp256k1 } from '@noble/curves/secp256k1';
import { accountApp } from '@pdsjs/account-ui';
import { createS3FetchBackupTarget } from '@pdsjs/blobs-s3';
import { PersonalDataServer } from '@pdsjs/core';
import { APP_PASSWORD_SCHEMA_SQL } from '@pdsjs/core/app-password';
import { SESSION_SCHEMA_SQL } from '@pdsjs/core/auth';
import {
  BLOB_FACETS_APPS_SQL,
  BLOB_FACETS_KINDS_SQL,
  BLOB_FACETS_TOTAL_SQL,
  BLOB_FACETS_UNUSED_SQL,
  blobListCursor,
  buildBlobListQuery,
} from '@pdsjs/core/blob-sql';
import { createVerifier } from '@pdsjs/core/verify';
import { PASSKEY_SCHEMA_SQL } from '@pdsjs/core/webauthn';
import { createDriveExtension } from '@pdsjs/drive/extension';
import { createGitExtension } from '@pdsjs/git/extension';
import {
  createCachedDidResolver,
  LexiconResolver,
} from '@pdsjs/lexicon-resolver';
import { createNpmExtension } from '@pdsjs/npm/extension';
import { createOciExtension } from '@pdsjs/oci/extension';
import { createSitesExtension } from '@pdsjs/sites/extension';
import { createSpacesExtension } from '@pdsjs/spaces/extension';
import {
  CF_API_PROXY_PREFIX,
  cfApiProxyEnabled,
  handleCfApiProxy,
} from './cf-api-proxy.js';
import { createSpaceStorage } from './space.js';

/** Durable Object storage key holding when hourly maintenance is next due. */
const MAINTENANCE_AT_KEY = 'maintenanceAt';
const MAINTENANCE_INTERVAL_MS = 60 * 60 * 1000;

/**
 * Verify a secp256k1 signature the way the reference stack does: SHA-256 the
 * data, then require a compact, low-S signature. Malformed input returns false
 * rather than throwing. Injected into the verifier for secp256k1 keys — the
 * curve the reference PDS gives every account — since @pdsjs/core/verify only
 * builds in P-256.
 *
 * @param {Uint8Array} publicKey - compressed 33-byte point from a did:key
 * @param {Uint8Array} data - the signed bytes
 * @param {Uint8Array} sig - 64-byte compact r||s
 * @returns {Promise<boolean>}
 */
async function secp256k1Verify(publicKey, data, sig) {
  try {
    const hash = new Uint8Array(
      await crypto.subtle.digest('SHA-256', /** @type {BufferSource} */ (data)),
    );
    return secp256k1.verify(sig, hash, publicKey, {
      format: 'compact',
      lowS: true,
    });
  } catch {
    return false;
  }
}

/**
 * Sign with secp256k1 account keys — the curve the reference PDS gives every
 * account, and the one some third-party services (Go feed generators) are
 * alone able to verify. WebCrypto cannot sign this curve, so core takes the
 * implementation as a port, mirroring the verifier above.
 *
 * @type {import('@pdsjs/core/ports').Secp256k1SignerPort}
 */
const k256Signer = {
  generatePrivateKey: () => secp256k1.utils.randomPrivateKey(),
  getPublicKey: (privateKey, compressed = true) =>
    secp256k1.getPublicKey(privateKey, compressed),
  async sign(privateKey, data) {
    const hash = new Uint8Array(
      await crypto.subtle.digest('SHA-256', /** @type {BufferSource} */ (data)),
    );
    // Low-S, compact r||s: what atproto verifies and what secp256k1Verify
    // above requires.
    return secp256k1.sign(hash, privateKey, { lowS: true }).toCompactRawBytes();
  },
};

/**
 * @typedef {import('@pdsjs/core/ports').BlobPort} BlobPort
 * @typedef {import('@pdsjs/core/ports').ActorStoragePort} ActorStoragePort
 * @typedef {import('@pdsjs/core/ports').SharedStoragePort} SharedStoragePort
 */

// ============================================================================
// D1 Storage Adapter (Durable Object SQLite)
// ============================================================================

/**
 * @template T
 * @typedef {Object} SqlCursor
 * @property {() => T[]} toArray - Get all rows as array
 * @property {() => T|undefined} one - Get single row (throws if multiple)
 * @property {number} rowsRead - Number of rows read
 * @property {number} rowsWritten - Number of rows written
 */

/**
 * @typedef {Object} D1SqlInterface
 * @property {(sql: string, ...params: any[]) => SqlCursor<any>} exec - Execute SQL query
 */

// ============================================================================
// Row type definitions for SQLite queries
// ============================================================================

/** @typedef {{cid: string, value: Uint8Array|ArrayBuffer}} RecordResult */
/** @typedef {{uri: string, cid: string, rkey: string, value: Uint8Array|ArrayBuffer}} RecordRow */
/** @typedef {{collection: string, rkey: string, cid: string}} RecordKeyRow */
/** @typedef {{seq: number, cid: string, rev: string}} CommitRow */
/** @typedef {{seq: number, did: string, commit_cid: string, evt: Uint8Array|ArrayBuffer}} EventRow */
/** @typedef {{mime_type: string, size: number, created_at: number}} BlobMetaRow */
/** @typedef {{cid: string}} CidRow */
/** @typedef {{data: string, expires_at: number}} OAuthRequestRow */
/** @typedef {{data: string}} OAuthTokenRow */

/**
 * Safe wrapper for .one() that returns null instead of throwing when no results
 * Cloudflare Durable Object SQLite throws "Expected exactly one result" when no rows
 * @template T
 * @param {SqlCursor<T>} cursor - SQL cursor from sql.exec()
 * @returns {T|null}
 */
function safeOne(cursor) {
  const rows = cursor.toArray();
  return rows.length > 0 ? rows[0] : null;
}

/** Durable Object SQL binds at most 100 parameters to one statement. */
const MAX_BOUND = 100;

/**
 * @template T
 * @param {T[]} items
 * @returns {T[][]}
 */
function chunks(items) {
  /** @type {T[][]} */
  const out = [];
  for (let i = 0; i < items.length; i += MAX_BOUND) {
    out.push(items.slice(i, i + MAX_BOUND));
  }
  return out;
}

/**
 * @typedef {Object} BlobLike
 * @property {ArrayBuffer} buffer
 * @property {number} byteOffset
 * @property {number} byteLength
 */

/**
 * Normalize BLOB value to Uint8Array
 * Cloudflare D1 returns ArrayBuffer for BLOB columns
 * @param {ArrayBuffer|Uint8Array|BlobLike|null|undefined} value
 * @returns {Uint8Array|null}
 */
function normalizeBlob(value) {
  if (!value) return null;
  if (value instanceof Uint8Array) return value;
  if (value instanceof ArrayBuffer) return new Uint8Array(value);
  if ('buffer' in value && value.buffer instanceof ArrayBuffer)
    return new Uint8Array(value.buffer, value.byteOffset, value.byteLength);
  return /** @type {Uint8Array} */ (value);
}

/**
 * Create storage adapter for Cloudflare D1 (Durable Object SQLite). Exported
 * for the tests, which run it over better-sqlite3.
 * @param {D1SqlInterface} sql - D1 SQL interface from state.storage.sql
 * @returns {ActorStoragePort}
 */
export function createActorStorage(sql) {
  // Initialize schema
  sql.exec(`
    CREATE TABLE IF NOT EXISTS metadata (
      key TEXT PRIMARY KEY,
      value BLOB
    );

    CREATE TABLE IF NOT EXISTS blocks (
      cid TEXT PRIMARY KEY,
      data BLOB
    );

    CREATE TABLE IF NOT EXISTS records (
      uri TEXT PRIMARY KEY,
      cid TEXT,
      collection TEXT,
      rkey TEXT,
      value BLOB
    );
    CREATE INDEX IF NOT EXISTS idx_records_collection ON records(collection, rkey);
    CREATE INDEX IF NOT EXISTS idx_records_cid ON records(cid);

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

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

    CREATE TABLE IF NOT EXISTS blobs (
      cid TEXT PRIMARY KEY,
      mime_type TEXT,
      size INTEGER,
      created_at INTEGER
    );

    CREATE TABLE IF NOT EXISTS record_blobs (
      blob_cid TEXT,
      record_uri TEXT,
      record_time INTEGER,
      PRIMARY KEY (blob_cid, record_uri)
    );
  `);

  // Databases created before record_time existed gain the column in place.
  try {
    sql.exec('ALTER TABLE record_blobs ADD COLUMN record_time INTEGER');
  } catch {
    // Already present
  }

  return {
    async getRecord(uri) {
      const row = /** @type {RecordResult|null} */ (
        safeOne(sql.exec('SELECT cid, value FROM records WHERE uri = ?', uri))
      );
      if (!row) return null;
      const value = normalizeBlob(row.value);
      if (!value) return null;
      return { cid: row.cid, value };
    },

    async listRecords(collection, cursor, limit, reverse = false) {
      // `order`/`compare` are fixed strings, not input — safe to interpolate.
      const order = reverse ? 'DESC' : 'ASC';
      const compare = reverse ? '<' : '>';
      /** @type {RecordRow[]} */
      const rows = cursor
        ? sql
            .exec(
              `SELECT uri, cid, rkey, value FROM records WHERE collection = ? AND rkey ${compare} ? ORDER BY rkey ${order} LIMIT ?`,
              collection,
              cursor,
              limit,
            )
            .toArray()
        : sql
            .exec(
              `SELECT uri, cid, rkey, value FROM records WHERE collection = ? ORDER BY rkey ${order} LIMIT ?`,
              collection,
              limit,
            )
            .toArray();
      return {
        records: rows.map((r) => ({
          uri: r.uri,
          cid: r.cid,
          value: normalizeBlob(r.value) || new Uint8Array(0),
        })),
        cursor: rows.length === limit ? rows[rows.length - 1].rkey : null,
      };
    },

    async listAllRecords() {
      /** @type {RecordKeyRow[]} */
      const rows = sql
        .exec(
          "SELECT collection, rkey, cid FROM records ORDER BY (collection || '/' || rkey)",
        )
        .toArray();
      return rows.map((r) => ({
        key: `${r.collection}/${r.rkey}`,
        cid: r.cid,
      }));
    },

    async putRecord(uri, cid, collection, rkey, value) {
      sql.exec(
        'INSERT OR REPLACE INTO records (uri, cid, collection, rkey, value) VALUES (?, ?, ?, ?, ?)',
        uri,
        cid,
        collection,
        rkey,
        value,
      );
    },

    async deleteRecord(uri) {
      sql.exec('DELETE FROM record_blobs WHERE record_uri = ?', uri);
      sql.exec('DELETE FROM records WHERE uri = ?', uri);
    },

    async getBlock(cid) {
      const row = safeOne(
        sql.exec('SELECT data FROM blocks WHERE cid = ?', cid),
      );
      return row ? normalizeBlob(row.data) : null;
    },

    async putBlock(cid, data) {
      sql.exec(
        'INSERT OR IGNORE INTO blocks (cid, data) VALUES (?, ?)',
        cid,
        data,
      );
    },

    async deleteBlocks(cids) {
      for (const part of chunks(cids)) {
        const marks = part.map(() => '?').join(',');
        sql.exec(`DELETE FROM blocks WHERE cid IN (${marks})`, ...part);
      }
    },

    async listBlocks(cursor, limit) {
      const rows = (
        cursor
          ? sql.exec(
              'SELECT cid, data FROM blocks WHERE cid > ? ORDER BY cid LIMIT ?',
              cursor,
              limit,
            )
          : sql.exec('SELECT cid, data FROM blocks ORDER BY cid LIMIT ?', limit)
      ).toArray();
      const blocks = rows.map((r) => ({
        cid: String(r.cid),
        data: /** @type {Uint8Array} */ (normalizeBlob(r.data)),
      }));
      const last = blocks.length === limit ? blocks[limit - 1].cid : null;
      return { blocks, cursor: last };
    },

    async recordCidsAmong(cids) {
      /** @type {string[]} */
      const found = [];
      for (const part of chunks(cids)) {
        const marks = part.map(() => '?').join(',');
        const rows = sql
          .exec(
            `SELECT DISTINCT cid FROM records WHERE cid IN (${marks})`,
            ...part,
          )
          .toArray();
        found.push(...rows.map((r) => String(r.cid)));
      }
      return found;
    },

    async getBlocksCollected() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'blocksCollected'"),
      );
      return String(row?.value) === '1';
    },

    async setBlocksCollected(collected) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('blocksCollected', ?)",
        collected ? '1' : '0',
      );
    },

    async getLatestCommit() {
      const row = /** @type {CommitRow|null} */ (
        safeOne(
          sql.exec(
            'SELECT seq, cid, rev FROM commits ORDER BY seq DESC LIMIT 1',
          ),
        )
      );
      return row || null;
    },

    async putCommit(seq, cid, rev, prev) {
      sql.exec(
        'INSERT INTO commits (seq, cid, rev, prev) VALUES (?, ?, ?, ?)',
        seq,
        cid,
        rev,
        prev,
      );
    },

    async getEvents(cursor, limit) {
      /** @type {EventRow[]} */
      const rows = cursor
        ? sql
            .exec(
              'SELECT seq, did, commit_cid, evt FROM seq_events WHERE seq > ? ORDER BY seq LIMIT ?',
              cursor,
              limit,
            )
            .toArray()
        : sql
            .exec(
              'SELECT seq, did, commit_cid, evt FROM seq_events ORDER BY seq LIMIT ?',
              limit,
            )
            .toArray();
      return {
        events: rows.map((r) => ({
          seq: r.seq,
          commit_cid: r.commit_cid,
          evt: normalizeBlob(r.evt) || new Uint8Array(0),
        })),
        cursor: rows.length > 0 ? rows[rows.length - 1].seq : cursor,
      };
    },

    async putEvent(seq, did, commitCid, evt) {
      sql.exec(
        'INSERT INTO seq_events (seq, did, commit_cid, evt) VALUES (?, ?, ?, ?)',
        seq,
        did,
        commitCid,
        evt,
      );
    },

    async getEventsBefore(before, limit) {
      /** @type {EventRow[]} */
      const rows = sql
        .exec(
          'SELECT seq, evt FROM seq_events WHERE seq < ? ORDER BY seq DESC LIMIT ?',
          before,
          limit,
        )
        .toArray();
      return rows.map((r) => ({
        seq: r.seq,
        evt: normalizeBlob(r.evt) || new Uint8Array(0),
      }));
    },

    async getLatestSeq() {
      // Max across both sequenced tables: a crash between putCommit and
      // putEvent leaves a commit whose seq the events table never saw, and
      // counting only events would hand that seq out again forever.
      const rows = sql
        .exec(
          'SELECT MAX(s) as seq FROM (SELECT MAX(seq) as s FROM seq_events UNION ALL SELECT MAX(seq) as s FROM commits)',
        )
        .toArray();
      return /** @type {number|null} */ (rows[0]?.seq) ?? 0;
    },

    async getBlob(cid) {
      const row = safeOne(
        sql.exec(
          'SELECT mime_type, size, created_at FROM blobs WHERE cid = ?',
          cid,
        ),
      );
      return row
        ? { mimeType: row.mime_type, size: row.size, createdAt: row.created_at }
        : null;
    },

    async putBlob(cid, mimeType, size) {
      sql.exec(
        'INSERT OR REPLACE INTO blobs (cid, mime_type, size, created_at) VALUES (?, ?, ?, ?)',
        cid,
        mimeType,
        size,
        Date.now(),
      );
    },

    async deleteBlob(cid) {
      sql.exec('DELETE FROM blobs WHERE cid = ?', cid);
    },

    async listBlobs(cursor, limit) {
      const rows = cursor
        ? sql
            .exec(
              'SELECT cid FROM blobs WHERE cid > ? ORDER BY cid LIMIT ?',
              cursor,
              limit,
            )
            .toArray()
        : sql
            .exec('SELECT cid FROM blobs ORDER BY cid LIMIT ?', limit)
            .toArray();
      return {
        cids: rows.map((r) => r.cid),
        cursor: rows.length === limit ? rows[rows.length - 1].cid : null,
      };
    },

    async listBlobDetails(cursor, limit, options = {}) {
      const query = buildBlobListQuery(cursor, limit, options);
      const rows = sql.exec(query.sql, ...query.params).toArray();
      const blobs = rows.map((r) => ({
        cid: /** @type {string} */ (r.cid),
        mimeType: /** @type {string} */ (r.mime_type),
        size: /** @type {number} */ (r.size),
        createdAt: /** @type {number} */ (r.created_at),
        time: /** @type {number} */ (r.time_ms),
        timeFromRecord: r.time_known === 1,
        records: r.uris ? String(r.uris).split(',') : [],
      }));
      return {
        blobs,
        cursor:
          blobs.length === limit
            ? blobListCursor(blobs[blobs.length - 1], options)
            : null,
      };
    },

    async getBlobFacets() {
      /** @param {{count?: unknown, bytes?: unknown}|null} row */
      const bucket = (row) => ({
        count: /** @type {number} */ (row?.count ?? 0),
        bytes: /** @type {number} */ (row?.bytes ?? 0),
      });
      const kinds = sql
        .exec(BLOB_FACETS_KINDS_SQL)
        .toArray()
        .map((r) => ({
          kind: /** @type {import('@pdsjs/core/ports').BlobKind} */ (r.kind),
          ...bucket(r),
        }));
      const apps = sql
        .exec(BLOB_FACETS_APPS_SQL)
        .toArray()
        .map((r) => ({
          authority: /** @type {string} */ (r.authority),
          ...bucket(r),
        }));
      return {
        total: bucket(safeOne(sql.exec(BLOB_FACETS_TOTAL_SQL))),
        unused: bucket(safeOne(sql.exec(BLOB_FACETS_UNUSED_SQL))),
        kinds,
        apps,
      };
    },

    async getOrphanedBlobs(olderThan) {
      const rows = sql
        .exec(
          'SELECT b.cid FROM blobs b LEFT JOIN record_blobs rb ON b.cid = rb.blob_cid WHERE rb.blob_cid IS NULL AND b.created_at < ?',
          olderThan,
        )
        .toArray();
      return rows.map((r) => r.cid);
    },

    async linkBlobToRecord(blobCid, recordUri, recordTime = null) {
      sql.exec(
        'INSERT OR REPLACE INTO record_blobs (blob_cid, record_uri, record_time) VALUES (?, ?, ?)',
        blobCid,
        recordUri,
        recordTime,
      );
    },

    async unlinkBlobsFromRecord(recordUri) {
      sql.exec('DELETE FROM record_blobs WHERE record_uri = ?', recordUri);
    },

    async listBlobRecordUris(blobCid) {
      const rows = sql
        .exec('SELECT record_uri FROM record_blobs WHERE blob_cid = ?', blobCid)
        .toArray();
      return rows.map((r) => String(r.record_uri));
    },

    async getDid() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'did'"),
      );
      if (!row) return null;
      if (typeof row.value === 'string') return row.value;
      if (row.value instanceof ArrayBuffer)
        return new TextDecoder().decode(row.value);
      return String(row.value);
    },

    async setDid(did) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('did', ?)",
        did,
      );
    },

    async getPrivateKey() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'privateKey'"),
      );
      return row ? normalizeBlob(row.value) : null;
    },

    async setPrivateKey(key) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('privateKey', ?)",
        key,
      );
    },

    async getKemKey() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'kemKey'"),
      );
      return row ? normalizeBlob(row.value) : null;
    },

    async setKemKey(key) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('kemKey', ?)",
        key,
      );
    },

    async getHandle() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'handle'"),
      );
      if (!row) return null;
      if (typeof row.value === 'string') return row.value;
      if (row.value instanceof ArrayBuffer)
        return new TextDecoder().decode(row.value);
      return String(row.value);
    },

    async getEmail() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'email'"),
      );
      if (!row) return null;
      const raw =
        row.value instanceof ArrayBuffer
          ? new TextDecoder().decode(row.value)
          : String(row.value);
      try {
        return JSON.parse(raw);
      } catch {
        return null;
      }
    },

    async setEmail(state) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('email', ?)",
        JSON.stringify(state),
      );
    },

    async getAccountPassword() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'accountPassword'"),
      );
      if (!row) return null;
      const raw =
        row.value instanceof ArrayBuffer
          ? new TextDecoder().decode(row.value)
          : String(row.value);
      try {
        return JSON.parse(raw);
      } catch {
        return null;
      }
    },

    async setAccountPassword(state) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('accountPassword', ?)",
        JSON.stringify(state),
      );
    },

    async getPlcToken() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'plcToken'"),
      );
      if (!row) return null;
      const raw =
        row.value instanceof ArrayBuffer
          ? new TextDecoder().decode(row.value)
          : String(row.value);
      try {
        return JSON.parse(raw);
      } catch {
        return null;
      }
    },

    async setPlcToken(token) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('plcToken', ?)",
        JSON.stringify(token),
      );
    },

    async getAccountStatus() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'accountStatus'"),
      );
      if (!row) return null;
      if (typeof row.value === 'string') return row.value;
      if (row.value instanceof ArrayBuffer)
        return new TextDecoder().decode(row.value);
      return String(row.value);
    },

    async setAccountStatus(status) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('accountStatus', ?)",
        status,
      );
    },

    async setHandle(handle) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('handle', ?)",
        handle,
      );
    },

    // Preferences storage
    async getPreferences() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'preferences'"),
      );
      return row ? JSON.parse(row.value) : [];
    },

    async setPreferences(preferences) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('preferences', ?)",
        JSON.stringify(preferences),
      );
    },

    // Backup settings and run history, one JSON document
    async getBackupState() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'backupState'"),
      );
      if (!row) return null;
      const raw =
        row.value instanceof ArrayBuffer
          ? new TextDecoder().decode(row.value)
          : String(row.value);
      try {
        return JSON.parse(raw);
      } catch {
        return null;
      }
    },

    async setBackupState(state) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('backupState', ?)",
        JSON.stringify(state),
      );
    },

    // The account app's own preferences, one JSON document
    async getAccountUiState() {
      const row = safeOne(
        sql.exec("SELECT value FROM metadata WHERE key = 'accountUiState'"),
      );
      if (!row) return null;
      const raw =
        row.value instanceof ArrayBuffer
          ? new TextDecoder().decode(row.value)
          : String(row.value);
      try {
        return JSON.parse(raw);
      } catch {
        return null;
      }
    },

    async setAccountUiState(state) {
      sql.exec(
        "INSERT OR REPLACE INTO metadata (key, value) VALUES ('accountUiState', ?)",
        JSON.stringify(state),
      );
    },
  };
}

// ============================================================================
// R2 Blob Adapter
// ============================================================================

// Every multipart part must be this size except the last: R2 requires the
// parts of one upload to be uniform. R2's minimum is 5 MiB.
const R2_PART_SIZE = 8 * 1024 * 1024;

// R2 accepts about one write per second to one key. A faster write fails with
// error 10058. These delays let the earlier write finish.
const R2_SAME_KEY_RETRY_MS = [250, 500, 1000, 2000];

/**
 * Run an R2 write, again after a delay while R2 refuses it with 10058.
 * @template T
 * @param {() => Promise<T>} write
 * @returns {Promise<T>}
 */
async function retrySameKeyLimit(write) {
  for (const delay of R2_SAME_KEY_RETRY_MS) {
    try {
      return await write();
    } catch (err) {
      if (!String(/** @type {Error} */ (err)?.message).includes('(10058)')) {
        throw err;
      }
      await new Promise((resolve) => setTimeout(resolve, delay));
    }
  }
  return write();
}

/**
 * @param {Uint8Array[]} chunks
 * @param {number} size
 * @returns {Uint8Array}
 */
function joinChunks(chunks, size) {
  const out = new Uint8Array(size);
  let at = 0;
  for (const chunk of chunks) {
    out.set(chunk, at);
    at += chunk.length;
  }
  return out;
}

/**
 * The runtime's native SHA-256 behind the core Hasher interface.
 * @returns {import('@pdsjs/core/ports').Hasher}
 */
function createDigestStreamHasher() {
  // DigestStream hangs off the Workers crypto global. The Node Crypto type
  // this workspace also loads wins the global merge and lacks it, hence the
  // cast.
  const workersCrypto = /** @type {{DigestStream: typeof DigestStream}} */ (
    /** @type {unknown} */ (crypto)
  );
  const stream = new workersCrypto.DigestStream('SHA-256');
  const writer = stream.getWriter();
  return {
    /** @param {Uint8Array} chunk */
    update(chunk) {
      return writer.write(chunk);
    },
    async digest() {
      await writer.close();
      return new Uint8Array(await stream.digest);
    },
  };
}

/**
 * Create R2 blob adapter
 * @param {R2Bucket} bucket - R2 bucket binding
 * @returns {BlobPort}
 */
export function createBlobs(bucket) {
  // A blob's key is its CID, so clients that upload the same bytes write the
  // same key. R2 refuses concurrent writes to one key. One Durable Object
  // serves the account, so this map sees every write to its keys.
  /** @type {Map<string, Promise<void>>} */
  const keyQueues = new Map();

  /**
   * Run `write` after the writes to `key` queued before it.
   * @param {string} key
   * @param {() => Promise<void>} write
   * @returns {Promise<void>}
   */
  function onKey(key, write) {
    const run = (keyQueues.get(key) ?? Promise.resolve()).then(write);
    const settled = run.then(
      () => {},
      () => {},
    );
    keyQueues.set(key, settled);
    settled.then(() => {
      if (keyQueues.get(key) === settled) keyQueues.delete(key);
    });
    return run;
  }

  /**
   * Put the blob at `key` unless R2 holds it. The same CID means the same
   * bytes, so a second put changes nothing.
   * @param {string} key
   * @param {() => Promise<unknown>} put
   */
  function putOnce(key, put) {
    return onKey(key, async () => {
      if (await bucket.head(key)) return;
      await retrySameKeyLimit(put);
    });
  }

  return {
    /**
     * @param {string} did
     * @param {string} cid
     */
    async get(did, cid) {
      const obj = await bucket.get(`${did}/${cid}`);
      if (!obj) return null;
      const data = new Uint8Array(await obj.arrayBuffer());
      const mimeType =
        obj.customMetadata?.mime_type || 'application/octet-stream';
      return { data, mimeType };
    },

    /**
     * @param {string} did
     * @param {string} cid
     * @param {Uint8Array} data
     * @param {string} mimeType
     */
    async put(did, cid, data, mimeType) {
      await putOnce(`${did}/${cid}`, () =>
        bucket.put(`${did}/${cid}`, data, {
          customMetadata: { mime_type: mimeType },
        }),
      );
    },

    /**
     * @param {string} did
     * @param {string} cid
     */
    async delete(did, cid) {
      const key = `${did}/${cid}`;
      await onKey(key, () => retrySameKeyLimit(() => bucket.delete(key)));
    },

    /**
     * @param {string} did
     * @param {string} cid
     * @param {{start: number, end: number}} [range] - Inclusive byte positions
     */
    async getStream(did, cid, range) {
      const obj = /** @type {R2ObjectBody|null} */ (
        await bucket.get(
          `${did}/${cid}`,
          range
            ? {
                range: {
                  offset: range.start,
                  length: range.end - range.start + 1,
                },
              }
            : undefined,
        )
      );
      if (!obj) return null;
      return {
        body: obj.body,
        mimeType: obj.customMetadata?.mime_type || 'application/octet-stream',
      };
    },

    /**
     * Bytes stage under a temporary key and land under the CID at commit. A
     * CID contains no slash, so `tmp/` keys are outside the space `get`
     * addresses. The R2 binding has no server-side copy, so commit reads the
     * staged object back and writes it to the final key as a stream; one part
     * buffer is the most this holds in memory. A blob within one part skips
     * the staging entirely and lands with a single put.
     * @param {string} did
     */
    async createWriter(did) {
      const tempKey = `${did}/tmp/${crypto.randomUUID()}`;
      /** @type {R2MultipartUpload|null} */
      let upload = null;
      /** @type {R2UploadedPart[]} */
      const parts = [];
      /** @type {Uint8Array[]} */
      let pending = [];
      let pendingSize = 0;

      async function flushPart() {
        if (!upload) upload = await bucket.createMultipartUpload(tempKey);
        const buffered = joinChunks(pending, pendingSize);
        const body = buffered.subarray(0, R2_PART_SIZE);
        const rest = buffered.subarray(R2_PART_SIZE);
        parts.push(await upload.uploadPart(parts.length + 1, body));
        pending = rest.length ? [rest] : [];
        pendingSize = rest.length;
      }

      return {
        /** @param {Uint8Array} chunk */
        async write(chunk) {
          pending.push(chunk);
          pendingSize += chunk.length;
          while (pendingSize >= R2_PART_SIZE) await flushPart();
        },

        /**
         * @param {string} cid
         * @param {string} mimeType
         */
        async commit(cid, mimeType) {
          const finalKey = `${did}/${cid}`;
          const customMetadata = { mime_type: mimeType };
          if (!upload) {
            const data = joinChunks(pending, pendingSize);
            await putOnce(finalKey, () =>
              bucket.put(finalKey, data, { customMetadata }),
            );
            return;
          }
          if (pendingSize > 0) {
            parts.push(
              await upload.uploadPart(
                parts.length + 1,
                joinChunks(pending, pendingSize),
              ),
            );
            pending = [];
            pendingSize = 0;
          }
          await upload.complete(parts);
          // A retry reads the staged object again, since a stream is read once.
          await putOnce(finalKey, async () => {
            const staged = await bucket.get(tempKey);
            if (!staged) throw new Error('The staged blob is gone.');
            const fixed = new FixedLengthStream(staged.size);
            const piped = staged.body.pipeTo(fixed.writable);
            await Promise.all([
              bucket.put(finalKey, fixed.readable, { customMetadata }),
              piped,
            ]);
          });
          await bucket.delete(tempKey);
        },

        async abort() {
          if (upload) await upload.abort();
          pending = [];
          pendingSize = 0;
        },
      };
    },
  };
}

/**
 * Backups in a second R2 bucket — separate from the live blob bucket, which
 * is the point of a backup. Core deletes anything under `snapshots/` and
 * `blobs/` that its retention policy no longer wants, so give it a bucket of
 * its own.
 * @param {R2Bucket} bucket - R2 bucket binding
 * @returns {import('@pdsjs/core/ports').BackupTargetPort}
 */
function createBackupTarget(bucket) {
  return {
    /**
     * @param {string} key
     * @param {Uint8Array} data
     * @param {string} [contentType]
     */
    async put(key, data, contentType) {
      await bucket.put(
        key,
        data,
        contentType ? { httpMetadata: { contentType } } : undefined,
      );
    },

    /** @param {string} key */
    async get(key) {
      const obj = await bucket.get(key);
      return obj ? new Uint8Array(await obj.arrayBuffer()) : null;
    },

    /** @param {string} prefix */
    async list(prefix) {
      /** @type {string[]} */
      const keys = [];
      /** @type {string|undefined} */
      let cursor;
      do {
        const result = await bucket.list({ prefix, cursor });
        for (const object of result.objects) keys.push(object.key);
        cursor = result.truncated ? result.cursor : undefined;
      } while (cursor);
      return keys;
    },

    /** @param {string} key */
    async delete(key) {
      await bucket.delete(key);
    },
  };
}

// ============================================================================
// WebSocket Adapter
// ============================================================================

/**
 * Create WebSocket port for Cloudflare Workers with Durable Object hibernation support
 * @param {DurableObjectState} state - Durable Object state for hibernation API
 * @returns {import('@pdsjs/core/ports').WebSocketPort}
 */
export function createWebSocket(state) {
  return {
    isUpgrade(request) {
      return request.headers.get('Upgrade') === 'websocket';
    },

    upgrade(_request, onConnect) {
      const pair = new WebSocketPair();
      const [client, server] = Object.values(pair);

      // Use hibernation API - keeps WebSocket alive across DO hibernation
      state.acceptWebSocket(server);

      // Call onConnect with sender interface
      onConnect({
        send(data) {
          try {
            server.send(data);
          } catch {
            // Client may have disconnected
          }
        },
        close() {
          try {
            server.close();
          } catch {
            // Already closed
          }
        },
      });

      return new Response(null, { status: 101, webSocket: client });
    },

    broadcast(data) {
      // Use hibernation API to get all connected WebSockets
      for (const ws of state.getWebSockets()) {
        try {
          ws.send(data);
        } catch {
          // The subscriber is gone. Close rather than swallow: an unclosed
          // socket stays in getWebSockets() and keeps absorbing the firehose,
          // so the server holds a connection it can no longer deliver on
          // while the subscriber, never told to go away, never reconnects.
          try {
            ws.close(1011, 'send failed');
          } catch {
            // Already closed.
          }
        }
      }
    },
  };
}

// ============================================================================
// Durable Object and Worker Entry Point
// ============================================================================

/**
 * The Email Service `send_email` binding, narrowed to the fields this adapter
 * uses. Cloudflare's own `SendEmail` type is the legacy forwarding envelope and
 * does not carry the outbound payload; the runtime takes `from` as a plain
 * address string, not an object.
 * @typedef {Object} CloudflareEmailer
 * @property {(message: {to: string, from: string, subject: string, text: string, html?: string}) => Promise<void>} send
 */

/**
 * Adapt the Email Service binding to core's EmailPort: core sends `to`,
 * `subject`, `text` and `html` but leaves the sender up to the platform, and
 * Cloudflare requires one.
 * @param {CloudflareEmailer} binding - The `EMAIL` binding from the worker's env
 * @param {string} from - The From address; its domain must be onboarded for Email Sending
 * @returns {import('@pdsjs/core/ports').EmailPort}
 */
export function createEmailer(binding, from) {
  return {
    async send({ to, subject, text, html }) {
      await binding.send({ to, from, subject, text, html });
    },
  };
}

/**
 * @typedef {Object} Env
 * @property {R2Bucket} [BLOBS] - R2 bucket for blob storage
 * @property {R2Bucket} [BLOB_BUCKET] - Alternative R2 bucket binding
 * @property {R2Bucket} [BACKUPS] - R2 bucket scheduled backups land in; absent, the backups page reports backups as unavailable
 * @property {string} [BACKUP_S3_ENDPOINT] - S3-compatible endpoint for off-Cloudflare backups (Bunny, B2, S3, MinIO); with the four other BACKUP_S3_* values set, this wins over the BACKUPS binding
 * @property {string} [BACKUP_S3_BUCKET] - Bucket (or storage zone) name
 * @property {string} [BACKUP_S3_REGION] - Signing region; some providers accept anything here
 * @property {string} [BACKUP_S3_ACCESS_KEY] - Secret: S3 access key id
 * @property {string} [BACKUP_S3_SECRET_KEY] - Secret: S3 secret access key
 * @property {string} JWT_SECRET - Secret for JWT signing
 * @property {string} [HOSTNAME] - PDS hostname
 * @property {string} [APPVIEW_URL] - AppView URL for proxying
 * @property {string} [APPVIEW_DID] - AppView DID
 * @property {string} [RELAY_URL] - Relay URL for firehose notifications
 * @property {string} [PLC_URL] - PLC directory URL for identity operations
 * @property {string} [PDS_PASSWORD] - Password for createSession
 * @property {string} [PDS_UPDATE_URL] - Where a person updates this deployment; the account page links there when a newer release is published
 * @property {string} [PDS_SCRIPT_NAME] - The worker script name this deployment runs as; the update link carries it so the update page can find the worker behind a custom domain
 * @property {string} [PDS_EXPERIMENTAL_CF_API_PROXY] - "true" forwards /cf-api/ paths to api.cloudflare.com for the site-hosted setup wizard
 * @property {string} [PDS_BLOB_UPLOAD_LIMIT] - Max blob upload size in bytes (default 50 MiB; a Worker accepts a request body of 100 MB on the Free and Pro plans)
 * @property {string} [PDS_ENABLE_SPACES] - "true" enables permissioned data
 * @property {string} [PDS_EXPERIMENTAL_GIT_HTTP] - "true" serves read-only git smart HTTP under /git/
 * @property {string} [PDS_EXPERIMENTAL_OCI] - "true" serves an OCI image registry under /v2/; pulls are anonymous, pushes take Basic credentials createSession accepts
 * @property {string} [PDS_EXPERIMENTAL_NPM] - "true" serves an npm registry under /npm/; installs are anonymous, publishes take credentials createSession accepts
 * @property {string} [PDS_EXPERIMENTAL_GIT_BROWSE] - "false" turns off file browsing of hosted git repositories on the account page (on by default)
 * @property {string} [PDS_EXPERIMENTAL_SITE_DOMAIN] - Apex domain static sites serve under; presence enables the sites feature. Site hostnames must also be routed to this Worker (apex route plus a wildcard zone route).
 * @property {string} [PDS_EXPERIMENTAL_SITE_APEX_NAME] - Site name served at the apex (default "home")
 * @property {string} [PDS_EXPERIMENTAL_QUERIES] - "true" serves dev.pdsjs.query.run over this repo's dev.pdsjs.query.def records
 * @property {string} [PDS_EXPERIMENTAL_DRIVE] - "true" backs the account page's Drive section over dev.pdsjs.drive.file records
 * @property {string} [PDS_EMAIL_FROM] - The From address account mail is sent as; its domain must be onboarded for Email Sending. Absent, no-reply@HOSTNAME is used. Neither set, mail cannot be sent and the account page reports email as unavailable.
 * @property {CloudflareEmailer} [EMAIL] - Email Service send_email binding. Present only when the binding is configured; without it mail cannot be sent.
 * @property {DurableObjectNamespace} PDS - Durable Object namespace
 */

/**
 * Durable Object class for per-user PDS instances
 */
export class PDSDurableObject {
  /**
   * @param {DurableObjectState} state
   * @param {Env} env
   */
  constructor(state, env) {
    this.state = state;
    this.env = env;

    // Initialize actor storage adapter from Durable Object SQLite
    const actorStorage = createActorStorage(state.storage.sql);

    // Shared account state — sessions, OAuth tokens, app passwords, passkeys —
    // lives in the Durable Object's own SQLite. It is strongly consistent, so a
    // read always sees its own writes.
    const sharedStorage = createSharedStorageSql(state.storage.sql);

    // Initialize blob adapter (R2 bucket)
    const blobs = createBlobs(
      /** @type {R2Bucket} */ (env.BLOBS || env.BLOB_BUCKET),
    );

    // Backups: an S3-compatible bucket elsewhere beats an R2 bucket here —
    // a copy outside Cloudflare survives what a same-account copy cannot.
    // With neither configured the feature reports itself unavailable.
    const backupTarget =
      env.BACKUP_S3_ENDPOINT &&
      env.BACKUP_S3_BUCKET &&
      env.BACKUP_S3_ACCESS_KEY &&
      env.BACKUP_S3_SECRET_KEY
        ? createS3FetchBackupTarget({
            endpoint: env.BACKUP_S3_ENDPOINT,
            bucket: env.BACKUP_S3_BUCKET,
            region: env.BACKUP_S3_REGION || 'us-east-1',
            accessKeyId: env.BACKUP_S3_ACCESS_KEY,
            secretAccessKey: env.BACKUP_S3_SECRET_KEY,
          })
        : env.BACKUPS
          ? createBackupTarget(env.BACKUPS)
          : undefined;

    // Initialize WebSocket adapter with hibernation support
    const webSocket = createWebSocket(state);

    // Initialize lexicon resolver for record validation
    // Wrap fetch to preserve `this` binding in Cloudflare Workers
    const lexiconResolver = new LexiconResolver({
      fetch: (url, init) => fetch(url, init),
    });

    // Signature verification, shared by createAccount (migration service
    // tokens) and spaces (credentials). P-256 is built into core; secp256k1 —
    // the curve the reference PDS gives every account — is supplied here via
    // @noble/curves. The resolver returns full DID documents; fetch is wrapped
    // to preserve its `this` binding, as elsewhere in this file.
    const verifier = createVerifier({ secp256k1: secp256k1Verify });
    // Built in the Durable Object constructor, so one cache serves every
    // request this object handles. Every space request verifies a credential,
    // which reads its issuer's key.
    const didResolver = createCachedDidResolver({
      plcUrl: env.PLC_URL,
      fetch: (url, init) => fetch(url, init),
    });

    // Permissioned data (proposal 0016), off by default. The storage is this
    // platform's; the space endpoints, the admin port and the account page's
    // space routes come from the spaces extension mounted below. Absent the
    // flag the space endpoints do not exist.
    /** @type {import('@pdsjs/core/ports').SpaceStoragePort|undefined} */
    let spaceStorageRef;
    if (env.PDS_ENABLE_SPACES === 'true') {
      // state.storage as well as .sql: commitSpaceWrite needs transactionSync,
      // which Durable Object SQL exposes on the storage object rather than
      // through exec.
      spaceStorageRef = createSpaceStorage(state.storage.sql, state.storage);
    }

    // Account mail. Both a binding and a sender are needed: Cloudflare requires
    // a From address whose domain is onboarded for Email Sending. The binding
    // alone can store an address but never prove anyone can read it, so mail is
    // only enabled when the sender is resolvable.
    /** @type {import('@pdsjs/core/ports').EmailPort|undefined} */
    const emailer =
      env.EMAIL && (env.PDS_EMAIL_FROM || env.HOSTNAME)
        ? createEmailer(
            env.EMAIL,
            env.PDS_EMAIL_FROM || `no-reply@${env.HOSTNAME}`,
          )
        : undefined;

    // Create PDS instance with port injection
    this.pds = new PersonalDataServer({
      // One alarm serves the hourly maintenance and any earlier extension work,
      // so a wakeup only ever moves the alarm closer.
      scheduleWakeup: async (at) => {
        const current = await state.storage.getAlarm();
        if (!current || at < current) await state.storage.setAlarm(at);
      },
      waitUntil: (work) => state.waitUntil(work),
      actorStorage,
      sharedStorage,
      blobs,
      webSocket,
      jwtSecret: env.JWT_SECRET,
      hostname: env.HOSTNAME,
      appviewUrl: env.APPVIEW_URL,
      appviewDid: env.APPVIEW_DID,
      relayUrl: env.RELAY_URL,
      plcUrl: env.PLC_URL,
      password: env.PDS_PASSWORD,
      emailer,
      updateUrl: env.PDS_UPDATE_URL,
      scriptName: env.PDS_SCRIPT_NAME,
      blobUploadLimit: env.PDS_BLOB_UPLOAD_LIMIT
        ? Number(env.PDS_BLOB_UPLOAD_LIMIT)
        : undefined,
      hasher: createDigestStreamHasher,
      lexiconResolver,
      spaceStorage: spaceStorageRef,
      accountApp,
      verifier,
      didResolver,
      k256Signer,
      backupTarget,
      // Feature packages on the extension seam. Static imports, the same as
      // spaces above.
      extensions: [
        ...(env.PDS_ENABLE_SPACES === 'true' ? [createSpacesExtension()] : []),
        createSitesExtension({
          domain: env.PDS_EXPERIMENTAL_SITE_DOMAIN,
          apexName: env.PDS_EXPERIMENTAL_SITE_APEX_NAME,
          queries: env.PDS_EXPERIMENTAL_QUERIES === 'true',
        }),
        createGitExtension({
          http: env.PDS_EXPERIMENTAL_GIT_HTTP === 'true',
          browse: env.PDS_EXPERIMENTAL_GIT_BROWSE !== 'false',
        }),
        ...(env.PDS_EXPERIMENTAL_OCI === 'true' ? [createOciExtension()] : []),
        ...(env.PDS_EXPERIMENTAL_NPM === 'true' ? [createNpmExtension()] : []),
        ...(env.PDS_EXPERIMENTAL_DRIVE === 'true'
          ? [createDriveExtension()]
          : []),
      ],
    });

    // Arm the alarm if not already set. Maintenance runs hourly — the same
    // cadence as the Node and Deno cleanup intervals — and each run decides for
    // itself whether cleanup or a scheduled backup is due. Extension work can
    // bring the alarm earlier.
    state.blockConcurrencyWhile(async () => {
      const currentAlarm = await state.storage.getAlarm();
      if (!currentAlarm) {
        await state.storage.setAlarm(Date.now() + 60 * 60 * 1000);
      }
    });

    // An account that predates KEM keys gets one as soon as this object
    // starts, as the Node and Deno servers do at startup, rather than at the
    // next hourly maintenance; until then no syncer can read its space repos.
    // Once the key exists this is a single storage read.
    void this.pds.maybeBackfillKemKey();
  }

  /**
   * Handle incoming request
   * @param {Request} request
   * @returns {Promise<Response>}
   */
  async fetch(request) {
    return this.pds.fetch(request);
  }

  /**
   * The alarm: hourly maintenance when it is due (blob cleanup, then a backup
   * if one is due), and extension work on every wakeup. Mirrors the intervals
   * the Node and Deno servers run.
   */
  async alarm() {
    const now = Date.now();
    let maintenanceAt =
      /** @type {number|undefined} */ (
        await this.state.storage.get(MAINTENANCE_AT_KEY)
      ) ?? 0;
    /** @type {number|null} */
    let next = null;
    try {
      if (now >= maintenanceAt) {
        // Advanced before the run, so a failing run waits an hour like a
        // passing one rather than refiring at once.
        maintenanceAt = now + MAINTENANCE_INTERVAL_MS;
        await this.state.storage.put(MAINTENANCE_AT_KEY, maintenanceAt);
        await this.pds.runBlobCleanup();
        await this.pds.maybeAnnounceIdentity();
        await this.pds.maybeBackfillKemKey();
        await this.pds.maybeRunScheduledBackup();
      }
    } finally {
      try {
        next = await this.pds.runExtensionTicks();
      } finally {
        // Re-arm no matter what: a failed run must not end the schedule.
        await this.state.storage.setAlarm(
          next === null ? maintenanceAt : Math.min(maintenanceAt, next),
        );
      }
    }
  }

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

  /**
   * Handle WebSocket close (hibernation API callback)
   * @param {WebSocket} ws
   * @param {number} code
   * @param {string} reason
   * @param {boolean} _wasClean
   */
  async webSocketClose(ws, code, reason, _wasClean) {
    // Finish the closing handshake. Until this runs the socket stays in
    // getWebSockets(), and a firehose broadcast keeps writing to a subscriber
    // that has already gone. (Runtimes on compatibility_date 2026-04-07 or
    // later reply automatically; this server's is older.)
    try {
      ws.close(code, reason);
    } catch {
      // Already closed.
    }
  }

  /**
   * Handle WebSocket errors (hibernation API callback)
   * @param {WebSocket} ws
   * @param {unknown} _error
   */
  async webSocketError(ws, _error) {
    // A socket that errored will never deliver again; drop it so it stops
    // collecting broadcasts.
    try {
      ws.close(1011, 'socket error');
    } catch {
      // Already closed.
    }
  }
}

// ============================================================================
// Shared storage adapter using Durable Object SQLite
// ============================================================================

/** @typedef {string | number | boolean | null | undefined} SqlBind */
/** @typedef {{did: string, handle: string | null, created_at: number}} ActorRow */
/** @typedef {{jti: string, did: string, label: string, scope: string, user_agent: string | null, created_at: string, refreshed_at: string | null, expires_at: number | null, next_jti: string | null}} SessionRow */
/** @typedef {{credential_id: string, did: string, name: string, public_key: string, algorithm: number, sign_count: number, created_at: string, last_used_at: string | null}} PasskeyRow */
/** @typedef {{name: string, privileged: number, created_at: string}} AppPasswordRow */
/** @typedef {{token_id: string, data: string}} OAuthTokenListRow */

/** @param {SessionRow} row @returns {import('@pdsjs/core/ports').SessionRecord} */
function toSessionRecord(row) {
  return {
    jti: row.jti,
    did: row.did,
    label: row.label,
    scope: row.scope,
    userAgent: row.user_agent,
    createdAt: row.created_at,
    refreshedAt: row.refreshed_at,
    expiresAt: row.expires_at ?? undefined,
    nextJti: row.next_jti ?? null,
  };
}

/** @param {PasskeyRow} row @returns {import('@pdsjs/core/ports').PasskeyRecord} */
function toPasskeyRecord(row) {
  return {
    credentialId: row.credential_id,
    did: row.did,
    name: row.name,
    publicKey: row.public_key,
    algorithm: row.algorithm,
    signCount: row.sign_count,
    createdAt: row.created_at,
    lastUsedAt: row.last_used_at,
  };
}

/** @param {AppPasswordRow} row @returns {import('@pdsjs/core/ports').AppPasswordRecord} */
function toAppPassword(row) {
  return {
    name: row.name,
    privileged: row.privileged === 1,
    createdAt: row.created_at,
  };
}

/**
 * Shared storage on the Durable Object's own SQLite — strongly consistent and
 * co-located with the repo, so every account-scoped record (sessions, OAuth,
 * app passwords, passkeys) reads back its own writes immediately. The schema
 * mirrors the Node adapter's, sharing the same core schema constants.
 * @param {D1SqlInterface} sql
 * @returns {SharedStoragePort}
 */
export function createSharedStorageSql(sql) {
  sql.exec(`
    CREATE TABLE IF NOT EXISTS actors (
      did TEXT PRIMARY KEY,
      handle TEXT,
      created_at INTEGER NOT NULL
    );
    CREATE UNIQUE INDEX IF NOT EXISTS actors_handle_lower_idx ON actors(lower(handle));

    CREATE TABLE IF NOT EXISTS oauth_requests (
      request_uri TEXT PRIMARY KEY,
      data TEXT,
      expires_at INTEGER
    );
    CREATE INDEX IF NOT EXISTS oauth_requests_expires_idx ON oauth_requests(expires_at);

    CREATE TABLE IF NOT EXISTS oauth_tokens (
      token_id TEXT PRIMARY KEY,
      did TEXT NOT NULL,
      data TEXT
    );
    CREATE INDEX IF NOT EXISTS oauth_tokens_did_idx ON oauth_tokens(did);

    CREATE TABLE IF NOT EXISTS dpop_jtis (
      jti TEXT PRIMARY KEY,
      expires_at INTEGER
    );
    CREATE INDEX IF NOT EXISTS dpop_jtis_expires_idx ON dpop_jtis(expires_at);
  `);
  sql.exec(APP_PASSWORD_SCHEMA_SQL);
  sql.exec(PASSKEY_SCHEMA_SQL);
  sql.exec(SESSION_SCHEMA_SQL);
  // A session table created before rotations named a successor lacks the
  // column, and CREATE TABLE IF NOT EXISTS will not add it. SQLite has no
  // ADD COLUMN IF NOT EXISTS, so a re-run just throws.
  try {
    sql.exec('ALTER TABLE sessions ADD COLUMN next_jti TEXT');
  } catch {
    // Column already present.
  }

  // Durable Object SQL rejects `undefined` bindings; map them to null.
  /** @param {SqlBind[]} args @returns {(string | number | boolean | null)[]} */
  const b = (args) => args.map((a) => (a === undefined ? null : a));
  /** @param {string} q @param {...SqlBind} args @returns {unknown} */
  const get = (q, ...args) => sql.exec(q, ...b(args)).toArray()[0];
  /** @param {string} q @param {...SqlBind} args @returns {unknown[]} */
  const all = (q, ...args) => sql.exec(q, ...b(args)).toArray();
  /** @param {string} q @param {...SqlBind} args @returns {number} rows written */
  const run = (q, ...args) => sql.exec(q, ...b(args)).rowsWritten;

  return {
    async getActor(did) {
      const row = /** @type {ActorRow | undefined} */ (
        get('SELECT did, handle, created_at FROM actors WHERE did = ?', did)
      );
      return row
        ? { did: row.did, handle: row.handle, createdAt: row.created_at }
        : null;
    },
    async resolveHandle(handle) {
      const row = /** @type {{did: string} | undefined} */ (
        get('SELECT did FROM actors WHERE lower(handle) = lower(?)', handle)
      );
      return row ? row.did : null;
    },
    async putActor(did, handle, createdAt) {
      run(
        'INSERT OR REPLACE INTO actors (did, handle, created_at) VALUES (?, ?, ?)',
        did,
        handle,
        createdAt,
      );
    },
    async deleteActor(did) {
      run('DELETE FROM actors WHERE did = ?', did);
    },

    async getOAuthRequest(requestUri) {
      const row = /** @type {OAuthRequestRow | undefined} */ (
        get(
          'SELECT data, expires_at FROM oauth_requests WHERE request_uri = ?',
          requestUri,
        )
      );
      if (!row) return null;
      if (row.expires_at < Date.now()) {
        run('DELETE FROM oauth_requests WHERE request_uri = ?', requestUri);
        return null;
      }
      try {
        return JSON.parse(row.data);
      } catch {
        run('DELETE FROM oauth_requests WHERE request_uri = ?', requestUri);
        return null;
      }
    },
    async putOAuthRequest(requestUri, data, expiresAt) {
      run(
        'INSERT OR REPLACE INTO oauth_requests (request_uri, data, expires_at) VALUES (?, ?, ?)',
        requestUri,
        JSON.stringify(data),
        expiresAt,
      );
    },
    async deleteOAuthRequest(requestUri) {
      run('DELETE FROM oauth_requests WHERE request_uri = ?', requestUri);
    },

    async getOAuthToken(tokenId) {
      const row = /** @type {OAuthTokenRow | undefined} */ (
        get('SELECT data FROM oauth_tokens WHERE token_id = ?', tokenId)
      );
      if (!row) return null;
      try {
        return JSON.parse(row.data);
      } catch {
        run('DELETE FROM oauth_tokens WHERE token_id = ?', tokenId);
        return null;
      }
    },
    async putOAuthToken(tokenId, data) {
      const tokenData = /** @type {{did: string}} */ (data);
      run(
        'INSERT OR REPLACE INTO oauth_tokens (token_id, did, data) VALUES (?, ?, ?)',
        tokenId,
        tokenData.did,
        JSON.stringify(data),
      );
    },
    async deleteOAuthToken(tokenId) {
      run('DELETE FROM oauth_tokens WHERE token_id = ?', tokenId);
    },
    async listOAuthTokensByDid(did) {
      const rows = /** @type {OAuthTokenListRow[]} */ (
        all('SELECT token_id, data FROM oauth_tokens WHERE did = ?', did)
      );
      return rows
        .map((r) => {
          try {
            return { ...JSON.parse(r.data), tokenId: r.token_id };
          } catch {
            return null;
          }
        })
        .filter(Boolean);
    },

    async checkAndStoreDpopJti(jti, expiresAt) {
      return (
        run(
          'INSERT OR IGNORE INTO dpop_jtis (jti, expires_at) VALUES (?, ?)',
          jti,
          expiresAt,
        ) > 0
      );
    },
    async cleanupExpiredDpopJtis() {
      run('DELETE FROM dpop_jtis WHERE expires_at < ?', Date.now());
    },

    async putAppPassword(did, name, hash, privileged, createdAt) {
      // INSERT OR IGNORE, not REPLACE: reusing a name must not silently
      // invalidate the password already issued under it.
      return (
        run(
          'INSERT OR IGNORE INTO app_passwords (did, name, hash, privileged, created_at) VALUES (?, ?, ?, ?, ?)',
          did,
          name,
          hash,
          privileged ? 1 : 0,
          createdAt,
        ) > 0
      );
    },
    async getAppPasswordByHash(did, hash) {
      const row = /** @type {AppPasswordRow | undefined} */ (
        get(
          'SELECT name, privileged, created_at FROM app_passwords WHERE did = ? AND hash = ?',
          did,
          hash,
        )
      );
      return row ? toAppPassword(row) : null;
    },
    async listAppPasswords(did) {
      const rows = /** @type {AppPasswordRow[]} */ (
        all(
          'SELECT name, privileged, created_at FROM app_passwords WHERE did = ? ORDER BY created_at',
          did,
        )
      );
      return rows.map(toAppPassword);
    },
    async deleteAppPassword(did, name) {
      return (
        run('DELETE FROM app_passwords WHERE did = ? AND name = ?', did, name) >
        0
      );
    },

    async putSession(session) {
      run(
        'INSERT OR REPLACE INTO sessions (jti, did, label, scope, user_agent, created_at, refreshed_at, expires_at, next_jti) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
        session.jti,
        session.did,
        session.label,
        session.scope,
        session.userAgent,
        session.createdAt,
        session.refreshedAt,
        session.expiresAt ?? null,
        session.nextJti ?? null,
      );
    },
    async getSession(jti) {
      const row = /** @type {SessionRow | undefined} */ (
        get('SELECT * FROM sessions WHERE jti = ?', jti)
      );
      return row ? toSessionRecord(row) : null;
    },
    async listSessions(did) {
      const rows = /** @type {SessionRow[]} */ (
        all('SELECT * FROM sessions WHERE did = ? ORDER BY created_at', did)
      );
      return rows.map(toSessionRecord);
    },
    async deleteSession(jti) {
      return run('DELETE FROM sessions WHERE jti = ?', jti) > 0;
    },
    async spendSession(jti, expiresAt, nextJti) {
      // Conditional on the successor, so that of two rotations racing the same
      // token only the first takes effect. `sql.exec` runs to completion, so
      // nothing interleaves between the test and the write.
      return (
        run(
          'UPDATE sessions SET expires_at = ?, next_jti = ? WHERE jti = ? AND (next_jti IS NULL OR next_jti = ?)',
          expiresAt,
          nextJti,
          jti,
          nextJti,
        ) > 0
      );
    },

    async putPasskey(record) {
      return (
        run(
          'INSERT OR IGNORE INTO passkeys (credential_id, did, name, public_key, algorithm, sign_count, created_at, last_used_at) VALUES (?, ?, ?, ?, ?, 0, ?, NULL)',
          record.credentialId,
          record.did,
          record.name,
          record.publicKey,
          record.algorithm,
          record.createdAt,
        ) > 0
      );
    },
    async getPasskey(credentialId) {
      const row = /** @type {PasskeyRow | undefined} */ (
        get('SELECT * FROM passkeys WHERE credential_id = ?', credentialId)
      );
      return row ? toPasskeyRecord(row) : null;
    },
    async listPasskeys(did) {
      const rows = /** @type {PasskeyRow[]} */ (
        all('SELECT * FROM passkeys WHERE did = ? ORDER BY created_at', did)
      );
      return rows.map(toPasskeyRecord);
    },
    async deletePasskey(did, credentialId) {
      return (
        run(
          'DELETE FROM passkeys WHERE did = ? AND credential_id = ?',
          did,
          credentialId,
        ) > 0
      );
    },
    async touchPasskey(credentialId, signCount, lastUsedAt) {
      run(
        'UPDATE passkeys SET sign_count = ?, last_used_at = ? WHERE credential_id = ?',
        signCount,
        lastUsedAt,
        credentialId,
      );
    },
  };
}

/**
 * Whether this request is for a site hostname: under the configured site
 * domain and not the PDS's own hostname. The PDS hostname is excluded even
 * when it sits under the site domain, so PDS traffic never takes the site
 * path. A `wrangler dev` localhost never matches.
 * @param {Request} request
 * @param {Env} env
 * @returns {boolean}
 */
export function siteHost(request, env) {
  if (!env.PDS_EXPERIMENTAL_SITE_DOMAIN) return false;
  const host = new URL(request.url).hostname.toLowerCase();
  const domain = env.PDS_EXPERIMENTAL_SITE_DOMAIN.toLowerCase();
  if (host !== domain && !host.endsWith(`.${domain}`)) return false;
  return host !== (env.HOSTNAME || '').split(':')[0].toLowerCase();
}

/**
 * Whether this request may be served from and stored in the shared edge
 * cache: a GET for a site hostname, never the PDS's own. The Durable Object
 * lives in one region while site visitors do not, so this is what keeps
 * repeat site traffic off the cross-region round trip. PDS responses carry
 * sessions and DPoP nonces and must never enter a shared cache.
 * @param {Request} request
 * @param {Env} env
 * @returns {boolean}
 */
export function siteCacheable(request, env) {
  if (request.method !== 'GET' || !siteHost(request, env)) return false;
  // The registries answer on every hostname, and their reads are mutable:
  // a packument gains versions and a manifest tag moves. A cached copy
  // would serve a package that was deleted or miss one just published.
  const pathname = new URL(request.url).pathname;
  return !pathname.startsWith('/v2/') && !pathname.startsWith('/npm/');
}

/**
 * Whether a response may enter the shared edge cache: only complete
 * successes, and only ones the origin marked shareable. The site handler's
 * 404s say no-store; blob errors and partials fail the status check.
 * @param {Response} response
 * @returns {boolean}
 */
export function responseStorable(response) {
  if (response.status !== 200 || response.headers.has('Set-Cookie')) {
    return false;
  }
  return !/no-store|private/i.test(response.headers.get('Cache-Control') || '');
}

/**
 * Whether a request on the PDS's own hostname may be served from the shared
 * edge cache: a public blob GET, nothing else. Blob bytes are
 * content-addressed and immutable, and the Durable Object lives in one
 * region while the video CDN and gallery visitors do not.
 * @param {Request} request
 * @returns {boolean}
 */
export function blobCacheable(request) {
  return (
    request.method === 'GET' &&
    new URL(request.url).pathname === '/xrpc/com.atproto.sync.getBlob'
  );
}

/**
 * getBlob through the shared edge cache. Only complete 200s are stored, so
 * the space-privacy filter in the Durable Object decides what ever enters
 * the cache. The cache answers ranged requests out of a stored 200; a
 * ranged request for a cold blob passes through to the Durable Object, and
 * a background fetch stores the whole object so later ranges stay at the
 * edge. A blob unpublished later can persist in edge caches until eviction,
 * like any CDN copy of once-public immutable content.
 * @param {Request} request
 * @param {{fetch: (request: Request) => Promise<Response>}} stub
 * @param {ExecutionContext} ctx
 * @returns {Promise<Response>}
 */
async function serveBlobThroughCache(request, stub, ctx) {
  const cache = /** @type {CacheStorage & {default: Cache}} */ (caches).default;
  const cached = await cache.match(request);
  if (cached) {
    const hit = new Response(cached.body, cached);
    hit.headers.set('X-Blob-Cache', 'hit');
    return hit;
  }
  const response = await stub.fetch(request);
  if (responseStorable(response)) {
    ctx.waitUntil(cache.put(request, response.clone()));
  } else if (response.status === 206) {
    const full = new Request(request.url);
    ctx.waitUntil(
      (async () => {
        const whole = await stub.fetch(full);
        if (responseStorable(whole)) {
          await cache.put(full, whole);
        } else {
          await whole.body?.cancel();
        }
      })(),
    );
  }
  return response;
}

/**
 * Worker entry point - single-tenant mode
 * All requests route to default DO instance
 * TODO: Add multi-tenant routing for hosting providers
 */
export default {
  /**
   * @param {Request} request
   * @param {Env} env
   * @param {ExecutionContext} ctx
   * @returns {Promise<Response>}
   */
  async fetch(request, env, ctx) {
    // The setup wizard's Cloudflare API calls, when the proxy is enabled.
    // Before site dispatch, so a site-hosted wizard reaches it same-origin.
    if (
      cfApiProxyEnabled(env) &&
      new URL(request.url).pathname.startsWith(CF_API_PROXY_PREFIX)
    ) {
      return handleCfApiProxy(request);
    }

    const stub = env.PDS.get(env.PDS.idFromName('default'));
    if (!siteHost(request, env)) {
      if (blobCacheable(request)) {
        return serveBlobThroughCache(request, stub, ctx);
      }
      return stub.fetch(request);
    }

    // Custom domains answer on both schemes, and a plain-HTTP page marks
    // the site "not secure" in browsers. Sites are HTTPS only. The URL's own
    // scheme reads https even for plain-HTTP visitors; the edge reports the
    // real one in X-Forwarded-Proto.
    const url = new URL(request.url);
    const proto = request.headers.get('X-Forwarded-Proto');
    if (proto ? proto === 'http' : url.protocol === 'http:') {
      url.protocol = 'https:';
      return Response.redirect(url.href, 301);
    }

    if (!siteCacheable(request, env)) return stub.fetch(request);

    const cache = /** @type {CacheStorage & {default: Cache}} */ (caches)
      .default;
    const cached = await cache.match(request);
    if (cached) {
      // cache.match ignores conditional headers, so answer If-None-Match
      // here; the ETag is the file's CID and survives redeploys unchanged.
      const etag = cached.headers.get('ETag');
      const conditional = request.headers.get('If-None-Match');
      if (etag && conditional?.split(',').some((t) => t.trim() === etag)) {
        const headers = new Headers(cached.headers);
        headers.delete('Content-Length');
        return new Response(null, { status: 304, headers });
      }
      const hit = new Response(cached.body, cached);
      hit.headers.set('X-Sites-Cache', 'hit');
      return hit;
    }

    const response = await stub.fetch(request);
    if (responseStorable(response)) {
      ctx.waitUntil(cache.put(request, response.clone()));
    }
    return response;
  },
};