import { randomId } from "../crypto";
import { recordUsageBuffered } from "./buffered";
import { addCounter } from "../counters";
import { REPO_PACK_BYTES } from "../counter-keys";

/**
 * Wraps a body so `onDone(bytes)` runs exactly once, when it has fully streamed or the client cancelled
 * (bytes already delivered are still metered).
 */
export function countingBody(
  body: ReadableStream<Uint8Array> | null,
  onDone: (bytes: number) => void,
  onChunk?: (chunk: Uint8Array) => void,
): ReadableStream<Uint8Array> | null {
  if (!body) {
    onDone(0);
    return null;
  }
  const reader = body.getReader();
  let bytes = 0;
  let done = false;
  const finish = () => {
    if (done) return;
    done = true;
    try {
      onDone(bytes);
    } catch (error) {
      console.warn("metering callback failed", error);
    }
  };
  return new ReadableStream<Uint8Array>({
    async pull(controller) {
      try {
        const next = await reader.read();
        if (next.done) {
          finish();
          controller.close();
          return;
        }
        bytes += next.value.byteLength;
        onChunk?.(next.value);
        controller.enqueue(next.value);
      } catch (error) {
        finish();
        controller.error(error);
      }
    },
    async cancel(reason) {
      finish();
      await reader.cancel(reason);
    },
  });
}

/**
 * Meters one git operation through the proxy: a count (`op` = push, fetch or clone) plus bytes in or out.
 * Pushes also grow the repository storage estimate by the received pack size.
 */
export async function meterGitOp(input: {
  accountId: string;
  repoId: string;
  op: "push" | "fetch" | "clone";
  bytes: number;
  requestId: string;
}): Promise<void> {
  const direction = input.op === "push" ? "in" : "out";
  await recordUsageBuffered({
    accountId: input.accountId,
    repoId: input.repoId,
    meter: "ops.git_ops",
    quantity: 1,
    attributes: { op: input.op },
    idempotencyKey: `git:${input.requestId}:op`,
  });
  const inserted = await recordUsageBuffered({
    accountId: input.accountId,
    repoId: input.repoId,
    meter: "ops.git_bytes",
    quantity: input.bytes,
    unit: "bytes",
    attributes: { op: input.op, direction },
    idempotencyKey: `git:${input.requestId}:bytes`,
  });
  if (inserted && input.op === "push") {
    await addCounter("repo-counts", input.repoId, REPO_PACK_BYTES, [input.bytes, 0, 0]);
  }
}

const HAVE = new TextEncoder().encode("have ");

/** True when `chunk` contains a pkt-line `have` (a clone sends none; a fetch negotiates with them). */
export function containsHave(chunk: Uint8Array): boolean {
  outer: for (let index = 0; index + HAVE.length <= chunk.length; index += 1) {
    for (let offset = 0; offset < HAVE.length; offset += 1) if (chunk[index + offset] !== HAVE[offset]) continue outer;
    return true;
  }
  return false;
}

/** Streaming `have` detector that also catches a `have ` split across chunk boundaries. */
export function haveDetector() {
  let tail = new Uint8Array(0);
  let seen = false;
  return {
    push(chunk: Uint8Array) {
      if (seen) return;
      const joined = new Uint8Array(tail.length + chunk.length);
      joined.set(tail);
      joined.set(chunk, tail.length);
      seen = containsHave(joined);
      tail = joined.slice(Math.max(0, joined.length - (HAVE.length - 1)));
    },
    get seen() {
      return seen;
    },
  };
}

/**
 * Meters a raw, archive or release-asset download as `network.downloads`, billed to the repo owner once the
 * body has streamed. The owner lookup runs after the response, off the request's critical path.
 */
export function meteredDownload(
  response: Response,
  kind: "raw" | "archive" | "asset",
  lookup: () => Promise<{ id: string; owner_id: string } | null>,
  waitUntil: (promise: Promise<unknown>) => void,
): Response {
  if (!response.ok || !response.body) return response;
  const requestId = randomId();
  const body = countingBody(response.body, (bytes) =>
    waitUntil(
      Promise.resolve()
        .then(lookup)
        .then((repo) =>
          repo
            ? recordUsageBuffered({
                accountId: repo.owner_id,
                repoId: repo.id,
                meter: "network.downloads",
                quantity: bytes,
                unit: "bytes",
                attributes: { kind },
                idempotencyKey: `download:${requestId}`,
              })
            : false,
        )
        .catch((error: unknown) => console.warn("download metering failed", error)),
    ),
  );
  return new Response(body, { status: response.status, statusText: response.statusText, headers: response.headers });
}