import { ChannelSync } from "../channel-sync";
import { MessageStore } from "../message-store";
import { MemoryOutboxStorage, Outbox } from "../outbox";
import {
  assertTotalOrder,
  channelUnreadCount,
  channelView,
  mentionBadgeFromRows,
  reactionsConverged,
  seqOrder,
  threadUnreadCount,
} from "../invariants";
import { CommitCrash, EmitCrash, HttpFault } from "./sim";
import { SimServer } from "./server";
import type { Message } from "@slackwsh/contracts";
export class SimClient {
  readonly store = new MessageStore({ messagesPerChannel: 10_000, maxChannels: 10 });
  readonly outboxStorage = new MemoryOutboxStorage();
  readonly outbox: Outbox;
  sync: ChannelSync;
  connected = true;
  socketId: string;
  extraSocketId: string | null = null;
  /** Last presence snapshot this client rendered. */
  readonly presence = new Map<number, "active" | "offline">();
  /** Server-authoritative unread, discarded and rehydrated on reconnect (§5.3). */
  displayedUnread = 0;
  displayedMentionBadge = 0;
  displayedThreadUnread = new Map<number, number>();
  private nextClientMsg = 0;
  private socketSeq = 0;

  constructor(
    readonly userId: number,
    private readonly server: SimServer,
    private readonly uuidOffset: number,
  ) {
    this.socketId = `s-${userId}-0`;
    this.socketSeq = 0;
    this.outbox = new Outbox(this.outboxStorage, (entry) => this.flushEntry(entry));
    this.sync = this.newSync(0);
    for (const id of server.memberIds()) this.presence.set(id, "offline");
  }

  private newSync(cursorSeq: number): ChannelSync {
    return new ChannelSync(
      String(this.server.channelId),
      this.store,
      (opts) => Promise.resolve(this.server.scrollback(opts)),
      cursorSeq,
    );
  }

  async open(): Promise<void> {
    this.server.connectSocket(this.userId, this.socketId);
    await this.catchUp();
    this.rehydrateReadState();
  }

  async catchUp(): Promise<void> {
    for (let attempt = 0; attempt < 5; attempt++) {
      this.sync = this.newSync(this.sync.getCursor());
      try {
        await this.sync.start();
        this.rehydrateReadState();
        return;
      } catch (err) {
        if (!(err instanceof HttpFault) || attempt === 4) throw err;
      }
    }
  }

  /** Channel re-open: drop the cache and load the latest window from the server. */
  async fullReconcile(): Promise<void> {
    this.store.clearChannel(String(this.server.channelId));
    this.sync = this.newSync(0);
    for (let attempt = 0; attempt < 5; attempt++) {
      try {
        await this.sync.start();
        this.rehydrateReadState();
        return;
      } catch (err) {
        if (!(err instanceof HttpFault) || attempt === 4) throw err;
        this.sync = this.newSync(0);
      }
    }
  }

  rehydrateReadState(): void {
    const log = this.server.allMessages();
    const lastRead = this.server.lastReadSeq.get(this.userId) ?? 0;
    this.displayedUnread = channelUnreadCount(log, lastRead, this.userId);
    this.displayedMentionBadge = mentionBadgeFromRows(
      log,
      this.server.mentions,
      this.userId,
      lastRead,
      this.server.memberIds(),
    );
    this.displayedThreadUnread.clear();
    for (const rootId of this.server.threadFollowers.keys()) {
      const cursor = this.server.lastReadReplySeq.get(`${this.userId}:${rootId}`) ?? 0;
      this.displayedThreadUnread.set(rootId, threadUnreadCount(log, rootId, cursor, this.userId));
    }
  }

  disconnect(graceful: boolean): void {
    this.connected = false;
    if (graceful) this.server.disconnectSocket(this.userId, this.socketId);
    else this.server.dropSocketUngraceful(this.socketId);
  }

  async reconnect(): Promise<void> {
    this.socketSeq += 1;
    this.socketId = `s-${this.userId}-${this.socketSeq}`;
    this.connected = true;
    this.server.connectSocket(this.userId, this.socketId);
    await this.outbox.replay();
    await this.catchUp();
  }

  addExtraSocket(): void {
    this.extraSocketId = `s-${this.userId}-extra`;
    this.server.connectSocket(this.userId, this.extraSocketId);
  }

  dropExtraSocket(graceful: boolean): void {
    if (!this.extraSocketId) return;
    if (graceful) this.server.disconnectSocket(this.userId, this.extraSocketId);
    else this.server.dropSocketUngraceful(this.extraSocketId);
    this.extraSocketId = null;
  }

  heartbeat(): void {
    if (this.connected) {
      this.server.heartbeat(this.socketId);
      if (this.extraSocketId) this.server.heartbeat(this.extraSocketId);
    }
  }

  async onEvent(event: { type: string; messageId?: number; seq?: number; userId?: number; status?: "active" | "offline" }): Promise<void> {
    if (!this.connected) return;
    try {
      switch (event.type) {
        case "message:created": {
          const message = this.server.messages.get(event.messageId!);
          if (!message) return;
          await this.sync.onLiveMessage(this.server.wire(message, this.userId));
          break;
        }
        case "message:edited": {
          const message = this.server.messages.get(event.messageId!);
          if (!message) return;
          this.sync.onMessageEdited(this.server.wire(message, this.userId));
          break;
        }
        case "message:deleted":
          this.sync.onMessageDeleted(event.messageId!, event.seq ?? 0);
          break;
        case "reaction:changed": {
          const message = this.server.messages.get(event.messageId!);
          if (!message) return;
          this.sync.onMessageEdited(this.server.wire(message, this.userId));
          break;
        }
        case "presence:changed":
          if (event.userId != null && event.status) this.presence.set(event.userId, event.status);
          break;
        default:
          break;
      }
    } catch (err) {
      if (err instanceof HttpFault) return;
      throw err;
    }
  }

  async send(text: string, opts: { parentId?: number | null; isBroadcast?: boolean } = {}): Promise<void> {
    this.nextClientMsg += 1;
    const clientMsgId = uuidFrom(this.uuidOffset + this.userId * 1_000_000 + this.nextClientMsg);
    await this.outbox.enqueue({
      clientMsgId,
      channelId: String(this.server.channelId),
      text,
      parentId: opts.parentId != null ? String(opts.parentId) : null,
      isBroadcast: opts.isBroadcast,
      createdAt: new Date(this.nextClientMsg).toISOString(),
    });
  }

  private async flushEntry(entry: { clientMsgId: string; text: string; parentId?: string | null; isBroadcast?: boolean }): Promise<void> {
    const parentId = entry.parentId != null && entry.parentId !== "" ? Number(entry.parentId) : null;
    try {
      const message = this.server.send({
        userId: this.userId,
        clientMsgId: entry.clientMsgId,
        text: entry.text,
        parentId,
        isBroadcast: entry.isBroadcast,
      });
      this.store.upsert(String(this.server.channelId), this.server.wire(message, this.userId));
    } catch (err) {
      if (err instanceof CommitCrash || err instanceof EmitCrash) throw err;
      throw err;
    }
  }

  ackUpToLatest(): void {
    const view = channelView(this.sync.getMessages());
    const latest = view[view.length - 1];
    if (!latest) return;
    this.server.ackRead(this.userId, latest.seq);
    this.rehydrateReadState();
  }

  markLatestUnread(): void {
    const view = channelView(this.sync.getMessages());
    const latest = view[view.length - 1];
    if (!latest) return;
    this.server.markUnread(this.userId, latest.seq);
    this.rehydrateReadState();
  }

  ackThreads(): void {
    for (const rootId of this.server.threadFollowers.keys()) {
      const replies = this.server.allMessages().filter((m) => Number(m.parentId) === rootId);
      const latest = replies[replies.length - 1];
      if (latest) this.server.ackThreadRead(this.userId, rootId, latest.seq);
    }
    this.rehydrateReadState();
  }

  renderedChannel(): Message[] {
    return channelView(this.sync.getMessages()).filter((m) => m.deletedAt == null);
  }
}

export function uuidFrom(n: number): string {
  return `00000000-0000-4000-8000-${n.toString(16).padStart(12, "0")}`;
}

export function assertInvariants(server: SimServer, clients: SimClient[]): void {
  const log = server.allMessages();
  const live = clients.filter((c) => c.connected);

  // I1 — every live client matches the server's latest channel-view page.
  const page = channelView(log).filter((m) => m.deletedAt == null).slice(-50);
  const serverView = seqOrder(page);
  for (const client of live.filter((c) => c.sync.getState() === "live")) {
    const view = seqOrder(client.renderedChannel());
    if (view.join(",") !== serverView.join(",")) {
      throw new Error(`I1: client ${client.userId} view ${view.join(",")} != server ${serverView.join(",")}`);
    }
  }
  for (const client of live) assertTotalOrder(client.sync.getMessages());

  const byClientMsg = new Map<string, number>();
  for (const message of log) {
    const prev = byClientMsg.get(message.clientMsgId);
    if (prev != null && prev !== message.id) throw new Error(`I3: duplicate client_msg_id ${message.clientMsgId}`);
    byClientMsg.set(message.clientMsgId, message.id);
  }

  // I3 — unique (channel, client_msg_id) already encoded by SimServer.byClientMsg
  if (server.byClientMsg.size !== new Set(log.map((m) => m.clientMsgId)).size) {
    throw new Error("I3: client_msg_id index drifted from the log");
  }

  // I4 — cursors are non-negative; GREATEST/LEAST applied only via clampReadSeq in server.
  for (const userId of server.memberIds()) {
    const cursor = server.lastReadSeq.get(userId) ?? 0;
    if (cursor < 0) throw new Error("I4: last_read_seq went negative");
    if (cursor > server.lastSeq) throw new Error("I4: last_read_seq ahead of log");
  }

  // I5 / I5b — displayed unread after rehydrate equals the spec formula.
  for (const client of clients) {
    const lastRead = server.lastReadSeq.get(client.userId) ?? 0;
    const expected = channelUnreadCount(log, lastRead, client.userId);
    if (client.displayedUnread !== expected) {
      throw new Error(`I5: user ${client.userId} displayed ${client.displayedUnread}, spec ${expected}`);
    }
    // Thread replies that are not broadcasts must not contribute.
    const hiddenReplies = log.filter(
      (m) =>
        m.parentId != null &&
        !m.isBroadcast &&
        m.seq > lastRead &&
        Number(m.authorId) !== client.userId &&
        m.deletedAt == null,
    ).length;
    const naive = log.filter(
      (m) => m.seq > lastRead && Number(m.authorId) !== client.userId && m.deletedAt == null,
    ).length;
    if (expected !== naive - hiddenReplies) {
      throw new Error(`I5: spec ${expected} != naive ${naive} minus ${hiddenReplies} hidden replies`);
    }
    for (const [rootId, displayed] of client.displayedThreadUnread) {
      const cursor = server.lastReadReplySeq.get(`${client.userId}:${rootId}`) ?? 0;
      const spec = threadUnreadCount(log, rootId, cursor, client.userId);
      if (displayed !== spec) throw new Error(`I5b: thread ${rootId} displayed ${displayed}, spec ${spec}`);
    }
  }

  // I6 — badges follow persisted rows, not current text.
  for (const client of clients) {
    const lastRead = server.lastReadSeq.get(client.userId) ?? 0;
    const fromRows = mentionBadgeFromRows(log, server.mentions, client.userId, lastRead, server.memberIds());
    if (client.displayedMentionBadge !== fromRows) {
      throw new Error(`I6: badge ${client.displayedMentionBadge} != rows ${fromRows}`);
    }
  }

  // I8 — live clients that have the message match server revision and reactions.
  for (const client of live) {
    if (client.sync.getState() !== "live") continue;
    for (const message of client.sync.getMessages()) {
      const authoritative = server.messages.get(Number(message.id));
      if (!authoritative) continue;
      if (message.revision > authoritative.revision) {
        throw new Error(`I8: client revision ${message.revision} ahead of server ${authoritative.revision}`);
      }
      if (message.revision === authoritative.revision) {
        if (message.text !== authoritative.text) throw new Error("I8: same revision, different text");
        const deleted = Boolean(message.deletedAt);
        const serverDeleted = Boolean(authoritative.deletedAt);
        if (deleted !== serverDeleted) throw new Error("I8: delete bit diverged at same revision");
        if (!reactionsConverged(message.reactions, authoritative.reactions)) {
          throw new Error("I8: reactions diverged at same revision");
        }
      }
    }
  }

  // I9 — live sockets never shown offline; zero sockets shown offline after TTL+sweep.
  for (const userId of server.memberIds()) {
    const liveCount = server.liveSocketCount(userId);
    const shown = server.displayedPresence.get(userId) ?? "offline";
    if (liveCount > 0 && shown === "offline") {
      throw new Error(`I9: user ${userId} has ${liveCount} live socket(s) but is shown offline`);
    }
  }
}