/**
 * RoomManager (TASK-0091 R2) — one authoritative room per resource.
 *
 * A room owns: its `Y.Doc` (page/database rooms only), the peer set,
 * the debounced persist state, and the hydrate-on-write buffer
 * (doc §4.2). The per-room update log (R5) lives in the server, keyed
 * by the room key.
 *
 * Lifecycle (doc §3 / R2):
 *  - cold room: hydrate from the {@link DocStore} BEFORE publishing the
 *    room (concurrent joins share the one hydration);
 *  - update while hydrating: buffered (max 5 s / 5,000 updates), applied
 *    to the hydrated doc in order (doc §4.2 hydrate-on-write race);
 *  - leave: 0 peers + dirty → the flush is awaited before eviction; a
 *    room never evicts with unsaved updates (GC after 5 min warm);
 *  - SIGTERM: `drainAll()` flushes every dirty room (≤ 10 s budget, R6).
 *
 * The persistence seam is {@link DocStore} (note 17 — the Prisma
 * implementation lands in TASK-0092 via the api's internal flush route;
 * tests + no-DB dev use {@link InMemoryDocStore}). The Yjs codec is the
 * shared `@factory/collaboration` codec (the same blocksToYDoc/
 * yDocToBlockList the client canvas uses — one codec, both sides).
 */

import * as Y from "yjs";
import {
  type BlockLike,
  blocksToYDoc,
  yDocToBlockList,
  applyBlockOperationsToYDoc,
  diffBlockSets,
} from "@factory/collaboration/codec";
import type { BlockOperation } from "@factory/shared";
import { presenceColor } from "@factory/shared/presence";
import type { RoomPeer } from "@factory/shared/realtime/frames";

/** The block id an op refers to (insert: the inserted block's id; the
 *  rest: the target's id) — the 3-way rebase's "who touched what" basis.
 *  An insert without an explicit id is a server-minted block: no external
 *  (REST) op can ever target it, so it never collides and is filtered out
 *  here (`undefined`). */
function opIdOf(op: BlockOperation): string | undefined {
  return op.op === "insert" ? op.block.id : op.id;
}

// --- the persistence seam (note 17) ---------------------------------------------

/**
 * The doc persistence seam. Production: Postgres (TASK-0092, the
 * {@link PrismaDocStore} with the `pages.version` CAS + flush-rebase).
 * Tests/dev: {@link InMemoryDocStore}.
 *
 * The seam returns the page's `pages.version` alongside the blocks,
 * atomically, because the flush-rebase (R2 of TASK-0092) needs the version
 * the doc was hydrated at (its CAS base) and the version a persist reached
 * (the new CAS base) in one consistent read — a two-call `hydrate()` +
 * `getVersion()` pair could race a concurrent REST write in between.
 */
export interface HydrateResult {
  /** Position-ordered blocks; `null` = the page has no blocks. */
  readonly blocks: readonly BlockLike[] | null;
  /** The `pages.version` at hydration time — the flush's CAS base. */
  readonly version: number;
}

export interface PersistResult {
  /** The new `pages.version` (the `doc.persisted` frame carries it). */
  readonly version: number;
  /**
   * Only set on a version conflict (the doc's CAS base was stale — a
   * concurrent non-Yjs writer bumped `pages.version` since the doc was
   * hydrated/last-flushed). `externalBlocks` is the store's current state
   * (the writer's result — the REST batch that ran concurrently); the room
   * rebases its CRDT on it (3-way merge, the doc §4.3 "flush wins" rule) and
   * retries the persist once.
   */
  readonly conflict?: {
    readonly externalBlocks: readonly BlockLike[];
  };
}

export interface DocStore {
  /** Hydrate blocks + the page version for a page (see {@link HydrateResult}). */
  hydrate(pageId: string): Promise<HydrateResult>;
  /**
   * Persist the page's full block set under the `pages.version` CAS
   * (soft-deletes are the store's concern). `baseVersion` is the version the
   * doc was hydrated at / last flushed to: if the current version differs,
   * the persist is a no-op and {@link PersistResult.conflict} carries the
   * store's external state (the concurrent writer) for the room to rebase on.
   * A successful persist returns the new version (no conflict).
   */
  persist(
    pageId: string,
    blocks: readonly BlockLike[],
    baseVersion: number,
  ): Promise<PersistResult>;
  /** Whether the store backend is reachable (`/readyz`). */
  ping(): Promise<boolean>;
}

/** A trivial in-memory store (tests + no-DB dev mode). */
export class InMemoryDocStore implements DocStore {
  private readonly pages = new Map<string, BlockLike[]>();
  private readonly versions = new Map<string, number>();
  private pingOk = true;

  setPingOk(ok: boolean): void {
    this.pingOk = ok;
  }

  seed(pageId: string, blocks: readonly BlockLike[]): void {
    this.pages.set(pageId, blocks.map((b) => ({ ...b })));
    // A seed is a "the page exists at version N" — the version bumps only on
    // persist, so the seed's version is whatever was already recorded (the
    // rebase tests simulate a concurrent external writer by bumping it).
    if (!this.versions.has(pageId)) this.versions.set(pageId, 1);
  }

  /**
   * Simulate a concurrent NON-Yjs writer (a REST `POST /v1/blocks` batch)
   * against the store: replace the page's blocks and bump `pages.version`.
   * The rebase test drives the conflict this way. (Production: the API's
   * block service does exactly this on the `blocks` table.)
   */
  writeExternal(pageId: string, blocks: readonly BlockLike[]): number {
    this.pages.set(pageId, blocks.map((b) => ({ ...b })));
    const v = (this.versions.get(pageId) ?? 1) + 1;
    this.versions.set(pageId, v);
    return v;
  }

  /** The page's current `pages.version` (tests / the rebase assertion). */
  versionOf(pageId: string): number {
    return this.versions.get(pageId) ?? 1;
  }

  async hydrate(pageId: string): Promise<HydrateResult> {
    const b = this.pages.get(pageId);
    return {
      blocks: b === undefined ? null : b.map((x) => ({ ...x })),
      version: this.versions.get(pageId) ?? 1,
    };
  }

  async persist(
    pageId: string,
    blocks: readonly BlockLike[],
    baseVersion: number,
  ): Promise<PersistResult> {
    const current = this.versions.get(pageId) ?? 1;
    if (current !== baseVersion) {
      // CAS failure: a concurrent writer bumped the version. Read the store's
      // current state back (the external writer's result) for the rebase.
      const ext = this.pages.get(pageId);
      return { version: current, conflict: { externalBlocks: ext === undefined ? [] : ext.map((x) => ({ ...x })) } };
    }
    this.pages.set(pageId, blocks.map((b) => ({ ...b })));
    const v = current + 1;
    this.versions.set(pageId, v);
    return { version: v };
  }

  async ping(): Promise<boolean> {
    return this.pingOk;
  }
}

// --- rooms -----------------------------------------------------------------------

/** A connected peer in a room. */
export interface Peer {
  readonly connId: string;
  readonly userId: string;
  readonly name: string;
  readonly color: string;
  /** The socket's `canEdit` flag (defense in depth — re-checked per frame). */
  canEdit: boolean;
  /** The peer's room roster entry (the `hello.roomPeers` payload). */
  roster: RoomPeer;
}

/** The injectable clock + timers (deterministic flush/GC tests). */
export interface RoomClock {
  readonly now: () => number;
  readonly setTimeout: (fn: () => void, ms: number) => unknown;
  readonly clearTimeout: (h: unknown) => void;
}

export function realClock(): RoomClock {
  return {
    now: () => Date.now(),
    setTimeout: (fn, ms) => setTimeout(fn, ms),
    clearTimeout: (h) => clearTimeout(h as ReturnType<typeof setTimeout>),
  };
}

export interface RoomManagerOpts {
  readonly store: DocStore;
  readonly clock?: RoomClock;
  /** Debounce before a flush (1 s, doc §4.3). */
  readonly persistDebounceMs?: number;
  /** Hard cap on dirty time (10 s, doc §4.3). */
  readonly persistHardCapMs?: number;
  /** Empty-room GC delay (5 min warm room, R2). */
  readonly roomGcMs?: number;
  /** Hydrate buffer cap (doc §4.2). */
  readonly hydrateBufferLimit?: number;
  /** Reports the room's current max logged update seq (the server's
   *  update log) — captured at the flush snapshot so the post-persist
   *  trim keeps exactly the post-flush entries (R5: "truncated after a
   *  successful Postgres persistence"). The seq domain is the LOG seq
   *  (per-room, restarts after eviction), NOT the page version. */
  readonly flushSeqProvider?: (roomKey: string) => number;
}

/** Events the room manager reports (metrics/frames). */
export interface RoomManagerEvents {
  /** A doc persisted (the server broadcasts `doc.persisted` + trims the log
   *  to post-flush entries, R5). `logSeqAtFlush` is the room's max logged
   *  update seq captured at the flush snapshot — the trim basis.
   *  `blockDelta` is the minimal op diff of this flush vs the last persisted
   *  state (the `page.updated` `blockDelta` additive field, TASK-0092 R2/R6). */
  onPersisted: (
    roomKey: string,
    resourceId: string,
    version: number,
    logSeqAtFlush: number,
    blockDelta: readonly BlockOperation[],
  ) => void;
  /** A room rebased its doc on a concurrent external (REST) write after a
   *  version conflict (the flush-rebase path, TASK-0092 R2). */
  onRebased?: (roomKey: string, resourceId: string, appliedExternalOps: number) => void;
  /** A room was evicted (GC / drain). */
  onRoomEvicted: (roomKey: string) => void;
}

/**
 * One room. (Exported for the unit tests; the server works through
 * {@link RoomManager}.)
 */
export class Room {
  readonly key: string;
  readonly kind: "page" | "database";
  readonly resourceId: string;

  /** The authoritative doc (page/database rooms). Mutable: hydration
   *  swaps it (a destroyed doc is never re-used — the yjs rule). */
  doc: Y.Doc;

  readonly peers = new Map<string, Peer>();

  /** Inbound updates buffered while a re-hydration is in flight
   *  (doc §4.2 hydrate-on-write race; the cold path hydrates before the
   *  room is published, so peers never see a half-built doc). */
  private hydrateBuffer: Uint8Array[] = [];
  private hydrating = false;

  private dirtySince: number | null = null;
  private flushTimer: unknown = null;
  private flushing: Promise<void> | null = null;
  /** The `pages.version` the doc was hydrated at / last flushed to — the
   *  persist CAS base (the flush-rebase guard, TASK-0092 R2). Also the
   *  per-update `pageVersion` the server records in the update log. */
  lastVersion = 0;
  /** The block set the doc last persisted as — the 3-way merge BASE for the
   *  flush-rebase + the `blockDelta` diff baseline. */
  private lastPersistedBlocks: readonly BlockLike[] = [];

  private gcTimer: unknown = null;
  private _evicted = false;

  constructor(
    key: string,
    kind: "page" | "database",
    resourceId: string,
    private readonly opts: Required<RoomManagerOpts>,
    private readonly events: RoomManagerEvents,
    seed: HydrateResult,
  ) {
    this.key = key;
    this.kind = kind;
    this.resourceId = resourceId;
    this.doc = this.buildDoc(seed.blocks);
    this.lastVersion = seed.version;
    // A hydrated page persists its baseline (so the first dirty flush diffs
    // against the real starting state, not an empty one). An empty page has
    // an empty baseline.
    this.lastPersistedBlocks = seed.blocks === null ? [] : seed.blocks.map((b) => ({ ...b }));
  }

  get isDirty(): boolean {
    return this.dirtySince !== null;
  }

  get isEmpty(): boolean {
    return this.peers.size === 0;
  }

  get isEvicted(): boolean {
    return this._evicted;
  }

  /** Number of buffered (not yet applied) hydration updates (tests). */
  get bufferedUpdates(): number {
    return this.hydrateBuffer.length;
  }

  private buildDoc(seed: readonly BlockLike[] | null): Y.Doc {
    const doc =
      seed === null || seed.length === 0
        ? new Y.Doc()
        : blocksToYDoc(seed, { pageId: this.kind === "page" ? this.resourceId : undefined });
    if (this.kind === "page") {
      doc.getMap("yMeta").set("pageId", this.resourceId);
    }
    return doc;
  }

  /**
   * Apply a remote (client) update. While a re-hydration is in flight,
   * the update is BUFFERED (doc §4.2) — applied in order once the doc is
   * ready. Returns `true` when applied now, `false` when buffered.
   */
  applyUpdate(bytes: Uint8Array): boolean {
    if (this.hydrating) {
      if (this.hydrateBuffer.length >= this.opts.hydrateBufferLimit) {
        throw new Error("hydrate buffer overflow");
      }
      this.hydrateBuffer.push(bytes);
      return false;
    }
    Y.applyUpdate(this.doc, bytes, "remote");
    this.markDirty();
    return true;
  }

  /** Encode the state delta for a client's state vector (R5 resume). */
  stateDelta(sinceVectorBytes: Uint8Array | null): Uint8Array {
    if (sinceVectorBytes === null || sinceVectorBytes.byteLength === 0) {
      return Y.encodeStateAsUpdate(this.doc);
    }
    // `encodeStateAsUpdate` takes the RAW state-vector bytes (not the decoded map).
    return Y.encodeStateAsUpdate(this.doc, sinceVectorBytes);
  }

  private markDirty(): void {
    if (this.dirtySince === null) {
      this.dirtySince = this.opts.clock.now();
    }
    this.scheduleFlush();
  }

  private scheduleFlush(): void {
    if (this.flushTimer !== null) {
      this.opts.clock.clearTimeout(this.flushTimer);
    }
    const age = this.dirtySince === null ? 0 : this.opts.clock.now() - this.dirtySince;
    const delay = Math.max(
      0,
      Math.min(this.opts.persistDebounceMs, this.opts.persistHardCapMs - age),
    );
    this.flushTimer = this.opts.clock.setTimeout(() => {
      void this.flush();
    }, delay);
  }

  /**
   * Flush the current doc state (debounce tick, room GC, or SIGTERM drain).
   * Concurrent flushes coalesce into the one in flight.
   *
   * TASK-0092 R2 (the flush-rebase, doc §4.3 "flush wins"): the persist is a
   * `pages.version` CAS. If a concurrent non-Yjs writer (a REST
   * `POST /v1/blocks` batch) bumped the version since the doc was hydrated
   * / last flushed, the persist is a no-op and the store returns its
   * external state. We rebase the CRDT on it — a 3-way merge whose BASE is
   * the last persisted block set: the external (REST) ops are applied to the
   * doc, so on any block the ROOM also touched the doc's CRDT edit wins
   * (flush-wins) and on blocks only REST touched the REST change is absorbed.
   * The persist then retries ONCE at the new version. A second conflict
   * (two concurrent external writes inside one flush window) is outside the
   * single-retry contract: the room stays dirty with its user edits intact
   * (a full-set persist, so nothing is lost) and the next flush converges.
   */
  flush(): Promise<void> {
    if (this.flushing !== null) {
      return this.flushing;
    }
    const p = (async () => {
      if (this.flushTimer !== null) {
        this.opts.clock.clearTimeout(this.flushTimer);
        this.flushTimer = null;
      }
      // Capture the log watermark BEFORE the (async) persist: entries logged
      // up to here are covered by this flush's doc snapshot.
      const logSeqAtFlush = this.opts.flushSeqProvider?.(this.key) ?? 0;

      // OURS (pre-rebase) + the set of blocks the ROOM's user edits touched
      // (the flush-wins basis). Captured BEFORE the rebase loop so "room
      // touched X" reflects user edits only — never the blocks a rebase pass
      // absorbed (absorbed external blocks must stay absorbable by LATER
      // external writes, not be mistaken for user edits).
      let blocks = yDocToBlockList(this.doc);
      const roomTouched = new Set(
        diffBlockSets(this.lastPersistedBlocks, blocks).map(opIdOf),
      );
      // The external state the doc is already reconciled to (starts at the
      // last persisted state; each rebase pass advances it).
      let externalBase = this.lastPersistedBlocks;

      let result = await this.opts.store.persist(this.resourceId, blocks, this.lastVersion);
      let attempts = 1;

      // Version conflict → rebase onto the external (REST) state + retry ONCE
      // (the spec's "retries once"). The rebase is idempotent + convergent:
      // each pass applies only the external changes since `externalBase`, so
      // re-applying is a no-op and later external writes are never clobbered.
      while (result.conflict !== undefined) {
        const externalOps = diffBlockSets(externalBase, result.conflict.externalBlocks);
        const absorb = externalOps.filter((op) => !roomTouched.has(opIdOf(op)));
        if (absorb.length > 0) {
          Y.transact(this.doc, () => {
            applyBlockOperationsToYDoc(this.doc, absorb);
          }, "realtime-flush-rebase");
        }
        this.events.onRebased?.(this.key, this.resourceId, absorb.length);
        externalBase = result.conflict.externalBlocks;
        blocks = yDocToBlockList(this.doc);

        if (attempts >= 2) {
          // The single retry was used up (two external writers inside one
          // flush window — outside the single-retry contract). The doc is
          // now in-sync with the LATEST external (the next flush's full-set
          // persist clobbers nothing); keep the room dirty with its user
          // edits intact, advance the CAS base (so the next flush doesn't
          // re-conflict on the same stale version), and re-arm the debounce.
          this.lastVersion = result.version;
          this.scheduleFlush();
          return;
        }
        attempts += 1;
        result = await this.opts.store.persist(this.resourceId, blocks, result.version);
      }

      const blockDelta = diffBlockSets(this.lastPersistedBlocks, blocks);
      this.lastVersion = result.version;
      this.lastPersistedBlocks = blocks.map((b) => ({ ...b }));
      this.dirtySince = null;
      this.events.onPersisted(this.key, this.resourceId, result.version, logSeqAtFlush, blockDelta);
    })();
    this.flushing = p;
    void p.finally(() => {
      this.flushing = null;
    });
    return p;
  }

  /** Evict (room GC / drain): flush-if-dirty, then drop the doc. */
  async evict(): Promise<void> {
    if (this._evicted) return;
    this._evicted = true;
    if (this.gcTimer !== null) {
      this.opts.clock.clearTimeout(this.gcTimer);
      this.gcTimer = null;
    }
    if (this.isDirty) {
      await this.flush();
    }
    this.doc.destroy();
    this.events.onRoomEvicted(this.key);
  }

  /**
   * Mark the room mid re-hydration (the manager, after an eviction
   * re-join — normally unreachable: evicted rooms are removed from the
   * manager map; the cold path hydrates before publishing).
   */
  setHydrating(): void {
    this.hydrating = true;
  }

  /** Finish a re-hydration: swap the doc + apply buffered updates. */
  completeHydration(hydrated: Y.Doc): void {
    if (!this.hydrating) return;
    this.doc.destroy();
    this.doc = hydrated;
    for (const u of this.hydrateBuffer) {
      Y.applyUpdate(this.doc, u, "remote");
    }
    this.hydrateBuffer = [];
    this.hydrating = false;
  }

  /** Arm (or disarm) the empty-room GC timer (R2: 5 min warm room). */
  notePeersChanged(): void {
    if (this._evicted) return;
    if (this.peers.size === 0) {
      if (this.gcTimer === null) {
        this.gcTimer = this.opts.clock.setTimeout(() => {
          this.gcTimer = null;
          void this.evict();
        }, this.opts.roomGcMs);
      }
    } else if (this.gcTimer !== null) {
      this.opts.clock.clearTimeout(this.gcTimer);
      this.gcTimer = null;
    }
  }

  roster(): readonly RoomPeer[] {
    return [...this.peers.values()].map((p) => p.roster);
  }

  /** A peer's display color (userId hash, the shared 20-color palette). */
  static colorFor(userId: string): string {
    return presenceColor(userId);
  }
}

// --- the manager -------------------------------------------------------------------

export class RoomManager {
  private readonly roomMap = new Map<string, Room>();
  private readonly events: RoomManagerEvents;
  private readonly opts: Required<RoomManagerOpts>;

  constructor(opts: RoomManagerOpts, events: RoomManagerEvents) {
    this.opts = {
      store: opts.store,
      clock: opts.clock ?? realClock(),
      persistDebounceMs: opts.persistDebounceMs ?? 1000,
      persistHardCapMs: opts.persistHardCapMs ?? 10_000,
      roomGcMs: opts.roomGcMs ?? 5 * 60_000,
      hydrateBufferLimit: opts.hydrateBufferLimit ?? 5_000,
      // No log wired (unit tests / no-Redis dev) → a 0 watermark: the
      // trim keeps everything (nothing is trimmed) — the safe default.
      flushSeqProvider: opts.flushSeqProvider ?? (() => 0),
    };
    this.events = events;
  }

  /**
   * Get-or-create a doc room. Cold rooms hydrate BEFORE the room is
   * published (a joining peer never syncs against an empty doc); warm
   * rooms are returned as-is.
   */
  async getOrCreate(
    key: string,
    kind: "page" | "database",
    resourceId: string,
  ): Promise<Room> {
    const existing = this.roomMap.get(key);
    if (existing !== undefined) return existing;
    // Cold start: hydrate BEFORE constructing/publishing the room — a
    // joining peer never syncs against an empty doc (note 14: the
    // resource id is globally unique, the room key IS the tenant scope;
    // the workspace is recorded by the server for metrics only).
    const seed: HydrateResult =
      kind === "page"
        ? await this.opts.store.hydrate(resourceId)
        : { blocks: null, version: 0 }; // database rooms: the DocStore is page-shaped
    const room = new Room(key, kind, resourceId, this.opts, this.events, seed);
    this.roomMap.set(key, room);
    return room;
  }

  get(key: string): Room | undefined {
    return this.roomMap.get(key);
  }

  /** Drop a room from the map (after `evict()`). */
  remove(key: string): void {
    this.roomMap.delete(key);
  }

  /** All live rooms (drain/metrics). */
  rooms(): Room[] {
    return [...this.roomMap.values()];
  }

  /**
   * SIGTERM drain (R6): flush every dirty room, evict all, within the
   * budget (overflow is logged by the caller). Returns the counts.
   */
  async drainAll(budgetMs: number): Promise<{ flushed: number; evicted: number }> {
    const deadline = this.opts.clock.now() + budgetMs;
    let flushed = 0;
    let evicted = 0;
    for (const room of this.rooms()) {
      if (room.isDirty) {
        if (this.opts.clock.now() > deadline) break;
        await room.flush();
        flushed += 1;
      }
      await room.evict();
      this.roomMap.delete(room.key);
      evicted += 1;
    }
    return { flushed, evicted };
  }
}

/** Build a peer from a handshake identity (color = the shared palette). */
export function makePeer(
  connId: string,
  userId: string,
  canEdit: boolean,
  name?: string,
): Peer {
  const roster: RoomPeer = {
    user: { id: userId, name: name ?? "Guest (shared)", color: Room.colorFor(userId) },
  };
  return {
    connId,
    userId,
    name: roster.user.name,
    color: roster.user.color,
    canEdit,
    roster,
  };
}