import type { AdapterContext, ArtifactByteRange, ArtifactStore } from "@rakazo/adapter-kit";
import { ArtifactTooLargeError } from "@rakazo/adapter-kit";
import { artifactMaxBytes, storeArtifactStream } from "@rakazo/adapters";
import type { Actor } from "@rakazo/contracts";
import { attachmentKindOf, isAllowedAttachmentMimeType } from "@rakazo/contracts";
import type { PrismaClient } from "@rakazo/db";
import { createGroupRepos, createRepos, IsolationError, requireMembership } from "@rakazo/db";
import type { Context, Hono } from "hono";
import { cancelBody } from "./http-body.js";

export const ARTIFACT_HTTP_PATH = "/api/artifacts";

/**
 * Node's default 5-minute requestTimeout cuts off a 500 MiB video upload on any uplink slower than
 * about 14 Mbit/s; 30 minutes fits it at about 2.5 Mbit/s. headersTimeout keeps its default, and
 * the per-upload stall timeout stops a slow-drip client from holding a connection that long.
 */
export const API_SERVER_OPTIONS = { requestTimeout: 30 * 60 * 1000 };
/** An upload that sends no bytes for this long is aborted and its partial file deleted. */
export const ARTIFACT_UPLOAD_STALL_TIMEOUT_MS = 60_000;

export class UploadStalledError extends Error {
  constructor() {
    super("Upload stalled");
    this.name = "UploadStalledError";
  }
}

/**
 * Yields the body's chunks and throws UploadStalledError when the next chunk takes longer than
 * stallMs. Only time spent waiting for the client counts, not time spent writing to disk.
 */
export async function* stallGuardedChunks(
  body: ReadableStream<Uint8Array>,
  stallMs: number,
): AsyncGenerator<Uint8Array> {
  const reader = body.getReader();
  try {
    for (;;) {
      let timer: ReturnType<typeof setTimeout> | undefined;
      const stalled = new Promise<never>((_, reject) => {
        timer = setTimeout(() => reject(new UploadStalledError()), stallMs);
      });
      try {
        const { done, value } = await Promise.race([reader.read(), stalled]);
        if (done) return;
        yield value;
      } finally {
        clearTimeout(timer);
      }
    }
  } finally {
    // Never wait on the client here: a stalled or hostile body may never confirm the cancel.
    void reader.cancel().catch(() => undefined);
    reader.releaseLock();
  }
}

export type ArtifactHttpDeps = {
  prisma: PrismaClient;
  artifacts: ArtifactStore;
  /** The signed-in user (cookie or bearer session), or null. */
  sessionUserId: (request: Request) => Promise<string | null>;
  /** Same rule as CORS. Raw uploads can be CORS "simple" requests, so the origin is checked. */
  trustedOrigin: (origin: string) => boolean;
  /** Overrides ARTIFACT_UPLOAD_STALL_TIMEOUT_MS (tests). */
  uploadStallTimeoutMs?: number;
};

/** Generated ids only; anything else is not an artifact. */
const ARTIFACT_ID = /^[A-Za-z0-9_-]{1,64}$/;
const MAX_RANGE_DIGITS = 15;

/**
 * Parses a single `bytes=` range against the stored size. Returns null when the header should
 * be ignored (absent or another unit) and "unsatisfiable" for multiple ranges or offsets
 * outside the file, which the caller answers with 416.
 */
export function parseByteRange(
  header: string | undefined,
  size: number,
): ArtifactByteRange | "unsatisfiable" | null {
  if (!header) return null;
  const value = header.trim();
  if (!value.toLowerCase().startsWith("bytes=")) return null;
  const spec = value.slice("bytes=".length).trim();
  const match = new RegExp(`^(\\d{0,${MAX_RANGE_DIGITS}})-(\\d{0,${MAX_RANGE_DIGITS}})$`).exec(
    spec,
  );
  if (!match || size <= 0) return "unsatisfiable";
  const [, first = "", last = ""] = match;
  if (first === "" && last === "") return "unsatisfiable";
  if (first === "") {
    const suffix = Number(last);
    if (suffix <= 0) return "unsatisfiable";
    return { start: Math.max(0, size - suffix), end: size - 1 };
  }
  const start = Number(first);
  const end = last === "" ? size - 1 : Number(last);
  if (start >= size || end < start) return "unsatisfiable";
  return { start, end: Math.min(end, size - 1) };
}

/** `inline` or `attachment` with an ASCII fallback and the exact UTF-8 name (RFC 6266). */
export function contentDisposition(name: string, inline: boolean): string {
  const cleaned =
    Array.from(name, (char) => {
      const code = char.codePointAt(0) ?? 0;
      return code < 0x20 || code === 0x7f || char === "/" || char === "\\" ? "_" : char;
    })
      .join("")
      .trim() || "download";
  const ascii = cleaned.replace(/[^ -~]/g, "_").replace(/["%;]/g, "_");
  const encoded = encodeURIComponent(cleaned).replace(
    /['()*]/g,
    (char) => `%${char.charCodeAt(0).toString(16).toUpperCase()}`,
  );
  return `${inline ? "inline" : "attachment"}; filename="${ascii}"; filename*=UTF-8''${encoded}`;
}

function adapterContext(actor: Actor, operationId: string, signal: AbortSignal): AdapterContext {
  return {
    operationId,
    traceId: operationId,
    spaceId: actor.spaceId,
    userId: actor.userId,
    signal,
  };
}

function notFound(c: Context) {
  return c.json({ error: "Not found" }, 404);
}

function artifactDto(row: {
  id: string;
  botId: string | null;
  groupId: string | null;
  runId: string | null;
  name: string;
  mimeType: string;
  size: number;
  createdAt: Date;
}) {
  return {
    id: row.id,
    botId: row.botId,
    groupId: row.groupId,
    runId: row.runId,
    name: row.name,
    mimeType: row.mimeType,
    size: row.size,
    createdAt: row.createdAt.toISOString(),
  };
}

/**
 * Streams artifacts over HTTP so media can play with seeking and large files never travel as
 * base64 RPC payloads. Authorization matches `artifacts.get` and `artifacts.create`.
 */
export function mountArtifactHttpRoutes(app: Hono, deps: ArtifactHttpDeps) {
  const repos = createRepos(deps.prisma);
  const groupRepos = createGroupRepos(deps.prisma);

  /** Same rule as artifacts.get: the owner's artifact of an active bot thread or group. */
  async function readableArtifact(
    userId: string,
    requestedSpaceId: string | undefined,
    id: string,
  ) {
    if (!ARTIFACT_ID.test(id)) return null;
    const row = await deps.prisma.artifact.findUnique({ where: { id } });
    if (!row) return null;
    // <video> and <img> cannot send the space header, so the artifact's space is the default;
    // membership is still required and an explicit header must name the same space.
    const actor = await requireMembership(
      deps.prisma,
      userId,
      requestedSpaceId ?? row.spaceId,
    ).catch(() => null);
    if (!actor || actor.spaceId !== row.spaceId || actor.userId !== row.userId) return null;
    try {
      if (row.groupId) await groupRepos.getGroupTarget(actor, row.groupId);
      else if (row.botId) await repos.getBot(actor, row.botId);
      else return null;
    } catch (error) {
      if (error instanceof IsolationError) return null;
      throw error;
    }
    return { row, actor };
  }

  app.get(`${ARTIFACT_HTTP_PATH}/:id`, async (c) => {
    const userId = await deps.sessionUserId(c.req.raw);
    if (!userId) return c.json({ error: "Unauthorized" }, 401);
    const found = await readableArtifact(
      userId,
      c.req.header("x-rakazo-space-id"),
      c.req.param("id"),
    );
    if (!found) return notFound(c);
    const { row, actor } = found;
    const context = adapterContext(actor, `artifact-http-get:${row.id}`, c.req.raw.signal);
    let size: number;
    try {
      ({ size } = await deps.artifacts.stat(row.storageKey, context));
    } catch {
      return notFound(c);
    }

    const mimeType = isAllowedAttachmentMimeType(row.mimeType)
      ? row.mimeType
      : "application/octet-stream";
    const inline = attachmentKindOf(mimeType) !== "file" && c.req.query("download") === undefined;
    const etag = /^[A-Za-z0-9]{16,128}$/.test(row.hash) ? `"${row.hash}"` : undefined;
    const headers = new Headers({
      "accept-ranges": "bytes",
      // Revalidate every time so a revoked thread stops serving from the browser cache.
      "cache-control": "private, no-cache",
      "content-disposition": contentDisposition(row.name, inline),
      "content-type": mimeType,
      "cross-origin-resource-policy": "same-origin",
      "x-content-type-options": "nosniff",
    });
    if (etag) headers.set("etag", etag);
    if (etag && c.req.header("if-none-match") === etag) {
      return new Response(null, { status: 304, headers });
    }

    const ifRange = c.req.header("if-range");
    const range =
      ifRange === undefined || ifRange === etag
        ? parseByteRange(c.req.header("range"), size)
        : null;
    if (range === "unsatisfiable") {
      headers.set("content-range", `bytes */${size}`);
      return new Response(null, { status: 416, headers });
    }
    const status = range ? 206 : 200;
    headers.set("content-length", String(range ? range.end - range.start + 1 : size));
    if (range) headers.set("content-range", `bytes ${range.start}-${range.end}/${size}`);
    // Hono answers HEAD from this handler and drops the body; never open a stream for it.
    if (c.req.method === "HEAD") return new Response(null, { status, headers });

    let body: ReadableStream<Uint8Array>;
    try {
      body = await deps.artifacts.createReadStream(row.storageKey, context, range ?? undefined);
    } catch {
      return notFound(c);
    }
    return new Response(body, { status, headers });
  });

  app.post(ARTIFACT_HTTP_PATH, async (c) => {
    const request = c.req.raw;
    const reject = (status: 400 | 401 | 403 | 404 | 408 | 413 | 415, error: string) => {
      cancelBody(request.body);
      return c.json({ error }, status);
    };
    const origin = c.req.header("origin");
    if (origin && !deps.trustedOrigin(origin)) return reject(403, "Forbidden");
    const userId = await deps.sessionUserId(request);
    if (!userId) return reject(401, "Unauthorized");
    const actor = await requireMembership(
      deps.prisma,
      userId,
      c.req.header("x-rakazo-space-id"),
    ).catch(() => null);
    if (!actor) return reject(401, "Unauthorized");

    const botId = c.req.query("botId") || undefined;
    const groupId = c.req.query("groupId") || undefined;
    const name = c.req.query("name") ?? "";
    if (Boolean(botId) === Boolean(groupId)) {
      return reject(400, "Provide exactly one of botId or groupId");
    }
    if (name.length < 1 || name.length > 255) return reject(400, "Invalid file name");
    const mimeType = (c.req.header("content-type") ?? "").split(";", 1)[0]!.trim().toLowerCase();
    if (!isAllowedAttachmentMimeType(mimeType)) {
      return reject(415, "Unsupported attachment type");
    }
    const maxBytes = artifactMaxBytes(mimeType);
    const declared = c.req.header("content-length");
    if (declared !== undefined) {
      if (!/^\d{1,16}$/.test(declared)) return reject(400, "Invalid content length");
      if (Number(declared) > maxBytes) return reject(413, "File is too large.");
      if (Number(declared) === 0) return reject(400, "File is empty.");
    }
    if (!request.body) return reject(400, "File is empty.");

    // Same targets as artifacts.create: an own active bot, or an active group of own bots.
    let ownerBotId: string | undefined;
    try {
      ownerBotId = botId
        ? (await repos.getBot(actor, botId)).id
        : (await groupRepos.getGroupTarget(actor, groupId!)).members[0]?.bot.id;
    } catch (error) {
      if (!(error instanceof IsolationError)) throw error;
    }
    if (!ownerBotId) return reject(404, "Not found");

    let row: Awaited<ReturnType<typeof storeArtifactStream>>;
    try {
      row = await storeArtifactStream(deps, {
        spaceId: actor.spaceId,
        userId: actor.userId,
        botId: ownerBotId,
        groupId: groupId ?? null,
        runId: null,
        name,
        mimeType,
        stream: stallGuardedChunks(
          request.body,
          deps.uploadStallTimeoutMs ?? ARTIFACT_UPLOAD_STALL_TIMEOUT_MS,
        ),
        maxBytes,
        signal: request.signal,
      });
    } catch (error) {
      if (error instanceof ArtifactTooLargeError) return reject(413, "File is too large.");
      if (error instanceof UploadStalledError) return reject(408, "Upload stalled.");
      throw error;
    }
    if (row.size === 0) {
      await deps.prisma.artifact.delete({ where: { id: row.id } });
      await deps.artifacts
        .remove(
          row.storageKey,
          adapterContext(actor, `artifact-http-post:${row.id}`, request.signal),
        )
        .catch(() => undefined);
      return c.json({ error: "File is empty." }, 400);
    }
    return c.json(artifactDto(row), 201);
  });
}