import { DurableObject } from "cloudflare:workers";
import * as decoding from "lib0/decoding";
import * as encoding from "lib0/encoding";
import * as syncProtocol from "y-protocols/sync";
import * as Y from "yjs";
import { HttpError } from "../errors";
import { isSafeTreePath } from "../git/writer";
import { broadcastControl, broadcastFrame, sendControl } from "./room-send";
import { ReadonlyOpenLimiter } from "./room-throttle";
import {
  applyCommitted,
  conflicts,
  filesMap,
  getMeta,
  MAX_ROOM_FILES,
  MAX_TEXT_BYTES,
  compactUpdates,
  evictClean,
  guardMetadata,
  pendingChanges,
  readAtCommit,
  rebaseRoom,
  resolveConflict,
  setBase,
  setMeta,
  textOf,
} from "./room-files";
import { agentClientId, awarenessFrame, removalFrame } from "./room-awareness";
import { participants, presenceFrames, reauthorizeSocket, removalFor, sanitizeAwareness } from "./room-presence";
import type { Attachment, Participant, RoomJoin } from "./room-presence";
import {
  agentKey,
  checkEdits,
  editCursor,
  expireAgents,
  liveAgents,
  removeAgent,
  sortEdits,
  touchAgent,
} from "./room-agents";
import { CLOSE_REAUTH, EDITOR_WS_PROTOCOL, MESSAGE_AWARENESS, MESSAGE_SYNC } from "~/lib/editor-protocol";
import type { ClientControl, RoomControl } from "~/lib/editor-protocol";

const MAX_SOCKETS = 50;
const CLOSE_TOO_LARGE = 1009;
const HTTP_NOT_FOUND = 404;
const MAX_FRAME_BYTES = 1024 * 1024;
/** Control frames are small JSON (`{type, path}`); anything bigger is not a legitimate client. */
const MAX_CONTROL_BYTES = 8 * 1024;
/** Short: a revoked or downgraded collaborator's edits stop within this window, not a minute later. */
const RECHECK_MS = 15 * 1000;
const STATUS_DEBOUNCE_MS = 300;

/** Per-socket state; it survives hibernation, so it must stay small (attachments are capped at 2 KiB). */
export type { RoomJoin } from "./room-presence";
import type { ApplyResult, TextEdit } from "./room-agents";

/**
 * One room per (repo, branch). A single Y.Doc holds every open file as `file:<path>` plus a `files`
 * metadata map. Updates are appended to SQLite and compacted, so the room survives eviction and
 * hibernation. Sockets carry their editor token and are re-authorized every 15 seconds.
 */
export class EditorRoom extends DurableObject<Env> {
  private doc = new Y.Doc();
  private versions = new Map<string, number>();
  private seeding = new Map<string, Promise<void>>();
  private readonlyOpens = new ReadonlyOpenLimiter();
  private statusTimer: ReturnType<typeof setTimeout> | null = null;
  private sql: SqlStorage;

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    this.sql = ctx.storage.sql;
    ctx.blockConcurrencyWhile(async () => this.load());
  }

  private load(): void {
    this.sql.exec(`CREATE TABLE IF NOT EXISTS updates (id INTEGER PRIMARY KEY AUTOINCREMENT, data BLOB NOT NULL);
      CREATE TABLE IF NOT EXISTS bases (path TEXT PRIMARY KEY, text TEXT);
      CREATE TABLE IF NOT EXISTS upstream (path TEXT PRIMARY KEY, text TEXT);
      CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
      CREATE TABLE IF NOT EXISTS versions (path TEXT PRIMARY KEY, version INTEGER NOT NULL);
      CREATE TABLE IF NOT EXISTS agents (name TEXT PRIMARY KEY, user_name TEXT NOT NULL, path TEXT, clock INTEGER NOT NULL, expires_at INTEGER NOT NULL);`);
    for (const row of this.sql.exec<{ data: ArrayBuffer }>("SELECT data FROM updates ORDER BY id")) {
      Y.applyUpdate(this.doc, new Uint8Array(row.data), "load");
    }
    for (const row of this.sql.exec<{ path: string; version: number }>("SELECT path, version FROM versions")) {
      this.versions.set(row.path, row.version);
    }
    this.doc.on("update", (update: Uint8Array, origin: unknown) => this.onUpdate(update, origin));
    guardMetadata(this.doc);
    this.doc.on("afterTransaction", (transaction: Y.Transaction) => this.bumpVersions(transaction));
  }

  /** The Worker authorizes the upgrade and passes the resolved session in `x-editor-join`. */
  async fetch(request: Request): Promise<Response> {
    if (request.headers.get("upgrade") !== "websocket") return new Response("Expected websocket", { status: 426 });
    const join: RoomJoin = JSON.parse(request.headers.get("x-editor-join") ?? "null");
    if (!join) return new Response("Missing session", { status: 400 });
    if (this.ctx.getWebSockets().length >= MAX_SOCKETS) return new Response("Room is full", { status: 503 });
    await this.adopt(join.repoId, join.branch, join.head);
    const pair = new WebSocketPair();
    const socket = pair[1];
    this.ctx.acceptWebSocket(socket, [join.userId]);
    const attachment: Attachment = {
      token: join.token,
      userId: join.userId,
      name: join.name,
      kind: join.kind,
      writable: join.writable,
      checkedAt: Date.now(),
      expiresAt: join.expiresAt,
    };
    socket.serializeAttachment(attachment);
    this.control(socket, {
      type: "welcome",
      head: getMeta(this.sql, "head") ?? join.head,
      branch: join.branch,
      writable: join.writable,
      user: join.name,
      clientLimit: MAX_SOCKETS,
    });
    const encoder = encoding.createEncoder();
    encoding.writeVarUint(encoder, MESSAGE_SYNC);
    syncProtocol.writeSyncStep1(encoder, this.doc);
    socket.send(encoding.toUint8Array(encoder));
    for (const frame of this.presenceFrames(socket)) socket.send(frame);
    this.control(socket, this.statusMessage());
    await this.ensureAlarm();
    return new Response(null, {
      status: 101,
      webSocket: pair[0],
      headers: { "sec-websocket-protocol": EDITOR_WS_PROTOCOL },
    });
  }

  async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer): Promise<void> {
    let attachment = this.attachment(socket);
    if (!attachment) return socket.close(CLOSE_REAUTH, "Reconnect");
    if (Date.now() - attachment.checkedAt > RECHECK_MS) {
      attachment = await this.reauthorize(socket, attachment);
      if (!attachment) return;
    }
    if (typeof message === "string") return this.onControl(socket, attachment, message);
    if (message.byteLength > MAX_FRAME_BYTES)
      return this.control(socket, { type: "error", message: "Update too large" });
    const decoder = decoding.createDecoder(new Uint8Array(message));
    const type = decoding.readVarUint(decoder);
    if (type === MESSAGE_SYNC) {
      const subtype = decoding.peekVarUint(decoder);
      if (subtype !== syncProtocol.messageYjsSyncStep1) {
        if (!attachment.writable) return;
      }
      const encoder = encoding.createEncoder();
      encoding.writeVarUint(encoder, MESSAGE_SYNC);
      syncProtocol.readSyncMessage(decoder, encoder, this.doc, socket);
      if (encoding.length(encoder) > 1) socket.send(encoding.toUint8Array(encoder));
    } else if (type === MESSAGE_AWARENESS) {
      this.onAwareness(socket, attachment, decoding.readVarUint8Array(decoder));
    }
  }

  async webSocketClose(socket: WebSocket): Promise<void> {
    this.dropPresence(socket);
    socket.close();
  }

  async webSocketError(socket: WebSocket): Promise<void> {
    this.dropPresence(socket);
  }

  private async onControl(socket: WebSocket, attachment: Attachment, raw: string): Promise<void> {
    if (raw.length > MAX_CONTROL_BYTES) return this.control(socket, { type: "error", message: "Message too large" });
    let message: ClientControl;
    try {
      message = JSON.parse(raw);
    } catch {
      return;
    }
    if (message.type === "ping") return this.control(socket, { type: "pong" });
    if (typeof message.path !== "string" || !isSafeTreePath(message.path)) {
      return this.control(socket, { type: "error", message: "Invalid path" });
    }
    if (message.type === "open") {
      if (!attachment.writable && !this.readonlyOpens.allow(socket))
        return this.control(socket, { type: "error", path: message.path, message: "Opening files too quickly" });
      await this.seed(message.path, attachment.writable).catch((error: unknown) =>
        this.control(socket, {
          type: "error",
          path: message.path,
          message: error instanceof Error ? error.message : "Couldn't open the file",
        }),
      );
      return;
    }
    if (!attachment.writable) return this.control(socket, { type: "error", message: "This session is read-only" });
    const files = filesMap(this.doc);
    if (message.type === "create") {
      if (files.size >= MAX_ROOM_FILES) return this.control(socket, { type: "error", message: "Too many open files" });
      const existing = files.get(message.path);
      if (existing?.state === "deleted") files.set(message.path, { ...existing, state: "seeded" });
      else if (!existing) {
        await this.seed(message.path).catch(() => undefined);
        if (!files.get(message.path)) files.set(message.path, { state: "created", kind: "text", size: 0 });
      }
    } else if (message.type === "delete") {
      const existing = files.get(message.path);
      if (existing?.state === "created") {
        this.doc.transact(() => {
          files.delete(message.path);
          const text = textOf(this.doc, message.path);
          text.delete(0, text.length);
        }, socket);
      } else if (existing) files.set(message.path, { ...existing, state: "deleted" });
    } else if (message.type === "resolve") {
      resolveConflict(this.doc, this.sql, message.path);
      this.scheduleStatus();
    }
  }

  private onAwareness(socket: WebSocket, attachment: Attachment, update: Uint8Array): void {
    const relayed = sanitizeAwareness(attachment, update);
    if (!relayed) return;
    socket.serializeAttachment(relayed.attachment);
    this.broadcast(relayed.frame, socket);
  }

  private presenceFrames(except: WebSocket): Uint8Array[] {
    const others = this.ctx.getWebSockets().filter((socket) => socket !== except);
    return presenceFrames(
      others.map((socket) => this.attachment(socket)),
      liveAgents(this.sql),
    );
  }

  private dropPresence(socket: WebSocket): void {
    const frame = removalFor(this.attachment(socket));
    if (frame) this.broadcast(frame, socket);
  }

  private onUpdate(update: Uint8Array, origin: unknown): void {
    if (origin === "load") return;
    this.sql.exec("INSERT INTO updates (data) VALUES (?)", update);
    const encoder = encoding.createEncoder();
    encoding.writeVarUint(encoder, MESSAGE_SYNC);
    syncProtocol.writeUpdate(encoder, update);
    this.broadcast(encoding.toUint8Array(encoder), origin instanceof WebSocket ? origin : null);
    this.scheduleStatus();
  }

  private bumpVersions(transaction: Y.Transaction): void {
    if (transaction.origin === "load") return;
    const oversized: Array<{ socket: WebSocket; path: string; text: Y.Text }> = [];
    queueMicrotask(() => {
      for (const { socket, path, text } of oversized) {
        this.doc.transact(() => text.delete(MAX_TEXT_BYTES, text.length - MAX_TEXT_BYTES), "limit");
        socket.close(CLOSE_TOO_LARGE, `${path} is over the browser editing limit`);
      }
    });
    for (const [name, type] of this.doc.share) {
      if (!name.startsWith("file:") || !transaction.changed.has(type)) continue;
      const path = name.slice("file:".length);
      const text = type as unknown as Y.Text;
      if (transaction.origin instanceof WebSocket && text.length > MAX_TEXT_BYTES) {
        oversized.push({ socket: transaction.origin, path, text });
      }
      const version = (this.versions.get(path) ?? 0) + 1;
      this.versions.set(path, version);
      this.sql.exec(
        "INSERT INTO versions (path, version) VALUES (?, ?) ON CONFLICT(path) DO UPDATE SET version = excluded.version",
        path,
        version,
      );
    }
  }

  /** Loads a file into the shared doc once; only writers may evict clean files to make room. */
  private async seed(path: string, mayEvict = true): Promise<void> {
    if (filesMap(this.doc).has(path)) return;
    const pending = this.seeding.get(path);
    if (pending) return pending;
    const task = (async () => {
      if (mayEvict && filesMap(this.doc).size >= MAX_ROOM_FILES) evictClean(this.doc, this.sql, this.openPaths());
      if (filesMap(this.doc).size >= MAX_ROOM_FILES) throw new HttpError(429, "Too many open files in this room");
      const repoId = getMeta(this.sql, "repoId");
      const head = getMeta(this.sql, "head");
      if (!repoId || !head) throw new HttpError(409, "Room not initialized");
      const file = (await readAtCommit(repoId, head, [path])).get(path);
      if (!file) throw new HttpError(404, `${path} doesn't exist on this branch`);
      if (filesMap(this.doc).has(path)) return;
      this.doc.transact(() => {
        setBase(this.sql, path, file.text);
        if (file.text !== null) textOf(this.doc, path).insert(0, file.text);
        filesMap(this.doc).set(path, { state: "seeded", kind: file.kind, size: file.size });
      }, "seed");
    })();
    this.seeding.set(path, task);
    try {
      await task;
    } finally {
      this.seeding.delete(path);
    }
  }

  /** Binds a fresh room to its repo and branch, and follows the branch head when it moved. */
  private async adopt(repoId: string, branch: string, head: string): Promise<void> {
    if (!getMeta(this.sql, "repoId")) {
      setMeta(this.sql, "repoId", repoId);
      setMeta(this.sql, "branch", branch);
      setMeta(this.sql, "head", head);
      return;
    }
    if (getMeta(this.sql, "repoId") !== repoId || getMeta(this.sql, "branch") !== branch) {
      throw new HttpError(409, "Room key mismatch");
    }
    if (getMeta(this.sql, "head") !== head) {
      await rebaseRoom(this.doc, this.sql, repoId, head);
      this.scheduleStatus();
    }
  }

  /** Called by the commit endpoint: what to commit, against which head, and what blocks it. */
  async pending(repoId: string, branch: string, head: string) {
    await this.adopt(repoId, branch, head);
    return {
      head: getMeta(this.sql, "head") ?? head,
      changes: pendingChanges(this.doc, this.sql),
      conflicts: conflicts(this.doc),
    };
  }

  async committed(input: {
    sha: string;
    branch: string;
    pullUrl: string | null;
    by: string;
    changes: Array<{ path: string; content: string | null }>;
  }): Promise<void> {
    const sameBranch = input.branch === getMeta(this.sql, "branch");
    applyCommitted(this.doc, this.sql, input.changes, sameBranch);
    if (sameBranch) setMeta(this.sql, "head", input.sha);
    this.broadcastControl({
      type: "committed",
      sha: input.sha,
      branch: input.branch,
      pullUrl: input.pullUrl,
      by: input.by,
    });
    this.scheduleStatus();
  }

  async snapshot(repoId: string, branch: string, head: string, path: string) {
    await this.adopt(repoId, branch, head);
    await this.seed(path);
    const meta = filesMap(this.doc).get(path) ?? null;
    return {
      path,
      text: meta && meta.kind !== "binary" && meta.kind !== "too-large" ? textOf(this.doc, path).toString() : null,
      kind: meta?.kind ?? null,
      state: meta?.state ?? null,
      version: this.versions.get(path) ?? 0,
      participants: this.participants(),
    };
  }

  /** Applies agent edits as one Yjs transaction; offsets are UTF-16 against the text at `baseVersion`. */
  async applyText(
    target: { repoId: string; branch: string; head: string; path: string },
    edits: TextEdit[],
    origin: { agent: string; user: string; userId: string },
    baseVersion?: number,
  ): Promise<ApplyResult> {
    await this.adopt(target.repoId, target.branch, target.head);
    const { path } = target;
    if (!filesMap(this.doc).has(path)) {
      const missing = await this.seed(path).then(
        () => false,
        (error: unknown) => {
          if (error instanceof HttpError && error.status === HTTP_NOT_FOUND) return true;
          throw error;
        },
      );
      if (missing) filesMap(this.doc).set(path, { state: "created", kind: "text", size: 0 });
    }
    const meta = filesMap(this.doc).get(path);
    const text = textOf(this.doc, path);
    const version = this.versions.get(path) ?? 0;
    if (!meta || meta.kind !== "text" || meta.state === "deleted") {
      return { ok: false, error: `${path} isn't an editable text file`, version, text: text.toString() };
    }
    if (baseVersion !== undefined && baseVersion !== version) {
      return { ok: false, error: "stale", version, text: text.toString() };
    }
    const sorted = sortEdits(edits);
    const invalid = checkEdits(sorted, text.length, MAX_TEXT_BYTES);
    if (invalid) return { ok: false, error: invalid, version, text: text.toString() };
    this.doc.transact(() => {
      for (const edit of sorted) {
        if (edit.to > edit.from) text.delete(edit.from, edit.to - edit.from);
        if (edit.insert) text.insert(edit.from, edit.insert);
      }
    }, `agent:${origin.agent}`);
    const cursor = editCursor(sorted);
    this.broadcast(
      awarenessFrame(touchAgent(this.sql, agentKey(origin.userId, origin.agent), origin.user, path, cursor)),
      null,
    );
    await this.ensureAlarm();
    return { ok: true, version: this.versions.get(path) ?? version, text: text.toString() };
  }

  /** Removes an agent's presence; agents are keyed per user, so a caller can only remove its own. */
  async leave(userId: string, agent: string): Promise<void> {
    const key = agentKey(userId, agent);
    const clock = removeAgent(this.sql, key);
    if (clock !== null) this.broadcast(removalFrame(agentClientId(key), clock), null);
  }

  /** Paths someone currently has a cursor in; eviction leaves those alone. */
  private openPaths(): Set<string> {
    return new Set(
      this.participants()
        .map((participant) => participant.path)
        .filter((path): path is string => !!path),
    );
  }

  private participants(): Participant[] {
    return participants(
      this.ctx.getWebSockets().map((socket) => this.attachment(socket)),
      liveAgents(this.sql),
    );
  }

  // --- Authorization, alarms and fan-out -----------------------------------------------------------

  private reauthorize(socket: WebSocket, attachment: Attachment): Promise<Attachment | null> {
    return reauthorizeSocket(socket, attachment, getMeta(this.sql, "repoId"), getMeta(this.sql, "branch"));
  }

  async alarm(): Promise<void> {
    const sockets = this.ctx.getWebSockets();
    await Promise.all(
      sockets.map(async (socket) => {
        const attachment = this.attachment(socket);
        if (attachment) await this.reauthorize(socket, attachment);
      }),
    );
    for (const agent of expireAgents(this.sql))
      this.broadcast(removalFrame(agentClientId(agent.name), agent.clock), null);
    compactUpdates(this.ctx.storage, this.doc);
    await this.ensureAlarm();
  }

  private async ensureAlarm(): Promise<void> {
    const agents = this.sql.exec<{ count: number }>("SELECT COUNT(*) AS count FROM agents").one().count;
    if (this.ctx.getWebSockets().length === 0 && agents === 0) return;
    if ((await this.ctx.storage.getAlarm()) === null) await this.ctx.storage.setAlarm(Date.now() + RECHECK_MS);
  }

  private scheduleStatus(): void {
    if (this.statusTimer) return;
    this.statusTimer = setTimeout(() => {
      this.statusTimer = null;
      this.broadcastControl(this.statusMessage());
    }, STATUS_DEBOUNCE_MS);
  }

  private statusMessage(): RoomControl {
    return {
      type: "status",
      head: getMeta(this.sql, "head") ?? "",
      changes: pendingChanges(this.doc, this.sql).map(({ path, status }) => ({ path, status })),
      conflicts: conflicts(this.doc),
    };
  }

  private attachment(socket: WebSocket): Attachment | null {
    return socket.deserializeAttachment() as Attachment | null;
  }

  private control(socket: WebSocket, message: RoomControl): void {
    sendControl(socket, message);
  }

  private broadcastControl(message: RoomControl): void {
    broadcastControl(this.ctx.getWebSockets(), message);
  }

  private broadcast(frame: Uint8Array, except: WebSocket | null): void {
    broadcastFrame(this.ctx.getWebSockets(), frame, except);
  }
}