/**
 * Offline delivery via the relay node's store-and-forward mailbox
 * (PROTOCOL.md §9). The mailbox is a second transport for the SAME bytes the
 * log replicates — sealed entry blocks, unixfs image blocks and (sealed for
 * transit) identity record blocks — so two members who are never online
 * together still converge. The log stays the single source of truth; the
 * node stores opaque blobs it cannot read, deduped per room by SHA-256.
 *
 * Deposit is author-only (each device uploads only what it created; the
 * persistent queue IS the catch-up story, server dedup absorbs re-sends).
 * Replay is two-phase: store every block first, then join in seq order, so
 * no join ever waits on a bitswap timeout. Unjoinable entries (ancestors
 * expired from the ring buffer) surface display-only and sit on a retry list
 * drained when live replication backfills the gap.
 *
 * Everything here is best-effort: a mailbox failure must never break live
 * chat, so every path swallows into typed results or console.warn.
 */

import { CID } from "multiformats/cid";
import { base58btc } from "multiformats/bases/base58";
import type { OrbitEntry } from "@orbitdb/core";
import { fromB64, type AtRestCipher, type Bytes } from "./crypto";
import type { LogEntryMeta, RoomLog } from "./log";

const MINT_TTL_SEC = 86_400; // ask for the max (24 h); the node may clamp
const REPLAY_THROTTLE_MS = 30_000;
const BACKOFF_BASE_MS = 1_000;
const BACKOFF_CAP_MS = 300_000;
// A solo offline drawing session can exceed 500 ops, and drop-oldest would
// silently lose the EARLIEST strokes — boards default higher (queueCap option).
const QUEUE_CAP = 2000; // drop-oldest beyond this (default; see MailboxSyncOptions.queueCap)
const SENT_CAP = 1000;
const DAG_CBOR = 0x71;
const SHA2_256 = 0x12;

// Deposit envelope, one mailbox message per block (deterministic so the
// node's SHA-256 dedup collapses re-sends):
//   0x01 | cid.bytes | block bytes          raw blocks (sealed entries, image blocks)
//   0x02 | seal(cid.bytes | block bytes)    identity records (plaintext blocks — §7b)
const RAW_BLOCK = 0x01;
const SEALED_BLOCK = 0x02;

export interface MailboxKeys {
  cursor: string;
  queue: string;
  sent: string;
  retry: string;
  ident: string;
}

type QItem = { t: "entry"; hash: string } | { t: "img"; cid: string; group: string } | { t: "ident" };

// ---- auth: browser open-mint, cached bearer (turn.ts pattern) --------------

interface MintResponse {
  token: string;
  ttl: number;
}

export class MailboxAuth {
  #credsUrl: string;
  #roomId: string;
  #cached: string | null = null;
  #validUntil = 0;
  #inflight: Promise<string> | null = null;

  constructor(credsUrl: string, roomId: string) {
    this.#credsUrl = credsUrl;
    this.#roomId = roomId;
  }

  get(): Promise<string> {
    if (this.#cached && Date.now() < this.#validUntil) return Promise.resolve(this.#cached);
    this.#inflight ??= this.#mint().finally(() => (this.#inflight = null));
    return this.#inflight;
  }

  /** Drop the cached token (server said 401/403) so the next get() re-mints. */
  invalidate(): void {
    this.#cached = null;
    this.#validUntil = 0;
  }

  async #mint(): Promise<string> {
    const url = `${this.#credsUrl}?room=${encodeURIComponent(this.#roomId)}&ttl=${MINT_TTL_SEC}`;
    const res = await fetch(url, { signal: AbortSignal.timeout(15_000) });
    if (!res.ok) throw new Error(`mailbox mint ${res.status}`);
    const c = (await res.json()) as MintResponse;
    if (!c.token) throw new Error("mailbox mint: no token");
    this.#cached = c.token;
    this.#validUntil = Date.now() + (c.ttl ?? 0) * 1000 - 300_000; // renew 5 min early
    return c.token;
  }
}

// ---- bare HTTP client with typed results ------------------------------------

export type DepositResult =
  | { ok: true; deduped: boolean; seq: string }
  | { ok: false; kind: "too_large" | "auth" | "backpressure" | "net" };

export interface ReplayMessage {
  seq: string;
  id: string;
  ts: number;
  payload: string; // base64
}

export interface ReplayPage {
  messages: ReplayMessage[];
  hasMore: boolean;
  oldestSeq: string | null;
  latestSeq: string | null;
}

export type ReplayResult = { ok: true; page: ReplayPage } | { ok: false; kind: "auth" | "backpressure" | "net" };

export class MailboxClient {
  #auth: MailboxAuth;
  #base: string; // <baseUrl>/rooms/<roomId>/messages

  constructor(baseUrl: string, roomId: string, auth: MailboxAuth) {
    this.#auth = auth;
    this.#base = `${baseUrl.replace(/\/$/, "")}/rooms/${encodeURIComponent(roomId)}/messages`;
  }

  async deposit(payload: Uint8Array): Promise<DepositResult> {
    const res = await this.#send(() => ({
      method: "POST",
      headers: { "content-type": "application/octet-stream" },
      body: payload as BodyInit,
    }));
    if (res === null) return { ok: false, kind: "net" };
    if (res.status === 201 || res.status === 200) {
      try {
        const b = (await res.json()) as { deduped?: boolean; seq?: string };
        return { ok: true, deduped: !!b.deduped, seq: b.seq ?? "" };
      } catch {
        return { ok: true, deduped: false, seq: "" };
      }
    }
    if (res.status === 413) return { ok: false, kind: "too_large" };
    if (res.status === 401 || res.status === 403) return { ok: false, kind: "auth" };
    return { ok: false, kind: res.status === 429 || res.status === 507 ? "backpressure" : "net" };
  }

  async replay(after: string | null, limit: number): Promise<ReplayResult> {
    const qs = `?limit=${limit}${after ? `&after=${encodeURIComponent(after)}` : ""}`;
    const res = await this.#send(() => ({ method: "GET" }), qs);
    if (res === null) return { ok: false, kind: "net" };
    if (res.status === 401 || res.status === 403) return { ok: false, kind: "auth" };
    if (!res.ok) return { ok: false, kind: res.status === 429 ? "backpressure" : "net" };
    try {
      const b = (await res.json()) as {
        messages?: ReplayMessage[];
        has_more?: boolean;
        oldest_seq?: string;
        latest_seq?: string;
      };
      return {
        ok: true,
        page: {
          messages: Array.isArray(b.messages) ? b.messages : [],
          hasMore: !!b.has_more,
          oldestSeq: b.oldest_seq ?? null,
          latestSeq: b.latest_seq ?? null,
        },
      };
    } catch {
      return { ok: false, kind: "net" };
    }
  }

  /** One request with the cached bearer; on 401/403 re-mint once and retry. */
  async #send(init: () => RequestInit, qs = ""): Promise<Response | null> {
    for (let attempt = 0; attempt < 2; attempt++) {
      let token: string;
      try {
        token = await this.#auth.get();
      } catch {
        return null;
      }
      let res: Response;
      try {
        res = await fetch(this.#base + qs, {
          ...init(),
          headers: { ...(init().headers as Record<string, string>), authorization: `Bearer ${token}` },
          signal: AbortSignal.timeout(20_000),
        });
      } catch {
        return null;
      }
      if ((res.status === 401 || res.status === 403) && attempt === 0) {
        this.#auth.invalidate();
        continue;
      }
      return res;
    }
    return null;
  }
}

// ---- seq ordering: opaque strings, server-ordered ----------------------------

/** a < b for mailbox seqs — never numeric-parse (they overflow doubles). */
function seqLt(a: string, b: string): boolean {
  return a.length !== b.length ? a.length < b.length : a < b;
}

// ---- the orchestrator --------------------------------------------------------

export interface MailboxSyncOptions {
  log: RoomLog;
  client: MailboxClient;
  cipher: AtRestCipher; // rc.mbxCipher() — seals identity records in transit
  maxMessageKb: number;
  keys: MailboxKeys; // pre-namespaced localStorage keys (nsKey(...))
  hasIdentity: boolean;
  onDisplayOnly?: (e: LogEntryMeta) => void;
  onNotice?: (msg: string) => void;
  onReplayed?: (count: number) => void;
  /** Deposit-queue cap, drop-oldest beyond it (default QUEUE_CAP). */
  queueCap?: number;
}

export class MailboxSync {
  #log: RoomLog;
  #client: MailboxClient;
  #cipher: AtRestCipher;
  #maxBytes: number;
  #keys: MailboxKeys;
  #hasIdentity: boolean;
  #queueCap: number;
  #onDisplayOnly: (e: LogEntryMeta) => void;
  #onNotice: (msg: string) => void;
  #onReplayed: (count: number) => void;

  #queue: QItem[] = [];
  #sent: string[] = []; // hashes/CIDs already deposited (ring; server dedup is the real guard)
  #retry: string[] = []; // stored-but-unjoinable entry hashes
  #flushing = false;
  #replaying = false;
  #lastReplay = 0;
  #attempt = 0;
  #retryTimer: ReturnType<typeof setTimeout> | null = null;
  #liveTimer: ReturnType<typeof setTimeout> | null = null;
  #noticed = false;
  #enqueueChain: Promise<void> = Promise.resolve(); // keeps image blocks ahead of their entry
  #listeners: Array<[string, () => void]> = [];

  constructor(opts: MailboxSyncOptions) {
    this.#log = opts.log;
    this.#client = opts.client;
    this.#cipher = opts.cipher;
    this.#maxBytes = opts.maxMessageKb * 1024;
    this.#keys = opts.keys;
    this.#hasIdentity = opts.hasIdentity;
    this.#queueCap = opts.queueCap ?? QUEUE_CAP;
    this.#onDisplayOnly = opts.onDisplayOnly ?? (() => {});
    this.#onNotice = opts.onNotice ?? (() => {});
    this.#onReplayed = opts.onReplayed ?? (() => {});
    this.#queue = loadJson<QItem[]>(this.#keys.queue, []);
    this.#sent = loadJson<string[]>(this.#keys.sent, []);
    this.#retry = loadJson<string[]>(this.#keys.retry, []);
  }

  /** Non-blocking: replay (which also drains the retry list), then flush. */
  start(): void {
    const kick = () => {
      if (document.visibilityState === "hidden") return;
      this.#attempt = 0;
      if (this.#retryTimer) clearTimeout(this.#retryTimer);
      this.#retryTimer = null;
      void this.#flush();
      void this.#replay(false);
    };
    this.#listeners = [
      ["visibilitychange", kick],
      ["pageshow", kick],
      ["online", kick],
    ];
    for (const [evt, fn] of this.#listeners) window.addEventListener(evt, fn);
    void (async () => {
      await this.#replay(true);
      await this.#flush();
    })();
  }

  stop(): void {
    for (const [evt, fn] of this.#listeners) window.removeEventListener(evt, fn);
    if (this.#retryTimer) clearTimeout(this.#retryTimer);
    if (this.#liveTimer) clearTimeout(this.#liveTimer);
  }

  /** Debug/e2e visibility. */
  get state(): { queued: number; retry: number; cursor: string | null } {
    return { queued: this.#queue.length, retry: this.#retry.length, cursor: loadStr(this.#keys.cursor) };
  }

  // ---- deposit side ----------------------------------------------------------

  /**
   * RoomLog.onAppended — this device authored a new entry. Synchronous up to
   * the persisted queue, so a page closed right after send still deposits on
   * the next visit.
   */
  enqueueEntry(hash: string): void {
    if (this.#sent.includes(hash)) return;
    this.#enqueueIdentityOnce();
    this.#push({ t: "entry", hash });
    saveJson(this.#keys.queue, this.#queue);
    void this.#flush();
  }

  /**
   * RoomLog.onImageStored — expand the image DAG into per-block items. The
   * expansion is async, so image blocks may enqueue after their entry; replay
   * is two-phase (all blocks stored before any join), so order across items
   * doesn't matter.
   */
  enqueueImage(rootCid: string): void {
    this.#enqueueChain = this.#enqueueChain
      .then(async () => {
        const blocks = await this.#log.imageDagBlocks(rootCid); // children first, root last
        this.#enqueueIdentityOnce();
        for (const b of blocks) {
          const cid = b.cid.toString();
          if (!this.#sent.includes(cid)) this.#push({ t: "img", cid, group: rootCid });
        }
        saveJson(this.#keys.queue, this.#queue);
      })
      .catch(() => {})
      .then(() => void this.#flush());
  }

  #enqueueIdentityOnce(): void {
    if (!this.#hasIdentity || loadStr(this.#keys.ident)) return;
    if (this.#queue.some((i) => i.t === "ident")) return;
    this.#queue.push({ t: "ident" }); // before the entry — receivers need it for canAppend
  }

  #push(item: QItem): void {
    this.#queue.push(item);
    if (this.#queue.length > this.#queueCap) this.#queue.splice(0, this.#queue.length - this.#queueCap);
  }

  async #flush(): Promise<void> {
    if (this.#flushing || this.#queue.length === 0) return;
    this.#flushing = true;
    try {
      while (this.#queue.length > 0) {
        const item = this.#queue[0];
        const payload = await this.#buildPayload(item);
        if (payload === null || payload.length > this.#maxBytes) {
          this.#dropHead(item); // unbuildable or over the frame cap — same as a 413
          continue;
        }
        const r = await this.#client.deposit(payload);
        if (r.ok) {
          this.#queue.shift();
          if (item.t === "entry") this.#markSent(item.hash);
          else if (item.t === "img") this.#markSent(item.cid);
          else saveStr(this.#keys.ident, "1");
          saveJson(this.#keys.queue, this.#queue);
          this.#attempt = 0;
          continue;
        }
        if (r.kind === "too_large") {
          this.#dropHead(item);
          continue;
        }
        // auth (already re-minted once inside the client), backpressure, net:
        this.#scheduleRetry();
        break;
      }
    } catch (err) {
      console.warn("mailbox flush failed", err);
      this.#scheduleRetry();
    } finally {
      this.#flushing = false;
    }
    // an enqueue that landed while we were exiting saw #flushing and bailed —
    // re-check so its item isn't stranded until the next wake kick
    if (this.#queue.length > 0 && !this.#retryTimer) void this.#flush();
  }

  /** Drop the head item — and, for an image block, its whole DAG group. */
  #dropHead(item: QItem): void {
    this.#queue.shift();
    if (item.t === "img") this.#queue = this.#queue.filter((i) => !(i.t === "img" && i.group === item.group));
    saveJson(this.#keys.queue, this.#queue);
  }

  #markSent(key: string): void {
    this.#sent.push(key);
    if (this.#sent.length > SENT_CAP) this.#sent.splice(0, this.#sent.length - SENT_CAP);
    saveJson(this.#keys.sent, this.#sent);
  }

  #scheduleRetry(): void {
    if (this.#retryTimer) return;
    const delay = Math.min(BACKOFF_BASE_MS * 2 ** this.#attempt++, BACKOFF_CAP_MS);
    this.#retryTimer = setTimeout(() => {
      this.#retryTimer = null;
      void this.#flush();
    }, delay);
  }

  async #buildPayload(item: QItem): Promise<Uint8Array | null> {
    if (item.t === "entry") {
      const bytes = await this.#log.sealedEntryBytes(item.hash);
      if (!bytes) return null;
      try {
        return concat([RAW_BLOCK], CID.parse(item.hash, base58btc).bytes, bytes);
      } catch {
        return null;
      }
    }
    if (item.t === "img") {
      try {
        const cid = CID.parse(item.cid);
        const bytes = await this.#log.rawBlock(cid);
        return bytes ? concat([RAW_BLOCK], cid.bytes, bytes) : null;
      } catch {
        return null;
      }
    }
    const blk = this.#log.myIdentityBlock();
    if (!blk) return null;
    const sealed = await this.#cipher.seal(concat([], blk.cid.bytes, blk.bytes));
    return concat([SEALED_BLOCK], sealed);
  }

  // ---- replay side -------------------------------------------------------------

  /** Live replication delivered an entry — a gap may have closed; drain soon. */
  onLiveEntry(): void {
    if (this.#retry.length === 0 || this.#liveTimer) return;
    this.#liveTimer = setTimeout(() => {
      this.#liveTimer = null;
      void this.#drainRetry();
    }, 1_500);
  }

  async #replay(force: boolean): Promise<void> {
    if (this.#replaying) return;
    if (!force && Date.now() - this.#lastReplay < REPLAY_THROTTLE_MS) return;
    this.#replaying = true;
    this.#lastReplay = Date.now();
    try {
      let cursor = loadStr(this.#keys.cursor); // absent ⇒ from the beginning (new device)
      const collected: OrbitEntry[] = [];
      let gap = false;
      for (let pages = 0; pages < 200; pages++) {
        const r = await this.#client.replay(cursor, 100);
        if (!r.ok) break; // transient — the next kick retries; joins below still run
        const p = r.page;
        if (cursor && p.oldestSeq && seqLt(cursor, p.oldestSeq)) gap = true;
        for (const m of p.messages) {
          try {
            await this.#processMessage(m, collected);
          } catch (err) {
            console.warn("mailbox message dropped", err);
          }
        }
        // cursor advances only past fully-processed pages; re-processing is idempotent
        if (p.messages.length > 0) {
          cursor = p.messages[p.messages.length - 1].seq;
          saveStr(this.#keys.cursor, cursor);
        }
        if (!p.hasMore) break;
      }
      // two-phase: every block is stored before the first join, so joins never
      // wait on the network. collected is in seq order (pages are ordered).
      const joined = await this.#joinPhase(collected);
      if (joined > 0) this.#onReplayed(joined);
      await this.#drainRetry();
      if ((gap || this.#retry.length > 0) && !this.#noticed) {
        this.#noticed = true;
        this.#onNotice("some offline messages expired before this device could fetch them");
      }
      void this.#flush(); // dedup makes double-deposits harmless
    } catch (err) {
      console.warn("mailbox replay failed", err);
    } finally {
      this.#replaying = false;
    }
  }

  /** Classify, hash-verify and store one replayed blob; collect entry candidates. */
  async #processMessage(m: ReplayMessage, collected: OrbitEntry[]): Promise<void> {
    const raw = fromB64(m.payload);
    if (raw.length < 2) return;
    let body: Bytes = raw.subarray(1);
    const sealed = raw[0] === SEALED_BLOCK;
    if (sealed) {
      const opened = await this.#cipher.open(body);
      if (opened === null) return; // not ours / tampered
      body = opened;
    } else if (raw[0] !== RAW_BLOCK) {
      return;
    }
    let cid: CID;
    let block: Uint8Array;
    try {
      [cid, block] = CID.decodeFirst(body);
    } catch {
      return;
    }
    // never trust the sender's CID: re-hash before anything touches the blockstore
    if (cid.multihash.code !== SHA2_256) return;
    const digest = new Uint8Array(await crypto.subtle.digest("SHA-256", block as BufferSource));
    if (!bytesEq(digest, cid.multihash.digest)) return;

    if (sealed) {
      await this.#log.putIdentityBlock(cid, block);
      return;
    }
    if (cid.code === DAG_CBOR) {
      const entry = await this.#log.decodeSealedEntry(block);
      if (!entry) return; // dag-cbor garbage — not a room entry
      if (await this.#log.hasEntry(entry.hash)) return;
      await this.#log.putEntryBlock(entry.hash, block);
      // on the retry list from the moment it's stored — a crash between store
      // and join must not strand it behind an advanced cursor
      this.#addRetry(entry.hash);
      collected.push(entry);
      return;
    }
    await this.#log.putRawBlock(cid, block); // unixfs image block (raw or dag-pb root)
  }

  /** Join in order; project unjoinables display-only; pin ingested image roots. */
  async #joinPhase(entries: OrbitEntry[]): Promise<number> {
    let joined = 0;
    const imgRoots = new Set<string>();
    for (const entry of entries) {
      const result = await this.#log.ingestEntry(entry);
      if (result === "joined") joined++;
      if (result !== "deferred") this.#removeRetry(entry.hash);
      if (result === "joined" || result === "deferred") {
        const meta = await this.#log.entryMeta(entry);
        if (meta) {
          if (result === "deferred") this.#onDisplayOnly(meta); // ChatStore dedups by hash
          const op = meta.op as { t?: string; cid?: string };
          if (op.t === "img" && typeof op.cid === "string") imgRoots.add(op.cid);
        }
      }
    }
    if (joined > 0) await this.#log.emitUnseen();
    for (const root of imgRoots) {
      // no online source may exist for a mailbox image — pin or the next gc eats it
      await this.#log.pinImageIfLocal(root).catch(() => {});
    }
    return joined;
  }

  /** Re-attempt stored-but-unjoined entries (their blocks are already local). */
  async #drainRetry(): Promise<void> {
    if (this.#retry.length === 0) return;
    const entries: OrbitEntry[] = [];
    for (const hash of [...this.#retry]) {
      if (await this.#log.hasEntry(hash)) {
        this.#removeRetry(hash); // live replication got there first
        continue;
      }
      const bytes = await this.#log.sealedEntryBytes(hash);
      const entry = bytes ? await this.#log.decodeSealedEntry(bytes) : null;
      if (!entry) {
        this.#removeRetry(hash); // block vanished (gc) — nothing to retry
        continue;
      }
      entries.push(entry);
    }
    const joined = await this.#joinPhase(entries);
    if (joined > 0) this.#onReplayed(joined);
  }

  #addRetry(hash: string): void {
    if (this.#retry.includes(hash)) return;
    this.#retry.push(hash);
    saveJson(this.#keys.retry, this.#retry);
  }

  #removeRetry(hash: string): void {
    const i = this.#retry.indexOf(hash);
    if (i === -1) return;
    this.#retry.splice(i, 1);
    saveJson(this.#keys.retry, this.#retry);
  }
}

// ---- small helpers ------------------------------------------------------------

function concat(prefix: number[], ...parts: Uint8Array[]): Bytes {
  const out = new Uint8Array(prefix.length + parts.reduce((n, p) => n + p.length, 0));
  out.set(prefix, 0);
  let off = prefix.length;
  for (const p of parts) {
    out.set(p, off);
    off += p.length;
  }
  return out;
}

function bytesEq(a: Uint8Array, b: Uint8Array): boolean {
  if (a.length !== b.length) return false;
  let diff = 0;
  for (let i = 0; i < a.length; i++) diff |= a[i] ^ b[i];
  return diff === 0;
}

// localStorage, tolerant of Safari private mode / quota (mailbox state is
// best-effort: losing it costs re-deposits the server dedups away)
function loadStr(key: string): string | null {
  try {
    return localStorage.getItem(key);
  } catch {
    return null;
  }
}

function saveStr(key: string, value: string): void {
  try {
    localStorage.setItem(key, value);
  } catch {}
}

function loadJson<T>(key: string, fallback: T): T {
  try {
    const raw = localStorage.getItem(key);
    return raw ? (JSON.parse(raw) as T) : fallback;
  } catch {
    return fallback;
  }
}

function saveJson(key: string, value: unknown): void {
  try {
    localStorage.setItem(key, JSON.stringify(value));
  } catch {}
}