/**
 * The per-room update log for reconnect resume (TASK-0091 R5).
 *
 * The doc §4.5 "in-memory ring buffer" is implemented in Redis (task
 * resolution, header note: Redis adds restart safety, identical bounds) —
 * `realtime:updates:{roomKey}` as a Redis list, capped at the LAST 2,000
 * entries OR ~10 MB (whichever hits first). On a successful Postgres
 * flush the log is trimmed to post-flush entries.
 *
 * A seq is a monotonic integer per room, starting at 1. A client that
 * reports `lastSeq` gets entries `(lastSeq, +∞)` when the log still
 * covers `lastSeq + 1` (fast path); a gap falls back to the state-vector
 * delta (always correct — the log is a speed optimization only, R5).
 *
 * {@link InMemoryUpdateLog} is the same contract without Redis — unit/
 * integration tests and the no-Redis dev mode use it; {@link
 * RedisUpdateLog} delegates to a minimal backend surface that ioredis
 * satisfies structurally (injectable — the tests pass a fake).
 */

import { REALTIME_LIMITS } from "@factory/shared/realtime/frames";

export interface UpdateLogEntry {
  readonly seq: number;
  /** The Yjs update bytes (base64 — Redis-safe). */
  readonly updateB64: string;
  /** The page version this update preceded (0 = before any flush). */
  readonly pageVersion: number;
}

export interface UpdateLog {
  /** Append an update; returns the assigned seq. */
  append(roomKey: string, updateB64: string, pageVersion: number): Promise<number>;
  /**
   * Entries with seq in `(afterSeq, +∞)`. `null` when the log does NOT
   * cover `afterSeq + 1` (gap — the caller falls back to state vectors).
   */
  entriesAfter(roomKey: string, afterSeq: number): Promise<UpdateLogEntry[] | null>;
  /** The highest seq ever appended (0 = empty). */
  maxSeq(roomKey: string): Promise<number>;
  /** Trim to entries strictly after `keepSeq` (post-flush). */
  trimAfter(roomKey: string, keepSeq: number): Promise<void>;
  /** Drop a room's log entirely (room GC). */
  clear(roomKey: string): Promise<void>;
}

// --- in-memory implementation (tests + no-Redis dev mode) -----------------------

interface MemEntry {
  seq: number;
  updateB64: string;
  pageVersion: number;
}

export class InMemoryUpdateLog implements UpdateLog {
  private readonly maxEntries: number;
  private readonly maxBytes: number;
  private readonly logs = new Map<string, { entries: MemEntry[]; maxSeq: number }>();

  constructor(opts?: { maxEntries?: number; maxBytes?: number }) {
    this.maxEntries = opts?.maxEntries ?? REALTIME_LIMITS.updateLogMaxEntries;
    this.maxBytes = opts?.maxBytes ?? REALTIME_LIMITS.updateLogMaxBytes;
  }

  private log(roomKey: string): { entries: MemEntry[]; maxSeq: number } {
    let l = this.logs.get(roomKey);
    if (l === undefined) {
      l = { entries: [], maxSeq: 0 };
      this.logs.set(roomKey, l);
    }
    return l;
  }

  async append(roomKey: string, updateB64: string, pageVersion: number): Promise<number> {
    const l = this.log(roomKey);
    const seq = l.maxSeq + 1;
    l.entries.push({ seq, updateB64, pageVersion });
    l.maxSeq = seq;
    this.evict(l);
    return seq;
  }

  private evict(l: { entries: MemEntry[]; maxSeq: number }): void {
    // Entry cap (drop oldest).
    while (l.entries.length > this.maxEntries) {
      l.entries.shift();
    }
    // Byte cap (base64 length is the on-the-wire size; the bound is soft
    // — a single oversized update may overshoot, recorded).
    let bytes = l.entries.reduce((n, e) => n + e.updateB64.length, 0);
    while (bytes > this.maxBytes && l.entries.length > 1) {
      bytes -= l.entries.shift()!.updateB64.length;
    }
  }

  async entriesAfter(roomKey: string, afterSeq: number): Promise<UpdateLogEntry[] | null> {
    const l = this.logs.get(roomKey);
    if (l === undefined) return afterSeq === 0 ? [] : null;
    if (l.entries.length === 0) return [];
    const oldest = l.entries[0]!.seq;
    // Gap: the client's next needed seq is no longer in the log.
    if (afterSeq + 1 < oldest) return null;
    return l.entries
      .filter((e) => e.seq > afterSeq)
      .map((e) => ({ seq: e.seq, updateB64: e.updateB64, pageVersion: e.pageVersion }));
  }

  async maxSeq(roomKey: string): Promise<number> {
    return this.logs.get(roomKey)?.maxSeq ?? 0;
  }

  async trimAfter(roomKey: string, keepSeq: number): Promise<void> {
    const l = this.logs.get(roomKey);
    if (l === undefined) return;
    l.entries = l.entries.filter((e) => e.seq > keepSeq);
  }

  async clear(roomKey: string): Promise<void> {
    this.logs.delete(roomKey);
  }

  /** Test/diagnostic: raw log size (entries). */
  size(roomKey: string): number {
    return this.logs.get(roomKey)?.entries.length ?? 0;
  }
}

// --- Redis implementation (production) ------------------------------------------

/**
 * The minimal Redis surface {@link RedisUpdateLog} needs. ioredis'
 * `Redis` satisfies it structurally (all methods are promise-based);
 * the integration tests inject a fake.
 */
export interface UpdateLogBackend {
  incr(key: string): Promise<number>;
  incrBy(key: string, delta: number): Promise<number>;
  get(key: string): Promise<string | null>;
  set(key: string, value: string): Promise<unknown>;
  rpush(key: string, value: string): Promise<number>;
  lrange(key: string, start: number, stop: number): Promise<string[]>;
  ltrim(key: string, start: number, stop: number): Promise<unknown>;
  llen(key: string): Promise<number>;
  del(...keys: string[]): Promise<number>;
}

const KEY = (roomKey: string): string => `realtime:updates:${roomKey}`;
const SEQ_KEY = (roomKey: string): string => `realtime:updates:${roomKey}:seq`;
const BYTES_KEY = (roomKey: string): string => `realtime:updates:${roomKey}:bytes`;

function parseEntry(raw: string): { seq: number; update: string; version: number } | null {
  try {
    const e = JSON.parse(raw) as { seq?: unknown; update?: unknown; version?: unknown };
    if (typeof e.seq !== "number" || typeof e.update !== "string") return null;
    return { seq: e.seq, update: e.update, version: typeof e.version === "number" ? e.version : 0 };
  } catch {
    return null;
  }
}

/**
 * Redis-backed update log. One JSON string per list element
 * (`{seq, update, version}`), one `RPUSH` per update; the list is
 * left-trimmed to the entry cap and (best-effort) the byte cap via a
 * stored byte counter.
 */
export class RedisUpdateLog implements UpdateLog {
  private readonly redis: UpdateLogBackend;
  private readonly maxEntries: number;
  private readonly maxBytes: number;

  constructor(opts: { redis: UpdateLogBackend; maxEntries?: number; maxBytes?: number }) {
    this.redis = opts.redis;
    this.maxEntries = opts.maxEntries ?? REALTIME_LIMITS.updateLogMaxEntries;
    this.maxBytes = opts.maxBytes ?? REALTIME_LIMITS.updateLogMaxBytes;
  }

  async append(roomKey: string, updateB64: string, pageVersion: number): Promise<number> {
    const seq = await this.redis.incr(SEQ_KEY(roomKey));
    const entry = JSON.stringify({ seq, update: updateB64, version: pageVersion });
    await this.redis.rpush(KEY(roomKey), entry);
    await this.redis.incrBy(BYTES_KEY(roomKey), entry.length);
    // Entry cap: keep the LAST maxEntries.
    const len = await this.redis.llen(KEY(roomKey));
    if (len > this.maxEntries) {
      await this.redis.ltrim(KEY(roomKey), len - this.maxEntries, -1);
      await this.remeasureBytes(roomKey);
    }
    // Byte cap (soft — a single giant update may overshoot; recorded).
    const bytes = Number((await this.redis.get(BYTES_KEY(roomKey))) ?? 0);
    if (bytes > this.maxBytes) {
      const len2 = await this.redis.llen(KEY(roomKey));
      if (len2 > 1) {
        await this.redis.ltrim(KEY(roomKey), 1, -1);
        await this.remeasureBytes(roomKey);
      }
    }
    return seq;
  }

  /** Re-sum the stored element lengths into the byte counter (post-trim). */
  private async remeasureBytes(roomKey: string): Promise<void> {
    const all = await this.redis.lrange(KEY(roomKey), 0, -1);
    const total = all.reduce((n, s) => n + s.length, 0);
    await this.redis.set(BYTES_KEY(roomKey), String(total));
  }

  async entriesAfter(roomKey: string, afterSeq: number): Promise<UpdateLogEntry[] | null> {
    const maxSeq = await this.maxSeq(roomKey);
    if (maxSeq === 0) return afterSeq === 0 ? [] : null;
    const all = await this.redis.lrange(KEY(roomKey), 0, -1);
    if (all.length === 0) return [];
    const first = parseEntry(all[0]!);
    if (first === null || afterSeq + 1 < first.seq) return null; // gap
    const out: UpdateLogEntry[] = [];
    for (const raw of all) {
      const e = parseEntry(raw);
      if (e !== null && e.seq > afterSeq) {
        out.push({ seq: e.seq, updateB64: e.update, pageVersion: e.version });
      }
    }
    return out;
  }

  async maxSeq(roomKey: string): Promise<number> {
    const v = await this.redis.get(SEQ_KEY(roomKey));
    return v === null ? 0 : Number(v);
  }

  async trimAfter(roomKey: string, keepSeq: number): Promise<void> {
    const all = await this.redis.lrange(KEY(roomKey), 0, -1);
    let drop = 0;
    for (const raw of all) {
      const e = parseEntry(raw);
      if (e === null || e.seq <= keepSeq) drop += 1;
      else break;
    }
    if (drop > 0) {
      await this.redis.ltrim(KEY(roomKey), drop, -1);
      await this.remeasureBytes(roomKey);
    }
  }

  async clear(roomKey: string): Promise<void> {
    await this.redis.del(KEY(roomKey), SEQ_KEY(roomKey), BYTES_KEY(roomKey));
  }
}