/**
 * npm registry HTTP API served from dev.pdsjs.npm.package records, so the
 * stock npm client can publish and install:
 *
 *   npm config set registry https://<pds-host>/npm/
 *   npm publish
 *   npm install <package>
 *
 * The surface lives under /npm/ because the npm protocol wants the URL
 * root, which the PDS owns. Tarball and manifest bytes are stored as blobs
 * exactly as published; the record holds the dist-tags, the integrity
 * strings and a blob reference per version, which is what holds the blobs
 * against orphan cleanup. The packument is assembled from the manifest
 * blobs, with each tarball URL rewritten to the requesting origin.
 *
 * 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 installs only.
 *
 * Authentication is the account's own: `npm login` verifies through
 * createSession and answers the password back as the token, so the stored
 * `_authToken` is the app password itself and nothing server-side has to
 * survive a restart. Installs are anonymous.
 *
 * This module must stay free of node-only imports; it runs on Cloudflare
 * Workers.
 */

import {
  cborEncodeDagCbor,
  cidToString,
  createBlobCid,
  createCid,
} from '@pdsjs/core/repo';
import { createSha256 } from '@pdsjs/core/sha256';
import { fromBase64, tarballHashes } from './integrity.js';
import { isValidName, NPM_PACKAGE_COLLECTION, nameToRkey } from './lexicon.js';
import {
  blobRef,
  parsePackageRecord,
  recordBlobCids,
  serializePackageRecord,
} from './record.js';

const TARBALL_MIME_TYPE = 'application/octet-stream';
const MANIFEST_MIME_TYPE = 'application/json';

/** Largest accepted tarball, when the caller sets no other bound. */
const DEFAULT_MAX_TARBALL_SIZE = 64 * 1024 * 1024;

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

const VERSION_PATTERN = /^[0-9A-Za-z.+-]{1,256}$/;
const TAG_PATTERN = /^[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}$/;

/**
 * @typedef {Object} NpmRegistryContext
 * @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] - publishes need it
 * @property {import('@pdsjs/core/ports').BlobPort} [blobs] - publishes need it
 * @property {(did: string, ops: Array<{action: string, uri: string, cid?: string}>) => Promise<Object>} [createCommit] - publishes need it
 * @property {import('@pdsjs/core/ports').HasherFactory} [hasher] - SHA-256 factory; absent, the plain JavaScript one
 * @property {number} [maxTarballSize] - largest accepted tarball, in bytes
 * @property {string} [collection] - record collection, for lexicon forks
 */

/**
 * @param {string} message
 * @param {number} status
 * @returns {Response}
 */
function npmError(message, status) {
  return Response.json({ error: message }, { status });
}

/**
 * The name without its scope, which names the tarball file.
 * @param {string} name
 * @returns {string}
 */
function basename(name) {
  return name.includes('/') ? name.split('/')[1] : name;
}

/**
 * @param {URL} url
 * @param {string} name
 * @param {string} version
 * @returns {string}
 */
function tarballUrl(url, name, version) {
  return `${url.origin}/npm/${name}/-/${basename(name)}-${version}.tgz`;
}

/**
 * @param {NpmRegistryContext} ctx
 * @returns {{handle: (request: Request) => Promise<Response|null>}}
 */
export function createNpmRegistry(ctx) {
  if (!ctx || typeof ctx.xrpc !== 'function') {
    throw new Error('createNpmRegistry requires an xrpc fetch function');
  }
  if (typeof ctx.getDid !== 'function') {
    throw new Error('createNpmRegistry requires getDid');
  }
  const {
    xrpc,
    getDid,
    actorStorage,
    blobs,
    createCommit,
    hasher = createSha256,
    maxTarballSize = DEFAULT_MAX_TARBALL_SIZE,
    collection = NPM_PACKAGE_COLLECTION,
  } = ctx;
  const publishEnabled = 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 ─────────────────────────────────────────────────────

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

  /**
   * @param {string} identifier
   * @param {string} password
   * @returns {Promise<boolean>}
   */
  async function credentialsValid(identifier, password) {
    const hash = hasher();
    await hash.update(new TextEncoder().encode(`${identifier}\0${password}`));
    const digest = await hash.digest();
    let key = '';
    for (const byte of digest) key += byte.toString(16).padStart(2, '0');

    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;
  }

  /**
   * True when the request carries credentials createSession accepts. A
   * Bearer token is the password alone; the account is the identifier,
   * since this server holds exactly one.
   * @param {Request} request
   * @returns {Promise<boolean>}
   */
  async function authorized(request) {
    const header = request.headers.get('authorization') ?? '';
    if (/^bearer\s/i.test(header)) {
      const did = await getDid();
      if (!did) return false;
      return credentialsValid(did, header.slice(6).trim());
    }
    if (/^basic\s/i.test(header)) {
      let decoded;
      try {
        decoded = atob(header.slice(5).trim());
      } catch {
        return false;
      }
      const didMatch = /^(did:[a-z0-9]+:[^:]+):(.*)$/s.exec(decoded);
      const at = decoded.indexOf(':');
      if (!didMatch && at < 0) return false;
      return credentialsValid(
        didMatch ? didMatch[1] : decoded.slice(0, at),
        didMatch ? didMatch[2] : decoded.slice(at + 1),
      );
    }
    return false;
  }

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

  /**
   * @param {string} did
   * @param {string} name
   * @returns {Promise<{cid: string, record: import('./record.js').NpmPackageRecord}|null>}
   */
  async function loadPackage(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 {{cid: string, value: unknown}} */ (
      await res.json()
    );
    const record = parsePackageRecord(body.value);
    return record ? { cid: body.cid, record } : 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 /npm/<name>: the packument, assembled from the manifest blobs with
   * tarball URLs rewritten to the requesting origin.
   * @param {Request} request
   * @param {URL} url
   * @param {string} did
   * @param {string} name
   * @returns {Promise<Response>}
   */
  async function servePackument(request, url, did, name) {
    const found = await loadPackage(did, name);
    if (!found) return npmError(`package not found: ${name}`, 404);
    const { cid, record } = found;

    const ifNoneMatch = request.headers.get('if-none-match');
    if (ifNoneMatch?.includes(cid)) {
      return new Response(null, { status: 304, headers: { ETag: `"${cid}"` } });
    }

    /** @type {Record<string, unknown>} */
    const versions = {};
    /** @type {Record<string, string>} */
    const time = {};
    for (const entry of record.versions) {
      const bytes = await readBlob(did, entry.manifest.ref.$link);
      if (!bytes) continue;
      /** @type {Record<string, unknown>} */
      let manifest;
      try {
        manifest = JSON.parse(new TextDecoder().decode(bytes));
      } catch {
        continue;
      }
      manifest.dist = {
        .../** @type {Record<string, unknown>} */ (manifest.dist ?? {}),
        tarball: tarballUrl(url, name, entry.version),
        integrity: entry.integrity,
        ...(entry.shasum ? { shasum: entry.shasum } : {}),
      };
      versions[entry.version] = manifest;
      if (entry.publishedAt) time[entry.version] = entry.publishedAt;
    }

    /** @type {Record<string, string>} */
    const distTags = {};
    for (const tag of record.distTags) distTags[tag.tag] = tag.version;

    return Response.json(
      {
        name,
        'dist-tags': distTags,
        versions,
        time: {
          created: record.createdAt,
          modified: record.updatedAt ?? record.createdAt,
          ...time,
        },
      },
      { headers: { ETag: `"${cid}"` } },
    );
  }

  /**
   * GET /npm/<name>/-/<file>.tgz: the version's tarball, streamed from
   * blob storage.
   * @param {URL} _url
   * @param {string} did
   * @param {string} name
   * @param {string} file
   * @returns {Promise<Response>}
   */
  async function serveTarball(_url, did, name, file) {
    const prefix = `${basename(name)}-`;
    if (!file.startsWith(prefix) || !file.endsWith('.tgz')) {
      return npmError(`tarball not found: ${file}`, 404);
    }
    const version = file.slice(prefix.length, -4);
    const found = await loadPackage(did, name);
    const entry = found?.record.versions.find((v) => v.version === version);
    if (!entry) return npmError(`tarball not found: ${file}`, 404);

    const params = new URLSearchParams({ did, cid: entry.tarball.ref.$link });
    const res = await xrpcFetch(`/xrpc/com.atproto.sync.getBlob?${params}`);
    if (!res.ok) return npmError(`tarball not found: ${file}`, 404);
    return new Response(res.body, {
      headers: {
        'Content-Type': TARBALL_MIME_TYPE,
        'Content-Length':
          res.headers.get('content-length') ?? String(entry.tarball.size),
      },
    });
  }

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

  /**
   * One write to a package record at a time, so concurrent publishes
   * 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;
  }

  /**
   * 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').NpmPackageRecord} record
   * @param {boolean} existed
   */
  async function writePackageRecord(did, name, record, existed) {
    const storage = /** @type {NonNullable<typeof actorStorage>} */ (
      actorStorage
    );
    const rkey = nameToRkey(name);
    const uri = `at://${did}/${collection}/${rkey}`;
    const encoded = cborEncodeDagCbor(serializePackageRecord(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 {string} did
   * @param {Uint8Array} bytes
   * @param {string} mimeType
   * @returns {Promise<string>} the blob's CID
   */
  async function storeBlob(did, bytes, mimeType) {
    const cid = cidToString(await createBlobCid(bytes));
    await /** @type {NonNullable<typeof blobs>} */ (blobs).put(
      did,
      cid,
      bytes,
      mimeType,
    );
    await /** @type {NonNullable<typeof actorStorage>} */ (
      actorStorage
    ).putBlob(cid, mimeType, bytes.length);
    return cid;
  }

  /**
   * Read the whole request body, refusing one past the cap. The publish
   * body carries the tarball base64-encoded, so the cap allows for the
   * encoding overhead.
   * @param {Request} request
   * @param {number} cap
   * @returns {Promise<Uint8Array|null>} null when the body exceeds the cap
   */
  async function readBody(request, cap) {
    if (!request.body) return new Uint8Array(0);
    const chunks = [];
    let size = 0;
    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 > cap) {
        await reader.cancel();
        return null;
      }
      chunks.push(chunk);
    }
    const joined = new Uint8Array(size);
    let at = 0;
    for (const chunk of chunks) {
      joined.set(chunk, at);
      at += chunk.length;
    }
    return joined;
  }

  /**
   * PUT /npm/<name>: publish. The body is the package document npm sends:
   * version manifests plus the tarballs base64-encoded in `_attachments`.
   * @param {Request} request
   * @param {URL} url
   * @param {string} did
   * @param {string} name
   * @returns {Promise<Response>}
   */
  async function publish(request, url, did, name) {
    // base64 inflates the tarball by a third; the slack covers manifests.
    const body = await readBody(
      request,
      Math.ceil(maxTarballSize * 1.4) + 1024 * 1024,
    );
    if (!body) {
      return npmError(
        `package exceeds the maximum size of ${maxTarballSize} bytes`,
        413,
      );
    }

    /** @type {Record<string, unknown>} */
    let doc;
    try {
      doc = JSON.parse(new TextDecoder().decode(body));
    } catch {
      return npmError('the publish body is not valid JSON', 400);
    }
    if (doc.name !== name) {
      return npmError('the document name does not match the URL', 400);
    }
    const versions = /** @type {Record<string, Record<string, unknown>>} */ (
      doc.versions && typeof doc.versions === 'object' ? doc.versions : {}
    );
    const attachments =
      /** @type {Record<string, {data?: unknown, length?: number}>} */ (
        doc._attachments && typeof doc._attachments === 'object'
          ? doc._attachments
          : {}
      );
    const docTags = /** @type {Record<string, unknown>} */ (
      doc['dist-tags'] && typeof doc['dist-tags'] === 'object'
        ? doc['dist-tags']
        : {}
    );
    if (Object.keys(versions).length === 0) {
      return npmError('the publish body names no versions', 400);
    }

    return withRecordLock(name, async () => {
      const existing = await loadPackage(did, name);
      const now = new Date().toISOString();
      /** @type {import('./record.js').NpmPackageRecord} */
      const record = existing?.record ?? {
        name,
        distTags: [],
        versions: [],
        createdAt: now,
      };

      for (const [version, manifest] of Object.entries(versions)) {
        if (!VERSION_PATTERN.test(version)) {
          return npmError(`invalid version: ${version}`, 400);
        }
        if (record.versions.some((v) => v.version === version)) {
          // npm expects 403 here and reports the version conflict itself.
          return npmError(
            `cannot publish over the previously published version ${version}`,
            403,
          );
        }
        const attachment =
          attachments[`${basename(name)}-${version}.tgz`] ??
          Object.values(attachments)[0];
        if (typeof attachment?.data !== 'string') {
          return npmError(`no tarball attached for version ${version}`, 400);
        }
        /** @type {Uint8Array} */
        let tarball;
        try {
          tarball = fromBase64(attachment.data);
        } catch {
          return npmError('the tarball attachment is not valid base64', 400);
        }
        if (tarball.length > maxTarballSize) {
          return npmError(
            `package exceeds the maximum size of ${maxTarballSize} bytes`,
            413,
          );
        }
        const hashes = await tarballHashes(tarball);
        const declared = /** @type {{integrity?: unknown}} */ (
          manifest.dist ?? {}
        ).integrity;
        if (typeof declared === 'string' && declared !== hashes.integrity) {
          return npmError(
            `the tarball hashes to ${hashes.integrity}, not the declared integrity`,
            400,
          );
        }

        const tarballCid = await storeBlob(did, tarball, TARBALL_MIME_TYPE);
        const manifestBytes = new TextEncoder().encode(
          JSON.stringify(manifest),
        );
        const manifestCid = await storeBlob(
          did,
          manifestBytes,
          MANIFEST_MIME_TYPE,
        );
        record.versions.push({
          version,
          tarball: blobRef(tarballCid, TARBALL_MIME_TYPE, tarball.length),
          manifest: blobRef(
            manifestCid,
            MANIFEST_MIME_TYPE,
            manifestBytes.length,
          ),
          integrity: hashes.integrity,
          shasum: hashes.shasum,
          publishedAt: now,
        });
      }

      for (const [tag, version] of Object.entries(docTags)) {
        if (typeof version !== 'string' || !TAG_PATTERN.test(tag)) continue;
        const entry = record.distTags.find((t) => t.tag === tag);
        if (entry) entry.version = version;
        else record.distTags.push({ tag, version });
      }

      record.updatedAt = now;
      await writePackageRecord(did, name, record, Boolean(existing));
      return Response.json({ ok: true, id: name }, { status: 201 });
    });
  }

  /**
   * Change one dist-tag: PUT with the version as a JSON string, or DELETE.
   * @param {Request} request
   * @param {string} did
   * @param {string} name
   * @param {string} tag
   * @returns {Promise<Response>}
   */
  async function changeDistTag(request, did, name, tag) {
    return withRecordLock(name, async () => {
      const found = await loadPackage(did, name);
      if (!found) return npmError(`package not found: ${name}`, 404);
      const { record } = found;

      if (request.method === 'DELETE') {
        if (!record.distTags.some((t) => t.tag === tag)) {
          return npmError(`tag not found: ${tag}`, 404);
        }
        record.distTags = record.distTags.filter((t) => t.tag !== tag);
      } else {
        let version;
        try {
          version = JSON.parse(await request.text());
        } catch {
          return npmError('the body is not a JSON version string', 400);
        }
        if (
          typeof version !== 'string' ||
          !record.versions.some((v) => v.version === version)
        ) {
          return npmError(`version not found: ${version}`, 404);
        }
        const entry = record.distTags.find((t) => t.tag === tag);
        if (entry) entry.version = version;
        else record.distTags.push({ tag, version });
      }

      record.updatedAt = new Date().toISOString();
      await writePackageRecord(did, name, record, true);
      return Response.json({ ok: true });
    });
  }

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

  /**
   * @param {string[]} segments - path segments after /npm/
   * @returns {{name: string, rest: string[]}|null}
   */
  function parseName(segments) {
    if (segments.length === 0) return null;
    if (segments[0].startsWith('@')) {
      if (segments.length < 2) return null;
      return {
        name: `${segments[0]}/${segments[1]}`,
        rest: segments.slice(2),
      };
    }
    return { name: segments[0], rest: segments.slice(1) };
  }

  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 !== '/npm' && !pathname.startsWith('/npm/')) return null;

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

        const tail = pathname.slice('/npm'.length);
        const segments = tail.split('/').filter((part) => part !== '');

        if (segments.length === 0) {
          return Response.json({});
        }

        // Meta routes, the ones npm serves under /-/.
        if (segments[0] === '-') {
          if (segments[1] === 'ping') return Response.json({});

          // PUT /-/user/org.couchdb.user:<name>, the npm login shape.
          if (
            request.method === 'PUT' &&
            segments[1] === 'user' &&
            segments.length === 3
          ) {
            /** @type {{name?: unknown, password?: unknown}} */
            let body;
            try {
              body = JSON.parse(await request.text());
            } catch {
              return npmError('the login body is not valid JSON', 400);
            }
            if (
              typeof body.name !== 'string' ||
              typeof body.password !== 'string' ||
              !(await credentialsValid(body.name, body.password))
            ) {
              return npmError('invalid credentials', 401);
            }
            // The token is the password itself: createSession verifies it on
            // every use, so nothing server-side has to survive a restart.
            return Response.json(
              {
                ok: true,
                id: `org.couchdb.user:${body.name}`,
                token: body.password,
              },
              { status: 201 },
            );
          }

          if (segments[1] === 'whoami') {
            if (!(await authorized(request))) {
              return npmError('authentication required', 401);
            }
            const res = await xrpcFetch(
              `/xrpc/com.atproto.repo.describeRepo?repo=${encodeURIComponent(did)}`,
            );
            const handle = res.ok
              ? /** @type {{handle?: string}} */ (await res.json()).handle
              : null;
            return Response.json({ username: handle || did });
          }

          if (segments[1] === 'package') {
            const parsed = parseName(segments.slice(2));
            if (!parsed || !isValidName(parsed.name)) {
              return npmError('invalid package name', 400);
            }
            if (parsed.rest[0] !== 'dist-tags') {
              return npmError('unsupported route', 404);
            }
            if (request.method === 'GET') {
              const found = await loadPackage(did, parsed.name);
              if (!found) {
                return npmError(`package not found: ${parsed.name}`, 404);
              }
              /** @type {Record<string, string>} */
              const tags = {};
              for (const tag of found.record.distTags) {
                tags[tag.tag] = tag.version;
              }
              return Response.json(tags);
            }
            const tag = parsed.rest[1];
            if (!tag || !TAG_PATTERN.test(tag)) {
              return npmError('invalid tag', 400);
            }
            if (!publishEnabled) {
              return npmError('this registry is read-only', 405);
            }
            if (!(await authorized(request))) {
              return npmError('authentication required', 401);
            }
            return await changeDistTag(request, did, parsed.name, tag);
          }

          // npm probes audit and search endpoints during installs; a plain
          // 404 makes the client skip the feature.
          return npmError('unsupported route', 404);
        }

        const parsed = parseName(segments);
        if (!parsed || !isValidName(parsed.name)) {
          return npmError('invalid package name', 400);
        }
        const { name, rest } = parsed;

        if (rest.length === 2 && rest[0] === '-') {
          if (request.method !== 'GET') {
            return npmError('unsupported method', 405);
          }
          return await serveTarball(url, did, name, rest[1]);
        }

        if (rest.length === 0) {
          if (request.method === 'GET') {
            return await servePackument(request, url, did, name);
          }
          if (request.method === 'PUT') {
            if (!publishEnabled) {
              return npmError('this registry is read-only', 405);
            }
            if (!(await authorized(request))) {
              return npmError('authentication required', 401);
            }
            return await publish(request, url, did, name);
          }
          if (request.method === 'DELETE') {
            return npmError(
              'unpublish is not supported; delete the dev.pdsjs.npm.package record instead',
              403,
            );
          }
        }

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