import { env } from "cloudflare:workers";
import { MAX_IN_LIST, all, chunks, first, placeholders, run } from "../db";
import { HttpError } from "../errors";
import { adjustStorage } from "./quota";
import type { RepoRow } from "~/lib/types";

/** Forks read LFS objects from their parents; bound the chain so lookups stay O(1) in D1 calls. */
const MAX_FORK_DEPTH = 3;
const PURGE_PAGES = 10;
const R2_LIST_LIMIT = 1000;

type ObjectRow = { repo_id: string; size: number };

export function objectKey(repoId: string, oid: string): string {
  return `lfs/${repoId}/${oid}`;
}

async function repoChain(repo: Pick<RepoRow, "id" | "fork_of_id">): Promise<string[]> {
  const chain = [repo.id];
  let parent = repo.fork_of_id;
  while (parent && chain.length < MAX_FORK_DEPTH) {
    chain.push(parent);
    const row = await first<{ fork_of_id: string | null }>("SELECT fork_of_id FROM repos WHERE id = ?", parent);
    parent = row?.fork_of_id ?? null;
  }
  return chain;
}

/** Finds which repo in the fork chain stores the object, preferring the repo itself. */
export async function findObject(repo: Pick<RepoRow, "id" | "fork_of_id">, oid: string): Promise<ObjectRow | null> {
  const own = await first<ObjectRow>(
    "SELECT repo_id, size FROM lfs_objects WHERE repo_id = ? AND oid = ?",
    repo.id,
    oid,
  );
  if (own || !repo.fork_of_id) return own;
  const chain = (await repoChain(repo)).slice(1);
  const rows = await all<ObjectRow>(
    `SELECT repo_id, size FROM lfs_objects WHERE oid = ? AND repo_id IN (${placeholders(chain.length)})`,
    oid,
    ...chain,
  );
  return chain.map((id) => rows.find((row) => row.repo_id === id)).find((row) => row !== undefined) ?? null;
}

/** Bulk variant of findObject for batch requests; IN-lists are chunked to stay under D1's parameter cap. */
export async function findObjects(
  repo: Pick<RepoRow, "id" | "fork_of_id">,
  oids: string[],
): Promise<Map<string, ObjectRow>> {
  const chain = await repoChain(repo);
  const found = new Map<string, ObjectRow & { rank: number }>();
  for (const group of chunks(oids, MAX_IN_LIST - chain.length)) {
    const rows = await all<ObjectRow & { oid: string }>(
      `SELECT repo_id, oid, size FROM lfs_objects
       WHERE repo_id IN (${placeholders(chain.length)}) AND oid IN (${placeholders(group.length)})`,
      ...chain,
      ...group,
    );
    for (const row of rows) {
      const rank = chain.indexOf(row.repo_id);
      const current = found.get(row.oid);
      if (!current || rank < current.rank) found.set(row.oid, { repo_id: row.repo_id, size: row.size, rank });
    }
  }
  return new Map([...found].map(([oid, row]) => [oid, { repo_id: row.repo_id, size: row.size }]));
}

export async function openObject(repo: Pick<RepoRow, "id" | "fork_of_id">, oid: string): Promise<R2ObjectBody | null> {
  const found = await findObject(repo, oid);
  return found ? env.BUCKET.get(objectKey(found.repo_id, oid)) : null;
}

/** Streams into R2 with a declared SHA-256 so R2 itself rejects content that does not match the oid. */
export async function storeObject(
  repo: Pick<RepoRow, "id" | "owner_id">,
  oid: string,
  size: number,
  body: ReadableStream<Uint8Array>,
): Promise<void> {
  const sized = body.pipeThrough(new FixedLengthStream(size));
  try {
    await env.BUCKET.put(objectKey(repo.id, oid), sized, {
      sha256: oid,
      httpMetadata: { contentType: "application/octet-stream" },
    });
  } catch (error) {
    console.warn("LFS upload rejected", oid, error);
    throw new HttpError(422, "Uploaded content does not match the declared oid and size");
  }
  const result = await run(
    "INSERT INTO lfs_objects (repo_id, oid, size, created_at) VALUES (?, ?, ?, ?) ON CONFLICT DO NOTHING",
    repo.id,
    oid,
    size,
    Date.now(),
  );
  if (result.meta.changes > 0) await adjustStorage(repo.owner_id, size);
}

/** Releases an owner's quota and deletes stored objects when a repository is removed (bounded per call). */
export async function purgeRepoLfs(repo: Pick<RepoRow, "id" | "owner_id">): Promise<void> {
  const total = await first<{ bytes: number | null }>(
    "SELECT SUM(size) AS bytes FROM lfs_objects WHERE repo_id = ?",
    repo.id,
  );
  await adjustStorage(repo.owner_id, -(total?.bytes ?? 0));
  await run("DELETE FROM lfs_objects WHERE repo_id = ?", repo.id);
  let cursor: string | undefined;
  for (let page = 0; page < PURGE_PAGES; page++) {
    const listing = await env.BUCKET.list({ prefix: `lfs/${repo.id}/`, cursor, limit: R2_LIST_LIMIT });
    if (listing.objects.length > 0) await env.BUCKET.delete(listing.objects.map((object) => object.key));
    if (!listing.truncated) return;
    cursor = listing.cursor;
  }
}