/**
 * Celilo Registry Client
 *
 * Implements the Cargo sparse registry protocol for fetching and publishing
 * modules. Default registry is celilo.computer; can be overridden via
 * CELILO_REGISTRY_URL or the --registry flag.
 */

import { createHash } from 'node:crypto';
import { readFile } from 'node:fs/promises';

export const DEFAULT_REGISTRY = 'https://celilo.computer/registry';

/** Attempts per idempotent registry read, including the first. */
const GET_ATTEMPTS = 3;
/** Per-attempt ceiling for metadata reads. Packages use DOWNLOAD_TIMEOUT_MS. */
const GET_TIMEOUT_MS = 30_000;
/** Linear backoff: 1s, then 2s. Short, because a deploy is waiting on this. */
const GET_RETRY_DELAY_MS = 1_000;
/**
 * Per-attempt ceiling for a package download.
 *
 * This was briefly 180s, on the reasoning that a package is megabytes and
 * metadata is not. That was wrong, and measurably so: the whole retry budget has
 * to fit inside the CALLER's window, and celilo's own e2e harness gives a command
 * 120s by default (`packages/e2e/src/container-manager.ts`). Three attempts at
 * 180s allows ~543s, so a slow download stopped failing as a download and started
 * being killed by the caller — a worse error, further from the cause.
 *
 * 30s x 3 attempts plus backoff is ~93s, which fits. The durability comes from
 * retrying, not from a wider window: a 2.6MB transfer that normally takes about a
 * second does not need 180s, it needs another go.
 */
const DOWNLOAD_TIMEOUT_MS = 30_000;

/**
 * Read a body to completion, reporting progress so a failure can say how far it
 * got. `Response.arrayBuffer()` gives no partial count when it throws.
 */
async function readCounting(
  resp: Response,
  onProgress: (bytes: number) => void,
): Promise<ArrayBuffer> {
  if (!resp.body) return resp.arrayBuffer();

  const reader = resp.body.getReader();
  const chunks: Uint8Array[] = [];
  let total = 0;

  try {
    for (;;) {
      const { done, value } = await reader.read();
      if (done) break;
      if (value) {
        chunks.push(value);
        total += value.byteLength;
        onProgress(total);
      }
    }
  } finally {
    reader.releaseLock();
  }

  const out = new Uint8Array(total);
  let offset = 0;
  for (const chunk of chunks) {
    out.set(chunk, offset);
    offset += chunk.byteLength;
  }
  return out.buffer;
}

/**
 * A status worth trying again. 5xx is the server failing, 408 and 429 are it
 * asking us to wait. Every other 4xx is a settled answer.
 */
function isRetryableStatus(status: number): boolean {
  return status >= 500 || status === 408 || status === 429;
}

/**
 * A settled answer from the registry. Retrying cannot change it, so `withRetry`
 * rethrows it immediately rather than spending the backoff on a known 404.
 */
class RegistryAnswer extends Error {}

/** `publish` writes `sha256:<hex>`; tolerate a bare hex digest from older entries. */
function normalizeCksum(value: string): string {
  return value.replace(/^sha256:/, '').toLowerCase();
}

/**
 * Is this index entry's `cksum` an actual digest we can check against?
 *
 * Not every entry carries one. The registry's bootstrap path publishes the
 * literal string `bootstrap` (packages/registry-server/src/bootstrap.ts), because
 * those modules are packaged on demand and have no stable digest to publish.
 * A sentinel means "no integrity data", which is a reason to skip the check and
 * not a reason to reject the package — treating it as a digest rejects every
 * bootstrap-served import.
 */
function isVerifiableCksum(value: string): boolean {
  return /^[0-9a-f]{64}$/.test(normalizeCksum(value));
}

export interface SparseConfig {
  dl: string;
  api: string;
}

export interface IndexEntry {
  name: string;
  vers: string;
  deps: string[];
  cksum: string;
  yanked: boolean;
}

export interface SearchResult {
  name: string;
  max_version: string;
  description: string;
  /** The module's declared glyph, absent when it declared none. */
  icon?: string;
}

export interface SearchResponse {
  modules: SearchResult[];
  total: number;
}

export interface ModuleMetadata {
  name: string;
  versions: Array<{ num: string; yanked: boolean; created_at: string }>;
}

/** A module-owner table entry (ce-1ch). Mirrors the registry-server shape. */
export interface ModuleOwner {
  moduleName: string;
  ownerSub: string;
  claimedAt: string;
  sourceGroup: string;
}

export class RegistryClient {
  readonly baseUrl: string;

  constructor(registryUrl?: string) {
    // Strip trailing slash for consistent path joining
    const raw = registryUrl ?? process.env.CELILO_REGISTRY_URL ?? DEFAULT_REGISTRY;
    this.baseUrl = raw.replace(/\/+$/, '');
  }

  // ── Sparse index ──────────────────────────────────────────────────────────

  async config(): Promise<SparseConfig> {
    const resp = await this.get(`${this.baseUrl}/index/config.json`);
    return resp.json() as Promise<SparseConfig>;
  }

  sparseIndexPath(name: string): string {
    if (name.length === 1) return `${this.baseUrl}/index/1/${name}`;
    if (name.length === 2) return `${this.baseUrl}/index/2/${name}`;
    if (name.length === 3) return `${this.baseUrl}/index/3/${name.slice(0, 2)}/${name}`;
    return `${this.baseUrl}/index/${name.slice(0, 2)}/${name.slice(2, 4)}/${name}`;
  }

  async getIndex(name: string): Promise<IndexEntry[]> {
    const resp = await fetch(this.sparseIndexPath(name), {
      signal: AbortSignal.timeout(15_000),
    });
    if (resp.status === 404) return [];
    if (!resp.ok) throw new Error(`Registry index error: HTTP ${resp.status}`);
    const text = await resp.text();
    return text
      .split('\n')
      .filter(Boolean)
      .map((line) => JSON.parse(line) as IndexEntry);
  }

  /** Returns the latest non-yanked version, or undefined if none. */
  latestVersion(entries: IndexEntry[]): IndexEntry | undefined {
    return [...entries].reverse().find((e) => !e.yanked);
  }

  // ── API ───────────────────────────────────────────────────────────────────

  async search(query: string, perPage = 25): Promise<SearchResponse> {
    const params = new URLSearchParams({ per_page: String(perPage) });
    if (query) params.set('q', query);
    const resp = await this.get(`${this.baseUrl}/api/v1/modules?${params}`);
    return resp.json() as Promise<SearchResponse>;
  }

  async getModule(name: string): Promise<ModuleMetadata> {
    const resp = await this.get(`${this.baseUrl}/api/v1/modules/${encodeURIComponent(name)}`);
    return resp.json() as Promise<ModuleMetadata>;
  }

  downloadUrl(name: string, version: string): string {
    return `${this.baseUrl}/api/v1/modules/${encodeURIComponent(name)}/${encodeURIComponent(version)}/download`;
  }

  /**
   * Fetch a module .netapp, verifying it against the index entry's `cksum`.
   *
   * `cksum` has always been computed at publish time and shipped in every index
   * entry, and nothing on this side ever checked it. A short transfer was
   * therefore accepted as a complete package and failed later in gunzip as
   * "zlib: unexpected end of file", which points at the package rather than at
   * the download that produced it.
   *
   * Verification is also what makes the retry mean anything: a fault nobody can
   * detect is a fault nobody can retry. Pass `expectedCksum` whenever the caller
   * holds the entry — both callers do.
   */
  async download(name: string, version: string, expectedCksum?: string): Promise<ArrayBuffer> {
    const url = this.downloadUrl(name, version);

    return this.withRetry(async () => {
      const startedAt = Date.now();
      const resp = await this.fetchOnce(url, DOWNLOAD_TIMEOUT_MS);

      // Read in chunks so a failure can say how far it got. A bare
      // `arrayBuffer()` that times out reports nothing, which is why three runs
      // of this failure told us only that it was slow and never how slow. Bytes
      // and elapsed together separate a steady trickle from a stall, and those
      // two want different fixes (a bigger window versus an idle timeout).
      let data: ArrayBuffer;
      let received = 0;
      try {
        data = await readCounting(resp, (n) => {
          received = n;
        });
      } catch (err) {
        const seconds = (Date.now() - startedAt) / 1000;
        const rate = seconds > 0 ? received / 1024 / seconds : 0;
        throw new Error(
          `Download of ${name}@${version} failed after ${received} bytes in ${seconds.toFixed(1)}s (${rate.toFixed(0)} KiB/s): ${err instanceof Error ? err.message : String(err)}`,
        );
      }

      if (expectedCksum && isVerifiableCksum(expectedCksum)) {
        const actual = createHash('sha256').update(new Uint8Array(data)).digest('hex');
        if (actual !== normalizeCksum(expectedCksum)) {
          throw new Error(
            `Package ${name}@${version} failed its integrity check: got ${data.byteLength} bytes with sha256 ${actual}, expected ${normalizeCksum(expectedCksum)}. A short or corrupted download, not a bad package.`,
          );
        }
      }

      return data;
    });
  }

  /**
   * Publish a module to the registry.
   *
   * Follows the Cargo binary protocol:
   *   [4-byte LE uint32: JSON length] [JSON metadata] [4-byte LE uint32: file length] [.netapp bytes]
   */
  async publish(opts: {
    name: string;
    version: string;
    netappPath: string;
    token: string;
    /**
     * One-line description from the module's manifest.yml. Optional
     * (server tolerates absence) but recommended — it's what shows
     * up in the registry's browse UI per
     * apps/celilo/designs/REGISTRY_BROWSE_UI.md (Phase 2 step 0).
     */
    description?: string;
    /**
     * The module's `manifest.yml#icon`, when it declares one. Optional and
     * server-tolerated in its absence, exactly like `description`
     * (openspec/changes/module-icons, D4).
     */
    icon?: string;
  }): Promise<{ ok: boolean; name: string; vers: string }> {
    const fileData = await readFile(opts.netappPath);
    const cksum = `sha256:${createHash('sha256').update(fileData).digest('hex')}`;

    const meta = JSON.stringify({
      name: opts.name,
      vers: opts.version,
      deps: [],
      cksum,
      ...(opts.description ? { description: opts.description } : {}),
      ...(opts.icon ? { icon: opts.icon } : {}),
    });
    const metaBuf = Buffer.from(meta, 'utf-8');

    const body = Buffer.allocUnsafe(4 + metaBuf.length + 4 + fileData.length);
    body.writeUInt32LE(metaBuf.length, 0);
    metaBuf.copy(body, 4);
    body.writeUInt32LE(fileData.length, 4 + metaBuf.length);
    fileData.copy(body, 4 + metaBuf.length + 4);

    const resp = await fetch(`${this.baseUrl}/api/v1/modules/new`, {
      method: 'PUT',
      headers: { Authorization: opts.token, 'Content-Type': 'application/octet-stream' },
      body,
      signal: AbortSignal.timeout(120_000),
    });

    if (!resp.ok) {
      throw new Error(await this.errorDetail(resp));
    }

    return resp.json() as Promise<{ ok: boolean; name: string; vers: string }>;
  }

  // ── Module ownership (admin — ce-1ch) ──────────────────────────────────────

  /** List the whole module-owner table. Requires an admin token. */
  async listOwners(token: string): Promise<ModuleOwner[]> {
    const resp = await this.authed(`${this.baseUrl}/api/v1/modules/owners`, token);
    const body = (await resp.json()) as { owners: ModuleOwner[] };
    return body.owners;
  }

  /** Show the owner of one module name, or null if unclaimed. Requires an admin token. */
  async getOwner(name: string, token: string): Promise<ModuleOwner | null> {
    const url = `${this.baseUrl}/api/v1/modules/owners/${encodeURIComponent(name)}`;
    const resp = await fetch(url, {
      headers: { Authorization: token },
      signal: AbortSignal.timeout(30_000),
    });
    if (resp.status === 404) return null;
    if (!resp.ok) throw new Error(await this.errorDetail(resp));
    const body = (await resp.json()) as { owner: ModuleOwner };
    return body.owner;
  }

  /** Reassign a module name to `ownerSub` (admin overwrite). Requires an admin token. */
  async setOwner(name: string, ownerSub: string, token: string): Promise<ModuleOwner> {
    const url = `${this.baseUrl}/api/v1/modules/owners/${encodeURIComponent(name)}`;
    const resp = await fetch(url, {
      method: 'POST',
      headers: { Authorization: token, 'Content-Type': 'application/json' },
      body: JSON.stringify({ ownerSub }),
      signal: AbortSignal.timeout(30_000),
    });
    if (!resp.ok) throw new Error(await this.errorDetail(resp));
    const body = (await resp.json()) as { owner: ModuleOwner };
    return body.owner;
  }

  // ── Internal ──────────────────────────────────────────────────────────────

  /** GET with an Authorization header, throwing the server's error detail on failure. */
  private async authed(url: string, token: string): Promise<Response> {
    const resp = await fetch(url, {
      headers: { Authorization: token },
      signal: AbortSignal.timeout(30_000),
    });
    if (!resp.ok) throw new Error(await this.errorDetail(resp));
    return resp;
  }

  /** Extract `errors[0].detail` from a Cargo-protocol error body, or fall back to the status. */
  private async errorDetail(resp: Response): Promise<string> {
    // The body can be a JSON error object, JSON null, prose, or nothing. Only
    // the first carries a detail; everything else falls back to the status.
    const body = (await resp.json().catch(() => null)) as {
      errors?: Array<{ detail: string }>;
    } | null;
    const detail = body?.errors?.[0]?.detail;
    return detail ? `${detail} (HTTP ${resp.status})` : `HTTP ${resp.status}`;
  }

  private async get(url: string): Promise<Response> {
    // 30s per attempt: a module .netapp can be tens of MB, and the download of a
    // ~19MB module over a multi-hop path (e2e sim NAT, real WAN) intermittently
    // times out mid-transfer.
    //
    // The window used to be the whole story — one attempt, and the response to
    // observed flakiness was widening it from 15s to 30s. A wider single window
    // is not a durable fetch, it is a bigger gap to fall through. Measured
    // 2026-09-04: `module import iptables` failed twice with "Download failed:
    // The operation timed out" on a loaded host, while the identical call had
    // succeeded minutes earlier. This client is the fleet's module delivery
    // path, so a single transient failing an import is an operator-facing
    // outage, not just an e2e flake.
    //
    // Every caller of get() is an idempotent read (config, search, metadata,
    // download), so retrying is safe. `publish` does not route through here.
    // A 4xx is an answer and is never retried: a 404 must stay fast and say
    // "not found" rather than stall for the whole backoff.
    return this.withRetry(() => this.fetchOnce(url));
  }

  /** One attempt. Throws `RegistryAnswer` for a status retrying cannot change. */
  private async fetchOnce(url: string, timeoutMs = GET_TIMEOUT_MS): Promise<Response> {
    const resp = await fetch(url, { signal: AbortSignal.timeout(timeoutMs) });
    if (resp.ok) return resp;

    const message = `Registry error at ${url}: HTTP ${resp.status}`;
    if (!isRetryableStatus(resp.status)) throw new RegistryAnswer(message);
    throw new Error(message);
  }

  /**
   * Retry a whole operation, not a piece of one.
   *
   * The unit matters. Retrying only the request leaves the body download
   * outside the retry, and a body is where a large transfer actually fails.
   */
  private async withRetry<T>(run: () => Promise<T>): Promise<T> {
    const startedAt = Date.now();
    let lastError: unknown;

    for (let attempt = 1; attempt <= GET_ATTEMPTS; attempt++) {
      try {
        return await run();
      } catch (err) {
        if (err instanceof RegistryAnswer) throw err;
        lastError = err;
        if (attempt < GET_ATTEMPTS) {
          await new Promise((resolve) => setTimeout(resolve, GET_RETRY_DELAY_MS * attempt));
        }
      }
    }

    // Exhausting the budget is itself the event worth reporting. The bare last
    // error said only what the final attempt said, which is the same string the
    // caller got before the retry existed, so the operator cannot tell whether
    // the retry ran (celilo#1264). The original error rides along as `cause`.
    const seconds = (Date.now() - startedAt) / 1000;
    const detail = lastError instanceof Error ? lastError.message : String(lastError);
    throw new Error(
      `Registry fetch failed after ${GET_ATTEMPTS} attempts over ${seconds.toFixed(1)}s: ${detail}`,
      { cause: lastError },
    );
  }
}