import { randomId } from "../crypto";
import { run } from "../db";
import { recordUsage } from "./record";

/** Wraps a response body so `onDone(bytes)` runs once the body has streamed (or failed) to the client. */
export function countingBody(
  body: ReadableStream<Uint8Array> | null,
  onDone: (bytes: number) => void,
  onChunk?: (chunk: Uint8Array) => void,
) {
  if (!body) {
    onDone(0);
    return null;
  }
  let bytes = 0;
  let done = false;
  const finish = () => {
    if (done) return;
    done = true;
    onDone(bytes);
  };
  return body.pipeThrough(
    new TransformStream<Uint8Array, Uint8Array>({
      transform(chunk, controller) {
        bytes += chunk.byteLength;
        onChunk?.(chunk);
        controller.enqueue(chunk);
      },
      flush: finish,
    }),
  );
}

/**
 * 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 recordUsage({
    accountId: input.accountId,
    repoId: input.repoId,
    meter: "ops.git_ops",
    quantity: 1,
    attributes: { op: input.op },
    idempotencyKey: `git:${input.requestId}:op`,
  });
  const inserted = await recordUsage({
    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 run(
      `INSERT INTO repo_storage_estimates (repo_id, pack_bytes, updated_at) VALUES (?, ?, ?)
       ON CONFLICT (repo_id) DO UPDATE SET pack_bytes = pack_bytes + excluded.pack_bytes, updated_at = excluded.updated_at`,
      input.repoId,
      input.bytes,
      Date.now(),
    );
  }
}

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

/**
 * 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(
      lookup()
        .then((repo) =>
          repo
            ? recordUsage({
                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 });
}