import { env } from "cloudflare:workers";
import { MAX_IN_LIST, all, batch, chunks, first, placeholders, run, stmt } from "../db";
import { HttpError } from "../errors";
import { randomId } from "../crypto";
import { hasPermission, loadRepoById, repoPermission } from "../repos";
import type { RepoRow, Viewer } 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;
/** Unreferenced blobs survive this long, so an upload racing the sweeper never loses its data. */
const ORPHAN_GRACE_MS = 60 * 60 * 1000;
const SWEEP_BATCH = 100;

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

type ChainRepo = Pick<RepoRow, "id" | "fork_of_id" | "owner_id" | "visibility">;

/**
 * Whether readers of `repo` may see objects stored in the private `ancestor`: only for a private fork,
 * and only while the fork's owner (the account whose collaborators can read it) can still read the
 * ancestor, so a removed collaborator's fork stops seeing new objects.
 */
async function ownerCanRead(repo: ChainRepo, ancestorId: string): Promise<boolean> {
  if (repo.visibility === "public") return false;
  const ancestor = await loadRepoById(ancestorId);
  if (!ancestor) return false;
  if (ancestor.owner_id === repo.owner_id) return true;
  const owner: Viewer = { id: repo.owner_id, name: "", displayName: null, email: "", isAdmin: false };
  return hasPermission(await repoPermission(ancestor, owner), "read");
}

/**
 * The repo plus the ancestors whose objects its readers may see: public ancestors, or private ones the
 * fork's owner can still read when the fork is private too. Stops at the first ancestor that fails.
 */
async function repoChain(repo: ChainRepo): Promise<string[]> {
  const chain = [repo.id];
  let parent = repo.fork_of_id;
  while (parent && chain.length < MAX_FORK_DEPTH) {
    const row = await first<{ fork_of_id: string | null; visibility: string }>(
      "SELECT fork_of_id, visibility FROM repos WHERE id = ?",
      parent,
    );
    if (!row) break;
    if (row.visibility !== "public" && !(await ownerCanRead(repo, parent))) break;
    chain.push(parent);
    parent = row.fork_of_id;
  }
  return chain;
}

/** Finds which repo in the fork chain stores the object, preferring the repo itself. */
export async function findObject(repo: ChainRepo, 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: ChainRepo,
  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 }]));
}

/** Opens the single stored copy, once `repo` (or a readable ancestor) links the oid. */
export async function openObject(repo: ChainRepo, oid: string): Promise<R2ObjectBody | null> {
  if (!(await findObject(repo, oid))) return null;
  const blob = await first<{ r2_key: string }>("SELECT r2_key FROM lfs_blobs WHERE oid = ? AND refs > 0", oid);
  return blob ? env.BUCKET.get(blob.r2_key) : null;
}

type LinkRepo = Pick<RepoRow, "id" | "owner_id">;

/** `live` requires another link to exist already (copy-on-write); uploads link the copy they just stored. */
function linkStatement(repo: LinkRepo, oid: string, size: number, live = true): D1PreparedStatement {
  return stmt(
    `INSERT OR IGNORE INTO lfs_objects (repo_id, oid, owner_id, size, created_at)
     SELECT ?, oid, ?, size, ? FROM lfs_blobs WHERE oid = ? AND size = ? AND (refs > 0 OR ? = 0)`,
    repo.id,
    repo.owner_id,
    Date.now(),
    oid,
    size,
    live ? 1 : 0,
  );
}

/**
 * Links already-stored objects into `repo` without copying them. Only for oids the caller has shown
 * `repo` may read (found in its readable fork chain); the triggers bill the owner once per oid.
 */
export async function linkObjects(repo: LinkRepo, objects: Array<{ oid: string; size: number }>): Promise<void> {
  if (objects.length > 0) await batch(objects.map((object) => linkStatement(repo, object.oid, object.size)));
}

/** Copy-on-write fork: the fork links every object of its parent, sharing the stored copies. */
export async function linkForkObjects(fork: LinkRepo, parentId: string): Promise<void> {
  await run(
    `INSERT OR IGNORE INTO lfs_objects (repo_id, oid, owner_id, size, created_at)
     SELECT ?, oid, ?, size, ? FROM lfs_objects WHERE repo_id = ?`,
    fork.id,
    fork.owner_id,
    Date.now(),
    parentId,
  );
}

/**
 * Streams into R2 with a declared SHA-256 so R2 itself rejects content that does not match the oid.
 * Every upload writes a fresh key; registering the blob and the link happen in one transaction, so a
 * concurrent sweep either finishes first (and our key becomes the copy) or sees refs > 0. When another
 * copy already exists, the fresh one is deleted and the existing copy is shared.
 */
export async function storeObject(
  repo: LinkRepo,
  oid: string,
  size: number,
  body: ReadableStream<Uint8Array>,
): Promise<void> {
  const key = `lfs/blobs/${oid}/${randomId()}`;
  const sized = body.pipeThrough(new FixedLengthStream(size));
  try {
    await env.BUCKET.put(key, 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 now = Date.now();
  await batch([
    stmt(
      `INSERT INTO lfs_blobs (oid, size, r2_key, refs, orphaned_at, created_at) VALUES (?, ?, ?, 0, ?, ?)
       ON CONFLICT (oid) DO NOTHING`,
      oid,
      size,
      key,
      now,
      now,
    ),
    linkStatement(repo, oid, size, false),
  ]);
  const stored = await first<{ r2_key: string }>("SELECT r2_key FROM lfs_blobs WHERE oid = ?", oid);
  if (stored?.r2_key !== key) await env.BUCKET.delete(key);
}

/** Unlinks a repository's objects; the triggers release billing and the sweeper deletes orphaned copies. */
export async function purgeRepoLfs(repo: Pick<RepoRow, "id">): Promise<void> {
  await run("DELETE FROM lfs_objects WHERE repo_id = ?", repo.id);
}

/** Cron: deletes stored copies that no repository has linked for the grace period. */
export async function sweepOrphanedLfsBlobs(): Promise<void> {
  const orphans = await all<{ oid: string; r2_key: string }>(
    "DELETE FROM lfs_blobs WHERE refs = 0 AND orphaned_at < ? RETURNING oid, r2_key",
    Date.now() - ORPHAN_GRACE_MS,
  );
  for (const group of chunks(orphans, SWEEP_BATCH)) await env.BUCKET.delete(group.map((orphan) => orphan.r2_key));
}