/**
 * @factory/collaboration — the realtime provider (TASK-0092 R4).
 *
 * A plain TS class (NOT a React component — the canvas's `EditorCanvas`
 * instantiates it): connects to `wsUrl/realtime?token=<JWT>&room=page:{pageId}`,
 * runs the TASK-0091 R3 protocol, drives the {@link YjsDocBinding}
 * (y-prosemirror on the canvas, a fake in tests), and exposes the state
 * machine the top-bar chip + offline banner consume.
 *
 * Protocol (client side — the authoritative schemas land in
 * `packages/shared/src/realtime/frames.ts`, TASK-0091; this client is the
 * lockstep mirror, see `types.ts`):
 *
 *   connect → `hello {connId, canEdit, serverTime, roomPeers}`
 *     → `sync.request {}` (local doc empty = full) or
 *       `sync.request {stateVector}` (local state = diff)
 *       + one binary `0x01` update of the FULL local state (the
 *       "ship all pending" step — the CRDT merge is idempotent)
 *       + `outbox.pending {count}` (the drain-load signal, TASK-0095 R4.2)
 *     → paged `sync.state {chunk, done}` (base64 Yjs updates, ≤ 64 KB)
 *       applied via the binding; `done` → state `connected`
 *   steady state:
 *     → local binding updates → binary `0x01` + raw Yjs bytes (dropped
 *       locally when `canEdit` is false → `onLocalUpdateBlocked`)
 *     → 30 s app-level `ping {ts, lastSeq}` → `ping-pong`
 *     → inbound binary `0x01` → `binding.applyUpdate`; `0x02` → the
 *       canvas's y-protocols awareness handling (`onAwarenessUpdate`)
 *
 * Close codes (TASK-0091 R6): 4001 → refresh the access token (cookie
 * refresh BFF) and reconnect ONCE (a second 4001 = terminal); 4002 →
 * terminal (no access — `onClosed`, no reconnect); 4000/4008/4009 →
 * backoff reconnect `min(1 s · 2^n, maxReconnectDelayMs ~2 s)` (the
 * < 3 s reconnect budget: first attempt ~1 s, second ~2 s); 4010 → retry
 * after a fixed 5 s (server shutdown drain).
 *
 * Offline (TASK-0092 R5): the offline persistence (y-indexeddb via
 * `offlineFactory`) is created BEFORE the socket opens, so a returning
 * tab has local state for a state-vector sync. `setOnline(false)` →
 * state `offline` (local edits keep applying to the doc — no edit is
 * ever blocked offline); `setOnline(true)` → reconnect.
 *
 * Determinism: clock + timers are injectable (AGENTS.md rule 25).
 */

import * as Y from "yjs";

import { AwarenessStore } from "./awareness.js";
import { bytesToBase64, blocksToYDoc, decodeBase64Yjs, type BlockLike } from "./codec.js";
import {
  OFFLINE_AUTOSAVE_INTERVAL_MS,
  createIndexeddbOfflineFactory,
  offlineDocName,
} from "./offline.js";
import {
  BINARY_KIND,
  CLOSE_HINTS,
  WS_CLOSE,
  type ClientFrame,
  type ConnectionState,
  type HelloFrame,
  type OfflinePersistence,
  type RealtimeProviderConfig,
  type ServerFrame,
  type WSAdapter,
} from "./types.js";

/** Reconnect backoff cap (default ~2 s → the < 3 s budget: first attempt
 *  ~1 s, second ~2 s — both inside the exit criterion). */
const DEFAULT_MAX_RECONNECT_DELAY_MS = 2000;
/** First backoff step (doubles per attempt, capped). */
const FIRST_BACKOFF_MS = 1000;
/** Graceful-shutdown retry delay (4010 — the server is draining). */
const SHUTDOWN_RETRY_MS = 5000;
/** App-level heartbeat cadence (the server's own ping period is 30 s). */
const DEFAULT_PING_INTERVAL_MS = 30_000;

/**
 * The realtime provider. Instantiate via {@link createRealtimeProvider};
 * call {@link RealtimeProvider.connect} on mount and
 * {@link RealtimeProvider.dispose} on unmount.
 */
export class RealtimeProvider {
  private readonly config: RealtimeProviderConfig;
  private readonly now: () => number;
  private readonly setTimeout: typeof globalThis.setTimeout;
  private readonly clearTimeout: typeof globalThis.clearTimeout;

  private state: ConnectionState = "closed";
  private socket: WSAdapter | null = null;
  private canEdit = false;
  private connId: string | null = null;
  private lastSeq: number | null = null;
  private reconnectAttempts = 0;
  private refreshedOnce = false;
  private closedByUs = false;
  private pingTimer: ReturnType<typeof globalThis.setTimeout> | null = null;
  private reconnectTimer: ReturnType<typeof globalThis.setTimeout> | null = null;
  private offline: OfflinePersistence | null = null;
  private localUpdateUnsub: (() => void) | null = null;
  private readonly awareness: AwarenessStore | null;

  constructor(config: RealtimeProviderConfig) {
    this.config = config;
    this.now = config.now ?? (() => Date.now());
    // `typeof globalThis.setTimeout` is an overload set (node `Timer` vs DOM
    // `number`); a plain lambda is not assignable to it, and no parameter
    // types flow through the `??` — so type the params explicitly and cast
    // the default (the runtime call is the same either way).
    this.setTimeout = (config.setTimeout ?? ((fn: unknown, ms?: number) =>
      globalThis.setTimeout(fn as Parameters<typeof globalThis.setTimeout>[0], ms))) as typeof globalThis.setTimeout;
    this.clearTimeout = (config.clearTimeout ?? ((id: unknown) =>
      globalThis.clearTimeout(id as Parameters<typeof globalThis.clearTimeout>[0]))) as typeof globalThis.clearTimeout;
    this.awareness =
      config.presence !== undefined
        ? new AwarenessStore({
            self: config.presence,
            ...(config.hasBlock !== undefined ? { hasBlock: config.hasBlock } : {}),
            onPeersChange: (peers) => this.config.onPeerChange?.(peers),
            now: this.now,
          })
        : null;
  }

  /** The current connection state (the top-bar StatusChip input). */
  get connectionState(): ConnectionState {
    return this.state;
  }

  /** The join capability flag (VIEW-only rooms drop local updates). */
  get isCanEdit(): boolean {
    return this.canEdit;
  }

  /** The connection identity assigned by the server (null before hello). */
  get connectionId(): string | null {
    return this.connId;
  }

  /** The awareness store (null when presence is disabled). */
  get awarenessStore(): AwarenessStore | null {
    return this.awareness;
  }

  /**
   * The pending outbox count (the banner's sum input — the CRDT-pending
   * proxy; the app adds its own outbox count, TASK-0095 R2).
   */
  get localPendingCount(): number {
    return this.config.getOutboxPendingCount?.() ?? 0;
  }

  /**
   * Open the offline persistence (BEFORE the socket — R5), seed the doc
   * if needed, and connect. No-op when already connecting/syncing/
   * connected; `offline` when the network is down (reconnects on
   * {@link setOnline}).
   */
  async connect(): Promise<void> {
    if (this.closedByUs) return;
    if (
      this.state === "connecting" ||
      this.state === "syncing" ||
      this.state === "connected"
    ) {
      return;
    }
    if (!this.isOnline()) {
      this.setState("offline");
      return;
    }
    this.setState("connecting");

    // R5: load the offline doc BEFORE the sync.request — a returning tab
    // gets a state-vector sync (diff) instead of a full doc.
    const factory = this.config.offlineFactory ?? createIndexeddbOfflineFactory();
    try {
      this.offline = await factory.create(
        offlineDocName(this.config.workspaceId, this.config.pageId),
        this.config.binding.doc,
        { autoSaveInterval: OFFLINE_AUTOSAVE_INTERVAL_MS },
      );
    } catch {
      this.offline = null; // no IndexedDB (e.g. private mode) — online-only
    }

    // Seed the doc from `initialBlocks` only when nothing else provided
    // state (no offline doc content, no ProseMirror view state yet).
    if (!this.config.binding.hasLocalState() && this.config.initialBlocks !== undefined) {
      this.seedInitialBlocks();
    }

    this.localUpdateUnsub ??= this.config.binding.onLocalUpdate((update) => {
      this.handleLocalUpdate(update);
    });

    this.openSocket();
  }

  /** The app's connectivity listener (online/offline events). */
  setOnline(online: boolean): void {
    if (online) {
      if (this.state === "offline") void this.connect();
    } else {
      this.setState("offline");
      this.stopPing();
      // Close the socket explicitly — an offline connection is a zombie
      // (the reconnect loop is off until setOnline(true)).
      const socket = this.socket;
      if (socket !== null && (socket.readyState === 0 || socket.readyState === 1)) {
        socket.close(1000, "client-offline");
      }
    }
  }

  /** Flush pending offline state + release (pagehide / unmount). */
  async dispose(): Promise<void> {
    this.closedByUs = true;
    this.stopPing();
    this.clearReconnect();
    const socket = this.socket;
    if (socket !== null) {
      socket.onopen = null;
      socket.onmessage = null;
      socket.onclose = null;
      socket.onerror = null;
      if (socket.readyState === 0 || socket.readyState === 1) {
        socket.close(1000, "dispose");
      }
      this.socket = null;
    }
    this.localUpdateUnsub?.();
    this.localUpdateUnsub = null;
    await this.offline?.dispose();
    this.offline = null;
    this.setState("closed");
  }

  // --- internals -------------------------------------------------------------

  private isOnline(): boolean {
    return this.config.isOnline?.() ?? true;
  }

  private setState(state: ConnectionState): void {
    if (this.state === state) return;
    this.state = state;
    this.config.onStateChange?.(state);
  }

  private socketUrl(): string {
    const base = this.config.wsUrl.replace(/\/+$/, "");
    const token = encodeURIComponent(this.config.getToken());
    const room = encodeURIComponent(`page:${this.config.pageId}`);
    return `${base}/realtime?token=${token}&room=${room}`;
  }

  private openSocket(): void {
    const socket = this.config.createSocket(this.socketUrl());
    this.socket = socket;

    socket.onopen = () => {
      // The room join is confirmed by `hello` — nothing is sent until
      // then (the server authorizes the join at handshake).
    };

    socket.onmessage = (event) => {
      const data = event.data;
      if (typeof data === "string") {
        this.handleTextFrame(data);
      } else {
        this.handleBinaryFrame(data instanceof Uint8Array ? data : new Uint8Array(data));
      }
    };

    socket.onclose = (event) => {
      this.handleClose(event.code);
    };

    socket.onerror = () => {
      // The close event follows (the WS contract); nothing to do here.
    };
  }

  private handleTextFrame(raw: string): void {
    let frame: ServerFrame;
    try {
      frame = JSON.parse(raw) as ServerFrame;
    } catch {
      this.config.onProtocolError?.({ code: "bad_frame", message: raw.slice(0, 200) });
      return;
    }
    switch (frame.type) {
      case "hello":
        this.handleHello(frame);
        break;
      case "sync.state": {
        this.config.binding.applyUpdate(decodeBase64Yjs(frame.chunk));
        if (frame.done) this.setState("connected");
        break;
      }
      case "ack":
        if (frame.seq > (this.lastSeq ?? 0)) this.lastSeq = frame.seq;
        break;
      case "doc.persisted":
        this.config.onDocPersisted?.({ pageId: frame.pageId, version: frame.version });
        break;
      case "comment.event":
        this.config.onCommentEvent?.(frame);
        break;
      case "notification.push":
        this.config.onNotificationPush?.(frame);
        break;
      case "hydrating":
        this.config.onHydrating?.(frame.pageId);
        break;
      case "error":
        this.config.onProtocolError?.({ code: frame.code, message: frame.message });
        break;
      case "ping-pong":
        break;
    }
  }

  private handleHello(frame: HelloFrame): void {
    this.connId = frame.connId;
    this.canEdit = frame.canEdit;
    this.reconnectAttempts = 0;
    this.refreshedOnce = false;
    this.awareness?.applyRoster(frame.roomPeers);
    this.setState("syncing");

    // R4: empty local doc → full sync; local state → state-vector diff.
    const hasLocal = this.config.binding.hasLocalState();
    const request: ClientFrame = hasLocal
      ? {
          type: "sync.request",
          stateVector: bytesToBase64(this.config.binding.stateVector()),
          ...(this.lastSeq !== null ? { lastSeq: this.lastSeq } : {}),
        }
      : { type: "sync.request" };
    this.send(request);

    // "Then any local-pending updates" — ship the full local state once
    // (the CRDT merge is idempotent; the server discards duplicates).
    if (hasLocal) {
      this.sendBinary(BINARY_KIND.Update, this.config.binding.encodeStateAsUpdate());
    }

    // The outbox drain-load signal (TASK-0095 R4.2).
    if (this.config.getOutboxPendingCount !== undefined) {
      this.send({ type: "outbox.pending", count: this.config.getOutboxPendingCount() });
    }

    this.startPing();
  }

  private handleBinaryFrame(bytes: Uint8Array): void {
    if (bytes.length === 0) return;
    const kind = bytes[0] ?? 0;
    const payload = bytes.slice(1);
    if (kind === BINARY_KIND.Update) {
      this.config.binding.applyUpdate(payload);
    } else if (kind === BINARY_KIND.Awareness) {
      this.config.onAwarenessUpdate?.(payload);
    }
  }

  private handleLocalUpdate(update: Uint8Array): void {
    if (!this.canEdit) {
      // Defense in depth (mirrors the server): a VIEW-only local update
      // is dropped, never sent.
      this.config.onLocalUpdateBlocked?.();
      return;
    }
    const socket = this.socket;
    if (
      socket !== null &&
      socket.readyState === 1 &&
      (this.state === "syncing" || this.state === "connected")
    ) {
      this.sendBinary(BINARY_KIND.Update, update);
    }
    // Offline / reconnecting: the update stays in the local doc; the
    // reconnect ships the full local state (idempotent merge).
  }

  private handleClose(code: number): void {
    this.stopPing();
    this.socket = null;
    if (this.closedByUs) return;

    const hint = CLOSE_HINTS[code] ?? "Connection lost — reconnecting…";

    // 4001 unauthorized: refresh the access token and reconnect ONCE
    // (a second 4001 after the refresh = terminal — not a token problem).
    if (code === WS_CLOSE.Unauthorized) {
      if (!this.refreshedOnce) {
        this.refreshedOnce = true;
        void this.refreshAndReconnect(hint);
        return;
      }
      this.setState("closed");
      this.config.onClosed?.(code, hint);
      this.closedByUs = true; // stop the reconnect loop
      return;
    }

    // 4002 forbidden: terminal — the user lost access (the app navigates).
    if (code === WS_CLOSE.Forbidden) {
      this.setState("closed");
      this.config.onClosed?.(code, hint);
      this.closedByUs = true;
      return;
    }

    // Offline: no reconnect loop — wait for setOnline(true).
    if (!this.isOnline()) {
      this.setState("offline");
      return;
    }

    this.setState("reconnecting");
    if (code === WS_CLOSE.ServerShutdown) {
      // Graceful drain: a fixed 5 s wait (the server is restarting).
      this.scheduleReconnect(SHUTDOWN_RETRY_MS);
      return;
    }

    const delay = Math.min(
      FIRST_BACKOFF_MS * 2 ** this.reconnectAttempts,
      this.config.maxReconnectDelayMs ?? DEFAULT_MAX_RECONNECT_DELAY_MS,
    );
    this.reconnectAttempts += 1;
    this.scheduleReconnect(delay);
  }

  private async refreshAndReconnect(hint: string): Promise<void> {
    try {
      await this.config.refreshAccessToken();
      this.setState("reconnecting");
      this.scheduleReconnect(0);
    } catch {
      this.setState("closed");
      this.config.onClosed?.(WS_CLOSE.Unauthorized, hint);
      this.closedByUs = true;
    }
  }

  private scheduleReconnect(delayMs: number): void {
    this.clearReconnect();
    this.reconnectTimer = this.setTimeout(() => {
      this.reconnectTimer = null;
      if (this.closedByUs) return;
      this.setState("connecting");
      this.openSocket();
    }, delayMs);
  }

  private clearReconnect(): void {
    if (this.reconnectTimer !== null) {
      this.clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }
  }

  private startPing(): void {
    this.stopPing();
    const interval = this.config.pingIntervalMs ?? DEFAULT_PING_INTERVAL_MS;
    const tick = () => {
      const socket = this.socket;
      if (
        socket !== null &&
        socket.readyState === 1 &&
        (this.state === "syncing" || this.state === "connected")
      ) {
        const ping: ClientFrame = {
          type: "ping",
          ts: this.now(),
          ...(this.lastSeq !== null ? { lastSeq: this.lastSeq } : {}),
        };
        this.send(ping);
      }
      this.pingTimer = this.setTimeout(tick, interval);
    };
    this.pingTimer = this.setTimeout(tick, interval);
  }

  private stopPing(): void {
    if (this.pingTimer !== null) {
      this.clearTimeout(this.pingTimer);
      this.pingTimer = null;
    }
  }

  private send(frame: ClientFrame): void {
    const socket = this.socket;
    if (socket !== null && socket.readyState === 1) {
      socket.send(JSON.stringify(frame));
    }
  }

  private sendBinary(kind: number, bytes: Uint8Array): void {
    const socket = this.socket;
    if (socket === null || socket.readyState !== 1) return;
    const out = new Uint8Array(1 + bytes.length);
    out[0] = kind;
    out.set(bytes, 1);
    socket.send(out);
  }

  private seedInitialBlocks(): void {
    // The canvas normally seeds the doc through the binding (the
    // ProseMirror view). `initialBlocks` is the task's `pageBlocks`
    // fallback: a plain block set the codec converts into the doc.
    const blocks = this.config.initialBlocks;
    if (blocks === undefined) return;
    const seeded = blocksToYDoc(blocks as unknown as BlockLike[], {
      pageId: this.config.pageId,
    });
    Y.applyUpdate(this.config.binding.doc, Y.encodeStateAsUpdate(seeded));
    seeded.destroy();
  }
}

/** The task's `createRealtimeProvider({pageId, workspaceId, …})` factory. */
export function createRealtimeProvider(config: RealtimeProviderConfig): RealtimeProvider {
  return new RealtimeProvider(config);
}