/**
 * OCI distribution HTTP API served from dev.pdsjs.oci.repo records, so
 * stock docker can push and pull:
 *
 *   docker pull <pds-host>/<name>:<tag>
 *   docker push <pds-host>/<name>:<tag>
 *
 * A layer is one blob: its OCI digest and its blob CID hold the same
 * SHA-256 bytes, so a blob GET is a digest re-encoding plus a getBlob.
 * Manifest bytes are stored as blobs untouched, because a digest covers the
 * exact serialization. The record keeps a blob reference for every stored
 * manifest, config and layer, which is what holds them against orphan
 * cleanup.
 *
 * Reads go through an injected fetch, typically the server's own fetch
 * function. Writes go straight to the ports, the way the drive writer
 * writes. Without the write ports the registry serves pulls only.
 *
 * This module must stay free of node-only imports; it runs on Cloudflare
 * Workers.
 */

import { cborEncodeDagCbor, cidToString, createCid } from '@pdsjs/core/repo';
import { createSha256 } from '@pdsjs/core/sha256';
import { digestFromHashBytes, digestToCid, isValidDigest } from './digest.js';
import {
  isValidName,
  isValidTag,
  nameToRkey,
  OCI_REPO_COLLECTION,
} from './lexicon.js';
import { MAX_MANIFEST_SIZE, parseManifest } from './manifest.js';
import {
  blobRef,
  parseRepoRecord,
  recordBlobCids,
  removeManifest,
  serializeRepoRecord,
} from './record.js';

const API_VERSION_HEADER = 'Docker-Distribution-API-Version';
const API_VERSION = 'registry/2.0';
const BLOB_MIME_TYPE = 'application/octet-stream';

/** How long an upload session waits between requests before it is swept. */
const UPLOAD_SESSION_TTL = 60 * 60 * 1000;

/** How long one verified credential pair stays good without re-checking. */
const AUTH_CACHE_TTL = 5 * 60 * 1000;

/** Largest accepted layer, when the caller sets no other bound. */
const DEFAULT_MAX_BLOB_SIZE = 10 * 1024 * 1024 * 1024;

/**
 * @typedef {Object} RegistryContext
 * @property {(request: Request) => Promise<Response>} xrpc - the PDS fetch
 *   function; requests are self-addressed XRPC calls
 * @property {() => Promise<string|null>} getDid
 * @property {import('@pdsjs/core/ports').ActorStoragePort} [actorStorage] - pushes need it
 * @property {import('@pdsjs/core/ports').BlobPort} [blobs] - pushes need it
 * @property {(did: string, ops: Array<{action: string, uri: string, cid?: string}>) => Promise<Object>} [createCommit] - pushes need it
 * @property {import('@pdsjs/core/ports').HasherFactory} [hasher] - SHA-256 factory; absent, the plain JavaScript one
 * @property {number} [maxBlobSize] - largest accepted layer, in bytes
 * @property {string} [collection] - record collection, for lexicon forks
 * @property {string} [realm] - realm named in the Basic challenge
 */

/**
 * @typedef {Object} UploadSession
 * @property {string} name
 * @property {import('@pdsjs/core/ports').BlobWriter} writer
 * @property {import('@pdsjs/core/ports').Hasher} hash
 * @property {number} size
 * @property {number} touched
 * @property {boolean} busy
 */

/**
 * @param {string} code - error code from the distribution spec
 * @param {string} message
 * @param {number} status
 * @param {Record<string, string>} [headers]
 * @returns {Response}
 */
function ociError(code, message, status, headers = {}) {
  return Response.json(
    { errors: [{ code, message }] },
    { status, headers: { [API_VERSION_HEADER]: API_VERSION, ...headers } },
  );
}

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

/**
 * The `start-end` byte range of an upload PATCH. The `bytes ` prefix some
 * clients send is accepted.
 * @param {string|null} header
 * @returns {{start: number, end: number}|null}
 */
function parseContentRange(header) {
  if (!header) return null;
  const match = /^(?:bytes )?(\d+)-(\d+)/.exec(header);
  if (!match) return null;
  return { start: Number(match[1]), end: Number(match[2]) };
}

/**
 * @param {RegistryContext} ctx
 * @returns {{handle: (request: Request) => Promise<Response|null>}}
 */
export function createRegistry(ctx) {
  if (!ctx || typeof ctx.xrpc !== 'function') {
    throw new Error('createRegistry requires an xrpc fetch function');
  }
  if (typeof ctx.getDid !== 'function') {
    throw new Error('createRegistry requires getDid');
  }
  const {
    xrpc,
    getDid,
    actorStorage,
    blobs,
    createCommit,
    hasher = createSha256,
    maxBlobSize = DEFAULT_MAX_BLOB_SIZE,
    collection = OCI_REPO_COLLECTION,
    realm = 'registry',
  } = ctx;
  const pushEnabled = Boolean(actorStorage && blobs && createCommit);

  /**
   * @param {string} path
   * @param {RequestInit} [init]
   * @returns {Promise<Response>}
   */
  const xrpcFetch = (path, init) =>
    xrpc(new Request(`https://pds.internal${path}`, init));

  // ── Authentication ─────────────────────────────────────────────────────
  //
  // Docker only pulls anonymously through the Bearer token flow: a Basic
  // challenge with no stored credentials fails outright, where a Bearer
  // challenge sends the client to the token endpoint, which grants pull to
  // anyone and push to valid Basic credentials. Requests carrying Basic
  // credentials directly are accepted too, so curl works without a token.

  /** Verified credential digests, value = epoch ms the entry expires. */
  const authCache = new Map();

  /** Issued Bearer tokens, value = {push, expires}. */
  const tokens = new Map();

  /** How long one issued token stays good. */
  const TOKEN_TTL = 30 * 60 * 1000;

  /**
   * @param {string} decoded - `identifier:password`
   * @returns {Promise<string>}
   */
  async function credentialKey(decoded) {
    const hash = hasher();
    await hash.update(new TextEncoder().encode(decoded));
    return digestFromHashBytes(await hash.digest());
  }

  /**
   * True when the request carries Basic credentials createSession accepts.
   * The check goes through createSession itself, so account passwords and
   * app passwords both work, exactly as docker login sends them.
   * @param {string} header - the Authorization header
   * @returns {Promise<boolean>}
   */
  async function basicAuthorized(header) {
    let decoded;
    try {
      decoded = atob(header.slice(5).trim());
    } catch {
      return false;
    }
    // The username may be a DID, which holds colons of its own, so the
    // split runs after the DID prefix when one is present.
    const didMatch = /^(did:[a-z0-9]+:[^:]+):(.*)$/s.exec(decoded);
    const at = decoded.indexOf(':');
    if (!didMatch && at < 0) return false;
    const identifier = didMatch ? didMatch[1] : decoded.slice(0, at);
    const password = didMatch ? didMatch[2] : decoded.slice(at + 1);

    const key = await credentialKey(decoded);
    const expires = authCache.get(key);
    if (expires && expires > Date.now()) return true;
    authCache.delete(key);

    const res = await xrpcFetch('/xrpc/com.atproto.server.createSession', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ identifier, password }),
    });
    if (!res.ok) return false;
    authCache.set(key, Date.now() + AUTH_CACHE_TTL);
    return true;
  }

  /**
   * What the request may do: null for no valid credentials at all.
   * @param {Request} request
   * @returns {Promise<{push: boolean}|null>}
   */
  async function resolveAccess(request) {
    const header = request.headers.get('authorization') ?? '';
    if (/^basic\s/i.test(header)) {
      return (await basicAuthorized(header)) ? { push: true } : null;
    }
    if (/^bearer\s/i.test(header)) {
      const grant = tokens.get(header.slice(6).trim());
      if (!grant || grant.expires < Date.now()) return null;
      return { push: grant.push };
    }
    return null;
  }

  /**
   * GET /v2/token, the realm the Bearer challenge names. Basic credentials
   * grant push and pull; a request without credentials gets pull only,
   * which is what makes anonymous docker pulls work. Bad credentials are
   * refused rather than downgraded, so docker login fails loudly.
   * @param {Request} request
   * @returns {Promise<Response>}
   */
  async function serveToken(request) {
    const header = request.headers.get('authorization') ?? '';
    let push = false;
    if (/^basic\s/i.test(header)) {
      if (!(await basicAuthorized(header))) {
        return ociError('UNAUTHORIZED', 'invalid credentials', 401);
      }
      push = true;
    }
    for (const [token, grant] of tokens) {
      if (grant.expires < Date.now()) tokens.delete(token);
    }
    const token = crypto.randomUUID();
    tokens.set(token, { push, expires: Date.now() + TOKEN_TTL });
    return Response.json(
      {
        token,
        access_token: token,
        expires_in: Math.floor(TOKEN_TTL / 1000),
      },
      { headers: { [API_VERSION_HEADER]: API_VERSION } },
    );
  }

  /**
   * @param {URL} url
   * @returns {Response}
   */
  function challenge(url) {
    return ociError('UNAUTHORIZED', 'authentication required', 401, {
      'WWW-Authenticate': `Bearer realm="${url.origin}/v2/token",service="${realm}"`,
    });
  }

  // ── Reads ──────────────────────────────────────────────────────────────

  /**
   * @param {string} did
   * @param {string} name
   * @returns {Promise<import('./record.js').OciRepoRecord|null>}
   */
  async function loadRepo(did, name) {
    const params = new URLSearchParams({
      repo: did,
      collection,
      rkey: nameToRkey(name),
    });
    const res = await xrpcFetch(`/xrpc/com.atproto.repo.getRecord?${params}`);
    if (!res.ok) return null;
    const body = /** @type {{value: unknown}} */ (await res.json());
    return parseRepoRecord(body.value);
  }

  /**
   * Size and type of one stored blob, or null when it is absent. The
   * metadata row answers when the port is present; otherwise a one-byte
   * ranged getBlob reads the size out of Content-Range.
   * @param {string} did
   * @param {string} digest
   * @returns {Promise<{size: number, mimeType: string}|null>}
   */
  async function blobMeta(did, digest) {
    const cid = digestToCid(digest);
    if (actorStorage?.getBlob) {
      const meta = await actorStorage.getBlob(cid);
      return meta ? { size: meta.size, mimeType: meta.mimeType } : null;
    }
    const params = new URLSearchParams({ did, cid });
    const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`, {
      headers: { Range: 'bytes=0-0' },
    });
    if (res.status === 206) {
      await res.body?.cancel();
      const match = /\/(\d+)$/.exec(res.headers.get('content-range') ?? '');
      return match
        ? {
            size: Number(match[1]),
            mimeType: res.headers.get('content-type') ?? BLOB_MIME_TYPE,
          }
        : null;
    }
    if (res.status === 416) {
      // An empty blob satisfies no byte range; the 416 still proves it
      // exists.
      return { size: 0, mimeType: BLOB_MIME_TYPE };
    }
    if (res.ok) {
      const data = await res.arrayBuffer();
      return {
        size: data.byteLength,
        mimeType: res.headers.get('content-type') ?? BLOB_MIME_TYPE,
      };
    }
    return null;
  }

  /**
   * @param {string} did
   * @param {string} cid
   * @returns {Promise<Uint8Array|null>}
   */
  async function readBlob(did, cid) {
    const params = new URLSearchParams({ did, cid });
    const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`);
    if (!res.ok) return null;
    return new Uint8Array(await res.arrayBuffer());
  }

  /**
   * GET or HEAD one blob, streaming and honouring Range.
   * @param {Request} request
   * @param {string} did
   * @param {string} digest
   * @returns {Promise<Response>}
   */
  async function serveBlob(request, did, digest) {
    if (!isValidDigest(digest)) {
      return ociError('DIGEST_INVALID', `unsupported digest: ${digest}`, 400);
    }
    const meta = await blobMeta(did, digest);
    if (!meta) {
      return ociError('BLOB_UNKNOWN', `blob unknown: ${digest}`, 404);
    }
    /** @type {Record<string, string>} */
    const headers = {
      [API_VERSION_HEADER]: API_VERSION,
      'Docker-Content-Digest': digest,
      'Content-Type': BLOB_MIME_TYPE,
      'Accept-Ranges': 'bytes',
      ETag: `"${digest}"`,
    };
    if (request.method === 'HEAD') {
      headers['Content-Length'] = String(meta.size);
      return new Response(null, { status: 200, headers });
    }
    const params = new URLSearchParams({ did, cid: digestToCid(digest) });
    /** @type {HeadersInit} */
    const forward = {};
    const range = request.headers.get('range');
    if (range) forward.Range = range;
    const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`, {
      headers: forward,
    });
    if (!res.ok && res.status !== 206 && res.status !== 416) {
      return ociError('BLOB_UNKNOWN', `blob unknown: ${digest}`, 404);
    }
    for (const name of ['Content-Length', 'Content-Range']) {
      const value = res.headers.get(name);
      if (value) headers[name] = value;
    }
    return new Response(res.body, { status: res.status, headers });
  }

  /**
   * The manifest entry a reference names: a digest directly, a tag through
   * the tag list.
   * @param {import('./record.js').OciRepoRecord} record
   * @param {string} reference
   * @returns {import('./record.js').ManifestEntry|null}
   */
  function resolveManifest(record, reference) {
    const digest = reference.includes(':')
      ? reference
      : record.tags.find((t) => t.tag === reference)?.digest;
    if (!digest) return null;
    return record.manifests.find((m) => m.digest === digest) ?? null;
  }

  /**
   * @param {Request} request
   * @param {string} did
   * @param {string} name
   * @param {string} reference - tag or digest
   * @returns {Promise<Response>}
   */
  async function serveManifest(request, did, name, reference) {
    const record = await loadRepo(did, name);
    if (!record) {
      return ociError('NAME_UNKNOWN', `repository unknown: ${name}`, 404);
    }
    const entry = resolveManifest(record, reference);
    if (!entry) {
      return ociError(
        'MANIFEST_UNKNOWN',
        `manifest unknown: ${reference}`,
        404,
      );
    }
    /** @type {Record<string, string>} */
    const headers = {
      [API_VERSION_HEADER]: API_VERSION,
      'Docker-Content-Digest': entry.digest,
      'Content-Type': entry.mediaType,
      ETag: `"${entry.digest}"`,
    };
    const ifNoneMatch = request.headers.get('if-none-match');
    if (ifNoneMatch?.includes(entry.digest)) {
      return new Response(null, { status: 304, headers });
    }
    const bytes = await readBlob(did, entry.blob.ref.$link);
    if (!bytes) {
      return ociError(
        'MANIFEST_UNKNOWN',
        `manifest unknown: ${reference}`,
        404,
      );
    }
    headers['Content-Length'] = String(bytes.length);
    return new Response(
      request.method === 'HEAD' ? null : /** @type {BodyInit} */ (bytes),
      { status: 200, headers },
    );
  }

  /**
   * @param {URL} url
   * @param {string} did
   * @param {string} name
   * @returns {Promise<Response>}
   */
  async function serveTagsList(url, did, name) {
    const record = await loadRepo(did, name);
    if (!record) {
      return ociError('NAME_UNKNOWN', `repository unknown: ${name}`, 404);
    }
    let tags = record.tags.map((t) => t.tag).sort();
    const last = url.searchParams.get('last');
    if (last) tags = tags.filter((t) => t > last);
    const rawN = url.searchParams.get('n');
    const n = rawN === null ? Number.NaN : Number(rawN);
    if (Number.isFinite(n) && n >= 0) tags = tags.slice(0, n);
    return Response.json(
      { name, tags },
      { headers: { [API_VERSION_HEADER]: API_VERSION } },
    );
  }

  /**
   * @param {string} did
   * @returns {Promise<Response>}
   */
  async function serveCatalog(did) {
    const params = new URLSearchParams({
      repo: did,
      collection,
      limit: '100',
    });
    const res = await xrpcFetch(`/xrpc/com.atproto.repo.listRecords?${params}`);
    /** @type {string[]} */
    let repositories = [];
    if (res.ok) {
      const body = /** @type {{records: Array<{value: unknown}>}} */ (
        await res.json()
      );
      repositories = body.records
        .map((r) => parseRepoRecord(r.value)?.name)
        .filter((name) => typeof name === 'string')
        .sort();
    }
    return Response.json(
      { repositories },
      { headers: { [API_VERSION_HEADER]: API_VERSION } },
    );
  }

  // ── Writes ─────────────────────────────────────────────────────────────

  /**
   * Multi-request uploads in flight, keyed by session id. In memory: a
   * restart loses them, and the client starts the upload over.
   * @type {Map<string, UploadSession>}
   */
  const uploads = new Map();

  /**
   * One write to a repository record at a time, so concurrent pushes
   * read-modify-write in sequence rather than losing updates.
   * @type {Map<string, Promise<unknown>>}
   */
  const recordLocks = new Map();

  /**
   * @template T
   * @param {string} name
   * @param {() => Promise<T>} fn
   * @returns {Promise<T>}
   */
  function withRecordLock(name, fn) {
    const previous = recordLocks.get(name) ?? Promise.resolve();
    const next = previous.then(fn, fn);
    recordLocks.set(
      name,
      next.catch(() => {}),
    );
    return next;
  }

  /**
   * @param {string} id
   * @returns {UploadSession}
   */
  function takeUpload(id) {
    const session = uploads.get(id);
    if (!session || session.busy) {
      throw new Error('upload unknown');
    }
    return session;
  }

  /** @param {string} id */
  async function dropUpload(id) {
    const session = uploads.get(id);
    if (!session) return;
    uploads.delete(id);
    try {
      await session.writer.abort();
    } catch {
      // an abort that fails leaves staged bytes for the platform to expire
    }
  }

  /** Abandon every session that has waited past the TTL. */
  async function sweepUploads() {
    const cutoff = Date.now() - UPLOAD_SESSION_TTL;
    for (const [id, session] of uploads) {
      if (session.touched < cutoff && !session.busy) await dropUpload(id);
    }
  }

  /**
   * Feed one request body into a session. A body that fails partway has
   * already fed the hash, so the session ends and the client starts over.
   * @param {UploadSession} session
   * @param {string} id
   * @param {ReadableStream<Uint8Array>|null} body
   * @returns {Promise<Response|null>} an error response, or null on success
   */
  async function appendBody(session, id, body) {
    if (!body) return null;
    session.busy = true;
    try {
      const reader = body.getReader();
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        const chunk =
          value instanceof Uint8Array ? value : new Uint8Array(value);
        session.size += chunk.length;
        if (session.size > maxBlobSize) {
          await reader.cancel();
          await dropUpload(id);
          return ociError(
            'SIZE_INVALID',
            `blob exceeds the maximum size of ${maxBlobSize} bytes`,
            413,
          );
        }
        await session.hash.update(chunk);
        await session.writer.write(chunk);
      }
      session.busy = false;
      session.touched = Date.now();
      return null;
    } catch (err) {
      await dropUpload(id);
      const message = err instanceof Error ? err.message : String(err);
      return ociError('BLOB_UPLOAD_INVALID', message, 400);
    }
  }

  /**
   * Land a finished session as the blob its digest names.
   * @param {UploadSession} session
   * @param {string} id
   * @param {string} digest - declared digest the bytes must hash to
   * @returns {Promise<Response>}
   */
  async function finishUpload(session, id, digest) {
    uploads.delete(id);
    if (!isValidDigest(digest)) {
      await session.writer.abort();
      return ociError('DIGEST_INVALID', `unsupported digest: ${digest}`, 400);
    }
    const computed = digestFromHashBytes(await session.hash.digest());
    if (computed !== digest) {
      await session.writer.abort();
      return ociError(
        'DIGEST_INVALID',
        `digest mismatch: body hashes to ${computed}`,
        400,
      );
    }
    const cid = digestToCid(digest);
    await /** @type {NonNullable<typeof actorStorage>} */ (
      actorStorage
    ).putBlob(cid, BLOB_MIME_TYPE, session.size);
    await session.writer.commit(cid, BLOB_MIME_TYPE);
    return new Response(null, {
      status: 201,
      headers: {
        [API_VERSION_HEADER]: API_VERSION,
        Location: `/v2/${session.name}/blobs/${digest}`,
        'Docker-Content-Digest': digest,
      },
    });
  }

  /**
   * @param {string} name
   * @returns {Promise<{id: string, session: UploadSession}>}
   */
  async function createUpload(name) {
    await sweepUploads();
    const id = crypto.randomUUID();
    const session = {
      name,
      writer:
        await /** @type {NonNullable<import('@pdsjs/core/ports').BlobPort['createWriter']>} */ (
          /** @type {NonNullable<typeof blobs>} */ (blobs).createWriter
        )(/** @type {string} */ (await getDid())),
      hash: hasher(),
      size: 0,
      touched: Date.now(),
      busy: false,
    };
    uploads.set(id, session);
    return { id, session };
  }

  /**
   * @param {string} name
   * @param {string} id
   * @param {UploadSession} session
   * @returns {Response}
   */
  function uploadAccepted(name, id, session) {
    return new Response(null, {
      status: 202,
      headers: {
        [API_VERSION_HEADER]: API_VERSION,
        Location: `/v2/${name}/blobs/uploads/${id}`,
        Range: `0-${Math.max(0, session.size - 1)}`,
        'Docker-Upload-UUID': id,
      },
    });
  }

  /**
   * Store the record, link its blobs and commit, the way the drive writer
   * does. The link runs before the commit, so a reap between the two cannot
   * take a blob the record already holds.
   * @param {string} did
   * @param {string} name
   * @param {import('./record.js').OciRepoRecord} record
   * @param {boolean} existed
   */
  async function writeRepoRecord(did, name, record, existed) {
    const storage = /** @type {NonNullable<typeof actorStorage>} */ (
      actorStorage
    );
    const rkey = nameToRkey(name);
    const uri = `at://${did}/${collection}/${rkey}`;
    const encoded = cborEncodeDagCbor(serializeRepoRecord(record));
    const cid = cidToString(await createCid(encoded));
    await storage.putBlock(cid, encoded);
    await storage.putRecord(uri, cid, collection, rkey, encoded);
    await storage.unlinkBlobsFromRecord(uri);
    for (const blobCid of recordBlobCids(record)) {
      await storage.linkBlobToRecord(blobCid, uri, Date.now());
    }
    await /** @type {NonNullable<typeof createCommit>} */ (createCommit)(did, [
      { action: existed ? 'update' : 'create', uri, cid },
    ]);
  }

  /**
   * @param {Request} request
   * @param {string} did
   * @param {string} name
   * @param {string} reference - tag or digest
   * @returns {Promise<Response>}
   */
  async function putManifest(request, did, name, reference) {
    const declared = request.headers.get('content-type') ?? BLOB_MIME_TYPE;
    const chunks = [];
    let size = 0;
    if (request.body) {
      const reader = request.body.getReader();
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        const chunk =
          value instanceof Uint8Array ? value : new Uint8Array(value);
        size += chunk.length;
        if (size > MAX_MANIFEST_SIZE) {
          await reader.cancel();
          return ociError('MANIFEST_INVALID', 'manifest too large', 400);
        }
        chunks.push(chunk);
      }
    }
    const bytes = joinChunks(chunks, size);

    /** @type {import('./manifest.js').ParsedManifest} */
    let parsed;
    try {
      parsed = parseManifest(bytes, declared);
    } catch (err) {
      const message = err instanceof Error ? err.message : String(err);
      return ociError('MANIFEST_INVALID', message, 400);
    }

    const hash = hasher();
    await hash.update(bytes);
    const digest = digestFromHashBytes(await hash.digest());
    if (reference.includes(':')) {
      if (reference !== digest) {
        return ociError(
          'DIGEST_INVALID',
          `manifest hashes to ${digest}, not ${reference}`,
          400,
        );
      }
    } else if (!isValidTag(reference)) {
      return ociError('TAG_INVALID', `invalid tag: ${reference}`, 400);
    }

    return withRecordLock(name, async () => {
      const existing = await loadRepo(did, name);

      // Everything the manifest references must already be here. Layers
      // with external urls or a foreign media type are distributed outside
      // this registry, so their absence is expected.
      /** @type {import('./record.js').LayerEntry[]} */
      const layerEntries = [];
      for (const descriptor of parsed.blobs) {
        if (!isValidDigest(descriptor.digest)) {
          return ociError(
            'DIGEST_INVALID',
            `unsupported digest: ${descriptor.digest}`,
            400,
          );
        }
        if (
          (descriptor.urls?.length ?? 0) > 0 ||
          descriptor.mediaType?.includes('foreign')
        ) {
          continue;
        }
        const meta = await blobMeta(did, descriptor.digest);
        if (!meta) {
          return ociError(
            'MANIFEST_BLOB_UNKNOWN',
            `blob unknown: ${descriptor.digest}`,
            404,
          );
        }
        layerEntries.push({
          digest: descriptor.digest,
          blob: blobRef(
            digestToCid(descriptor.digest),
            descriptor.mediaType ?? meta.mimeType,
            meta.size,
          ),
        });
      }
      for (const child of parsed.children) {
        if (!existing?.manifests.some((m) => m.digest === child.digest)) {
          return ociError(
            'MANIFEST_BLOB_UNKNOWN',
            `manifest unknown: ${child.digest}`,
            404,
          );
        }
      }

      const manifestCid = digestToCid(digest);
      const storage = /** @type {NonNullable<typeof actorStorage>} */ (
        actorStorage
      );
      await /** @type {NonNullable<typeof blobs>} */ (blobs).put(
        /** @type {string} */ (did),
        manifestCid,
        bytes,
        parsed.mediaType,
      );
      await storage.putBlob(manifestCid, parsed.mediaType, bytes.length);

      const now = new Date().toISOString();
      /** @type {import('./record.js').OciRepoRecord} */
      const record = existing ?? {
        name,
        tags: [],
        manifests: [],
        layers: [],
        createdAt: now,
      };
      record.updatedAt = now;
      if (!record.manifests.some((m) => m.digest === digest)) {
        record.manifests.push({
          digest,
          mediaType: parsed.mediaType,
          blob: blobRef(manifestCid, parsed.mediaType, bytes.length),
        });
      }
      for (const entry of layerEntries) {
        if (!record.layers.some((l) => l.digest === entry.digest)) {
          record.layers.push(entry);
        }
      }
      if (!reference.includes(':')) {
        const tag = record.tags.find((t) => t.tag === reference);
        if (tag) tag.digest = digest;
        else record.tags.push({ tag: reference, digest });
      }
      await writeRepoRecord(did, name, record, Boolean(existing));

      return new Response(null, {
        status: 201,
        headers: {
          [API_VERSION_HEADER]: API_VERSION,
          Location: `/v2/${name}/manifests/${digest}`,
          'Docker-Content-Digest': digest,
        },
      });
    });
  }

  /**
   * Delete a tag, or a manifest with every tag naming it. Removing a
   * manifest drops the layer references no remaining manifest holds, and
   * the orphan sweep frees the bytes.
   * @param {string} did
   * @param {string} name
   * @param {string} reference - tag or digest
   * @returns {Promise<Response>}
   */
  async function deleteManifest(did, name, reference) {
    return withRecordLock(name, async () => {
      const record = await loadRepo(did, name);
      if (!record) {
        return ociError('NAME_UNKNOWN', `repository unknown: ${name}`, 404);
      }

      if (!reference.includes(':')) {
        if (!record.tags.some((t) => t.tag === reference)) {
          return ociError('MANIFEST_UNKNOWN', `tag unknown: ${reference}`, 404);
        }
        record.tags = record.tags.filter((t) => t.tag !== reference);
        record.updatedAt = new Date().toISOString();
        await writeRepoRecord(did, name, record, true);
        return new Response(null, {
          status: 202,
          headers: { [API_VERSION_HEADER]: API_VERSION },
        });
      }

      if (!record.manifests.some((m) => m.digest === reference)) {
        return ociError(
          'MANIFEST_UNKNOWN',
          `manifest unknown: ${reference}`,
          404,
        );
      }
      // The record does not hold which layers each manifest references, so
      // the remaining manifests are re-read to compute what stays.
      const stillReferenced = new Set();
      for (const entry of record.manifests) {
        if (entry.digest === reference) continue;
        const bytes = await readBlob(did, entry.blob.ref.$link);
        if (!bytes) continue;
        const parsed = parseManifest(bytes, entry.mediaType);
        for (const descriptor of parsed.blobs) {
          stillReferenced.add(descriptor.digest);
        }
      }
      const next = removeManifest(record, reference, stillReferenced);
      next.updatedAt = new Date().toISOString();
      await writeRepoRecord(did, name, next, true);
      return new Response(null, {
        status: 202,
        headers: { [API_VERSION_HEADER]: API_VERSION },
      });
    });
  }

  // ── Dispatch ───────────────────────────────────────────────────────────

  /**
   * @param {string[]} segments - path segments after /v2/
   * @returns {{route: string, name: string, arg: string}|null}
   */
  function matchRoute(segments) {
    const s = segments;
    const n = s.length;
    if (n >= 3 && s[n - 2] === 'manifests') {
      return {
        route: 'manifests',
        name: s.slice(0, n - 2).join('/'),
        arg: s[n - 1],
      };
    }
    if (n >= 3 && s[n - 2] === 'tags' && s[n - 1] === 'list') {
      return { route: 'tags', name: s.slice(0, n - 2).join('/'), arg: '' };
    }
    if (n >= 4 && s[n - 3] === 'blobs' && s[n - 2] === 'uploads') {
      return {
        route: 'upload',
        name: s.slice(0, n - 3).join('/'),
        arg: s[n - 1],
      };
    }
    if (n >= 3 && s[n - 2] === 'blobs' && s[n - 1] === 'uploads') {
      return { route: 'uploads', name: s.slice(0, n - 2).join('/'), arg: '' };
    }
    if (n >= 3 && s[n - 2] === 'blobs') {
      return {
        route: 'blobs',
        name: s.slice(0, n - 2).join('/'),
        arg: s[n - 1],
      };
    }
    if (n >= 3 && s[n - 2] === 'referrers') {
      return {
        route: 'referrers',
        name: s.slice(0, n - 2).join('/'),
        arg: s[n - 1],
      };
    }
    return null;
  }

  return {
    /**
     * @param {Request} request
     * @returns {Promise<Response|null>} null when the path is not a
     *   registry route
     */
    async handle(request) {
      const url = new URL(request.url);
      const pathname = decodeURIComponent(url.pathname);
      if (pathname !== '/v2' && !pathname.startsWith('/v2/')) return null;

      try {
        const did = await getDid();
        if (!did) {
          return ociError('NAME_UNKNOWN', 'server not initialized', 404);
        }

        const segments = pathname.split('/').filter((part) => part !== '');
        // ['v2', ...]. The ping answers the Bearer challenge to a request
        // without valid credentials: docker reads its auth scheme from this
        // response, and without the challenge it never attaches credentials
        // to a push. A client with no credentials takes a pull-only token
        // from the realm, which is how anonymous pulls work.
        if (segments.length === 1) {
          if (!(await resolveAccess(request))) return challenge(url);
          return Response.json(
            {},
            { headers: { [API_VERSION_HEADER]: API_VERSION } },
          );
        }
        if (segments.length === 2 && segments[1] === 'token') {
          return await serveToken(request);
        }
        if (segments.length === 2 && segments[1] === '_catalog') {
          return await serveCatalog(did);
        }

        const match = matchRoute(segments.slice(1));
        if (!match) {
          return ociError('UNSUPPORTED', 'unsupported route', 404);
        }
        const { route, name, arg } = match;
        if (!isValidName(name)) {
          return ociError('NAME_INVALID', `invalid name: ${name}`, 400);
        }

        const mutating = !['GET', 'HEAD'].includes(request.method);
        if (mutating) {
          if (!pushEnabled) {
            return ociError('UNSUPPORTED', 'registry is read-only', 405);
          }
          const access = await resolveAccess(request);
          if (!access) return challenge(url);
          if (!access.push) {
            return ociError('DENIED', 'pull-only credentials', 403);
          }
        }

        if (route === 'blobs') {
          if (mutating) {
            return ociError(
              'UNSUPPORTED',
              'blobs are removed with their manifests',
              405,
            );
          }
          return await serveBlob(request, did, arg);
        }

        if (route === 'manifests') {
          if (request.method === 'PUT') {
            return await putManifest(request, did, name, arg);
          }
          if (request.method === 'DELETE') {
            return await deleteManifest(did, name, arg);
          }
          if (mutating) {
            return ociError('UNSUPPORTED', 'unsupported method', 405);
          }
          return await serveManifest(request, did, name, arg);
        }

        if (route === 'tags') {
          if (mutating) {
            return ociError('UNSUPPORTED', 'unsupported method', 405);
          }
          return await serveTagsList(url, did, name);
        }

        if (route === 'referrers') {
          // The referrers API is not implemented; clients fall back to the
          // tag schema on 404.
          return ociError('UNSUPPORTED', 'referrers not supported', 404);
        }

        if (route === 'uploads' && request.method === 'POST') {
          const mount = url.searchParams.get('mount');
          if (mount && isValidDigest(mount)) {
            // Blobs are stored per account rather than per repository, so a
            // mount is an existence check.
            if (await blobMeta(did, mount)) {
              return new Response(null, {
                status: 201,
                headers: {
                  [API_VERSION_HEADER]: API_VERSION,
                  Location: `/v2/${name}/blobs/${mount}`,
                  'Docker-Content-Digest': mount,
                },
              });
            }
          }
          const { id, session } = await createUpload(name);
          const digest = url.searchParams.get('digest');
          if (digest) {
            const failed = await appendBody(session, id, request.body);
            if (failed) return failed;
            return await finishUpload(session, id, digest);
          }
          return uploadAccepted(name, id, session);
        }

        if (route === 'upload') {
          /** @type {UploadSession} */
          let session;
          try {
            session = takeUpload(arg);
          } catch {
            return ociError('BLOB_UPLOAD_UNKNOWN', 'upload unknown', 404);
          }
          if (request.method === 'PATCH') {
            const range = parseContentRange(
              request.headers.get('content-range'),
            );
            if (range && range.start !== session.size) {
              return ociError(
                'BLOB_UPLOAD_INVALID',
                `next chunk starts at ${session.size}`,
                416,
              );
            }
            const failed = await appendBody(session, arg, request.body);
            if (failed) return failed;
            return uploadAccepted(name, arg, session);
          }
          if (request.method === 'PUT') {
            const digest = url.searchParams.get('digest');
            if (!digest) {
              return ociError('DIGEST_INVALID', 'digest missing', 400);
            }
            const failed = await appendBody(session, arg, request.body);
            if (failed) return failed;
            return await finishUpload(session, arg, digest);
          }
          if (request.method === 'DELETE') {
            await dropUpload(arg);
            return new Response(null, {
              status: 204,
              headers: { [API_VERSION_HEADER]: API_VERSION },
            });
          }
          if (request.method === 'GET') {
            return new Response(null, {
              status: 204,
              headers: {
                [API_VERSION_HEADER]: API_VERSION,
                Range: `0-${Math.max(0, session.size - 1)}`,
                'Docker-Upload-UUID': arg,
              },
            });
          }
          return ociError('UNSUPPORTED', 'unsupported method', 405);
        }

        return ociError('UNSUPPORTED', 'unsupported route', 404);
      } catch (err) {
        const message = err instanceof Error ? err.message : String(err);
        return ociError('UNKNOWN', message, 500);
      }
    },
  };
}