import type { Message } from "@slackwsh/contracts";
import { MessageStore } from "./message-store";

export type ChannelSyncState = "catchingUp" | "live";

export interface ScrollbackFetcher {
  (opts: { afterSeq?: number; beforeSeq?: number; limit: number }): Promise<Message[]>;
}

/**
 * The catch-up/live boundary (§5.3, invariant I7): "On join the client
 * sends its cursor and enters catchingUp. Live events arriving in that
 * window are buffered client-side, not rendered. When backfill completes
 * the buffer is merged by seq, deduplicated, and flushed; only then does
 * the channel enter live." Buffering client-side (not server-side) keeps
 * gateway nodes stateless with respect to catch-up and survives a gateway
 * crash mid-backfill — see ARCHITECTURE §5.3.
 *
 * Also owns gap detection: a live message arriving with `seq` more than one
 * past `expectedSeq` is a gap. The client never renders across a gap — it
 * re-fetches instead. Silently rendering past a gap makes order corruption
 * permanent, which is the whole reason this class exists rather than just
 * trusting whatever socket.io delivers.
 */
export class ChannelSync {
  private static readonly INITIAL_PAGE_SIZE = 50;
  private state: ChannelSyncState = "catchingUp";
  private liveBuffer: Message[] = [];
  private expectedSeq: number;
  private resyncing = false;
  private olderHistoryExhausted = false;

  constructor(
    private readonly channelId: string,
    private readonly store: MessageStore,
    private readonly fetchScrollback: ScrollbackFetcher,
    cursorSeq: number,
  ) {
    this.expectedSeq = cursorSeq;
  }

  getState(): ChannelSyncState {
    return this.state;
  }

  /** Runs the backfill, merges anything buffered while it was in flight,
   * then flips to `live`. Call once per channel-open. */
  async start(): Promise<void> {
    this.state = "catchingUp";
    const isInitialLoad = this.expectedSeq === 0;
    const backfill = await this.fetchScrollback(
      isInitialLoad
        ? { limit: ChannelSync.INITIAL_PAGE_SIZE }
        : { afterSeq: this.expectedSeq, limit: 200 },
    );
    if (isInitialLoad) {
      this.olderHistoryExhausted = backfill.length < ChannelSync.INITIAL_PAGE_SIZE;
    }
    for (const message of backfill) {
      this.store.upsert(this.channelId, message);
      this.advanceExpectedSeq(message);
    }

    // Merge-by-seq, dedup, flush: buffered live events with seq already
    // covered by backfill are no-ops (MessageStore.upsert is idempotent by
    // id); anything genuinely newer than the backfill's tail gets applied
    // in order, exactly as if it had arrived live after catch-up finished.
    const buffered = [...this.liveBuffer].sort((a, b) => a.seq - b.seq);
    this.liveBuffer = [];
    for (const message of buffered) {
      if (message.seq <= this.expectedSeq) continue;
      this.store.upsert(this.channelId, message);
      this.advanceExpectedSeq(message);
    }

    this.state = "live";
  }

  /** Loads the page immediately before the oldest cached message. Older
   * history never moves the live cursor backwards; it is merged into the
   * same idempotent, seq-sorted store and can therefore overlap safely with
   * realtime events or a retried request. */
  async loadOlder(limit = ChannelSync.INITIAL_PAGE_SIZE): Promise<Message[]> {
    if (this.olderHistoryExhausted) return [];

    const oldest = this.store.getMessages(this.channelId)[0];
    if (!oldest) {
      this.olderHistoryExhausted = true;
      return [];
    }

    const page = await this.fetchScrollback({ beforeSeq: oldest.seq, limit });
    for (const message of page) {
      this.store.upsert(this.channelId, message);
    }
    if (page.length < limit) this.olderHistoryExhausted = true;
    return page;
  }

  hasOlderMessages(): boolean {
    return !this.olderHistoryExhausted;
  }

  /** Feed a `message:created` event from the socket. */
  async onLiveMessage(message: Message): Promise<void> {
    if (this.state === "catchingUp") {
      this.liveBuffer.push(message);
      return;
    }

    if (message.seq > this.expectedSeq + 1 && !this.resyncing) {
      await this.resync(message.seq);
    }

    // Applied unconditionally after any resync: the resync fetch may not
    // itself have returned this exact message (e.g. it raced the write),
    // but we already have it in hand, so there's no reason to drop it —
    // upsert's own id-dedup makes this safe even if resync did include it.
    if (message.seq > this.expectedSeq) {
      this.store.upsert(this.channelId, message);
      this.advanceExpectedSeq(message);
    }
  }

  onMessageEdited(message: Message): void {
    this.store.upsert(this.channelId, message);
  }

  onMessageDeleted(messageId: string | number, seq: number): void {
    this.store.applyDelete(this.channelId, messageId, seq);
  }

  getMessages(): Message[] {
    return this.store.getMessages(this.channelId);
  }

  getCursor(): number {
    return this.expectedSeq;
  }

  private advanceExpectedSeq(message: Message): void {
    if (message.seq > this.expectedSeq) this.expectedSeq = message.seq;
  }

  /** A detected gap: re-fetch rather than render across it. Re-entrant calls
   * while a resync is already in flight are dropped — the in-flight fetch
   * will already cover whatever gap triggered them. */
  private async resync(upToSeq: number): Promise<void> {
    this.resyncing = true;
    try {
      // API ScrollQuery.limit max is 200 — page until the gap is closed.
      while (this.expectedSeq < upToSeq) {
        const gap = await this.fetchScrollback({
          afterSeq: this.expectedSeq,
          beforeSeq: upToSeq + 1,
          limit: 200,
        });
        if (gap.length === 0) break;
        for (const message of gap) {
          this.store.upsert(this.channelId, message);
          this.advanceExpectedSeq(message);
        }
        if (gap.length < 200) break;
      }
    } finally {
      this.resyncing = false;
    }
  }
}