import { DEFAULT_REVISION_ID, MESSAGE_VERSION } from "../constants";
import { EventBus } from "../helpers/event_bus";
import { deepEquals, isDefined } from "../helpers/misc";
import { UuidGenerator } from "../helpers/uuid";
import { SelectiveHistory as RevisionLog } from "../history/selective_history";
import {
  Client,
  ClientId,
  ClientPosition,
  CollaborativeEvent,
} from "../types/collaborative/session";
import {
  ClientJoinedMessage,
  ClientLeftMessage,
  ClientMovedMessage,
  CollaborationMessage,
  RemoteRevisionMessage,
  RemoteRevisionsSquishedMessage,
  RevisionRedoneMessage,
  RevisionUndoneMessage,
  SnapshotCreatedMessage,
  StateUpdateMessage,
  TransportService,
} from "../types/collaborative/transport_service";
import { Command, CoreCommand } from "../types/commands";

import { HistoryChange } from "../types/history";
import { Lazy, UID } from "../types/misc";
import { WorkbookData } from "../types/workbook_data";
import { ICommandSquisher } from "./command_squisher";
import { transformAll } from "./ot/ot";
import { Revision } from "./revisions";

export class ClientDisconnectedError extends Error {}

export class Session extends EventBus<CollaborativeEvent> {
  /**
   * Positions of the others client.
   */
  private clients: Record<ClientId, Client | undefined> = {};
  private clientId: ClientId = "local";

  private pendingMessages: StateUpdateMessage[] = [];

  /**
   * Stored position of the client, if session.move is called while the client is not in a session.
   * Will be used to send the position to the server when the client joins.
   */
  private awaitingClientPosition: ClientPosition | undefined;

  private waitingAck: boolean = false;
  /**
   * Flag used to block all commands when an undo or redo is triggered, until
   * it is accepted on the server
   */
  private waitingUndoRedoAck: boolean = false;
  private isReplayingInitialRevisions = false;

  private processedRevisions: Set<UID> = new Set();
  private lastRevisionMessage:
    | RevisionUndoneMessage
    | RevisionRedoneMessage
    | RemoteRevisionMessage
    | RemoteRevisionsSquishedMessage
    | SnapshotCreatedMessage
    | undefined = undefined;

  private lastLocalOperation: Revision | undefined;
  /**
   * Manages the collaboration between multiple users on the same spreadsheet.
   * It can forward local state changes to other users to ensure they all eventually
   * reach the same state.
   * It also manages the positions of each client in the spreadsheet to provide
   * a visual indication of what other users are doing in the spreadsheet.
   *
   * @param revisions
   * @param transportService communication channel used to send and receive messages
   * between all connected clients
   * @param serverRevisionId
   * @param commandSquisher used to squish and unsquish commands to reduce the size of messages sent to the server
   * @param shouldVerifySquish either the commands should be unsquished and verified before being sent to the server
   */
  constructor(
    private revisions: RevisionLog<Revision>,
    private transportService: TransportService<CollaborationMessage>,
    private serverRevisionId: UID = DEFAULT_REVISION_ID,
    private commandSquisher: ICommandSquisher,
    private shouldVerifySquish: boolean = true
  ) {
    super();
  }

  canApplyOptimisticUpdate() {
    return !this.waitingUndoRedoAck;
  }

  /**
   * Add a new revision to the collaborative session.
   * It will be transmitted to all other connected clients.
   */
  save(rootCommand: Command, commands: readonly CoreCommand[], changes: HistoryChange[]) {
    if (!commands.length || !changes.length || !this.canApplyOptimisticUpdate()) {
      return;
    }
    const revision = new Revision(
      UuidGenerator.uuidv4(),
      this.clientId,
      commands,
      rootCommand,
      changes,
      Date.now()
    );
    this.revisions.append(revision.id, revision);
    // REQUEST_REDO just repeats the last operation, the
    // last operation is still the same and should not change.
    if (rootCommand.type !== "REQUEST_REDO") {
      this.lastLocalOperation = revision;
    }
    this.trigger("new-local-state-update", { id: revision.id });
    this.sendUpdateMessage({
      type: "REMOTE_REVISION",
      version: MESSAGE_VERSION,
      serverRevisionId: this.serverRevisionId,
      nextRevisionId: revision.id,
      clientId: revision.clientId,
      commands: revision.commands,
      rootCommandType: rootCommand.type,
    });
  }

  undo(revisionId: UID) {
    this.waitingUndoRedoAck = true;
    this.sendUpdateMessage({
      type: "REVISION_UNDONE",
      version: MESSAGE_VERSION,
      serverRevisionId: this.serverRevisionId,
      nextRevisionId: UuidGenerator.uuidv4(),
      undoneRevisionId: revisionId,
    });
  }

  redo(revisionId: UID) {
    this.waitingUndoRedoAck = true;
    this.sendUpdateMessage({
      type: "REVISION_REDONE",
      version: MESSAGE_VERSION,
      serverRevisionId: this.serverRevisionId,
      nextRevisionId: UuidGenerator.uuidv4(),
      redoneRevisionId: revisionId,
    });
  }

  join(client?: Client) {
    if (client) {
      this.clients[client.id] = client;
      this.clientId = client.id;
    } else {
      this.clients["local"] = { id: "local", name: "local" };
      this.clientId = "local";
    }
    this.transportService.onNewMessage(this.clientId, this.onMessageReceived.bind(this));
    if (this.awaitingClientPosition) {
      this.move(this.awaitingClientPosition);
      this.awaitingClientPosition = undefined;
    }
  }

  loadInitialMessages(messages: StateUpdateMessage[]) {
    const start = performance.now();
    const numberOfCommands = messages.reduce(
      (acc, message) => acc + (message.type === "REMOTE_REVISION" ? message.commands.length : 1),
      0
    );
    this.isReplayingInitialRevisions = true;
    for (const message of messages) {
      this.onMessageReceived(message);
    }
    this.isReplayingInitialRevisions = false;
    console.debug("Replayed", numberOfCommands, "commands in", performance.now() - start, "ms");
  }

  /**
   * Notify the server that the user client left the collaborative session
   */
  async leave(data?: Lazy<WorkbookData>) {
    if (
      data &&
      Object.keys(this.clients).length === 1 &&
      this.lastRevisionMessage &&
      this.lastRevisionMessage?.type !== "SNAPSHOT_CREATED"
    ) {
      await this.snapshot(data());
    }
    delete this.clients[this.clientId];
    this.transportService.leave(this.clientId);
    await this.sendToTransport({
      type: "CLIENT_LEFT",
      clientId: this.clientId,
      version: MESSAGE_VERSION,
    });
  }

  /**
   * Send a snapshot of the spreadsheet to the collaboration server
   */
  async snapshot(data: WorkbookData) {
    if (this.pendingMessages.length !== 0) {
      return;
    }
    const snapshotId = UuidGenerator.uuidv4();
    await this.sendToTransport({
      type: "SNAPSHOT",
      nextRevisionId: snapshotId,
      serverRevisionId: this.serverRevisionId,
      data: { ...data, revisionId: snapshotId },
      version: MESSAGE_VERSION,
    });
  }

  getCurrentClient(): Client {
    return this.getClient(this.clientId);
  }

  getClient(clientId: ClientId): Client {
    const client = this.clients[clientId];
    if (!client) {
      throw new ClientDisconnectedError("The client left the session");
    }
    return client;
  }

  getConnectedClients(): Set<Client> {
    return new Set(Object.values(this.clients).filter(isDefined));
  }

  getRevisionId(): UID {
    return this.serverRevisionId;
  }

  isFullySynchronized(): boolean {
    return this.pendingMessages.length === 0;
  }

  /**
   * Get the last local revision
   * */
  getLastLocalNonEmptyRevision(): Revision | undefined {
    return this.lastLocalOperation;
  }

  /**
   * Notify that the position of the client has changed
   */
  move(position: ClientPosition) {
    // this method could be called before the client joins the session, or after he left (because of the debounce)
    if (!this.clients[this.clientId]) {
      this.awaitingClientPosition = position;
      return;
    }
    const currentPosition = this.clients[this.clientId]?.position;
    if (
      currentPosition?.col === position.col &&
      currentPosition.row === position.row &&
      currentPosition.sheetId === position.sheetId
    ) {
      return;
    }
    const type = currentPosition ? "CLIENT_MOVED" : "CLIENT_JOINED";
    const client = this.getCurrentClient();
    this.clients[this.clientId] = { ...client, position };
    void this.sendToTransport({
      type,
      version: MESSAGE_VERSION,
      client: { ...client, position },
    }).then(() => {
      if (this.pendingMessages.length > 0 && !this.waitingAck) {
        this.sendPendingMessage();
      }
    });
  }

  /**
   * Handles messages received from other clients in the collaborative
   * session.
   */
  private onMessageReceived(message: CollaborationMessage) {
    if (this.isAlreadyProcessed(message)) {
      return;
    }
    if (this.isWrongServerRevisionId(message)) {
      this.trigger("unexpected-revision-id");
      return;
    }
    switch (message.type) {
      case "CLIENT_MOVED":
        this.onClientMoved(message);
        break;
      case "CLIENT_JOINED":
        this.onClientJoined(message);
        break;
      case "CLIENT_LEFT":
        this.onClientLeft(message);
        break;
      case "REVISION_REDONE": {
        this.revisions.redo(
          message.redoneRevisionId,
          message.nextRevisionId,
          message.serverRevisionId
        );
        this.trigger("revision-redone", {
          revisionId: message.redoneRevisionId,
          commands: this.revisions.get(message.redoneRevisionId).commands,
        });
        break;
      }
      case "REVISION_UNDONE":
        this.revisions.undo(
          message.undoneRevisionId,
          message.nextRevisionId,
          message.serverRevisionId
        );

        this.trigger("revision-undone", {
          revisionId: message.undoneRevisionId,
          commands: this.revisions.get(message.undoneRevisionId).commands,
        });
        break;
      case "REMOTE_REVISION":
        const { clientId, timestamp } = message;
        const unsquishedCommands = this.commandSquisher.unsquish(message.commands);
        const revision = new Revision(
          message.nextRevisionId,
          clientId,
          unsquishedCommands,
          undefined,
          undefined,
          timestamp
        );
        if (revision.clientId !== this.clientId) {
          this.revisions.insert(revision.id, revision, message.serverRevisionId);
          const pendingCommands = this.pendingMessages
            .filter((msg) => msg.type === "REMOTE_REVISION")
            .map((msg) => (msg as RemoteRevisionMessage).commands)
            .flat();
          this.trigger("remote-revision-received", {
            commands: transformAll(unsquishedCommands, pendingCommands),
          });
        }
        break;
      case "SNAPSHOT_CREATED": {
        const revision = new Revision(
          message.nextRevisionId,
          "server",
          [],
          undefined,
          undefined,
          Date.now()
        );
        this.revisions.insert(revision.id, revision, message.serverRevisionId);
        this.dropPendingHistoryMessages();
        this.trigger("snapshot");
        this.lastLocalOperation = undefined;
        break;
      }
    }
    this.acknowledge(message);
    this.trigger("collaborative-event-received");
  }

  private onClientMoved(message: ClientMovedMessage) {
    if (message.client.id !== this.clientId) {
      this.clients[message.client.id] = message.client;
    }
  }

  /**
   * Register the new client and send your
   * own position back.
   */
  private onClientJoined(message: ClientJoinedMessage) {
    if (message.client.id !== this.clientId) {
      this.clients[message.client.id] = message.client;
      const client = this.clients[this.clientId];
      if (client) {
        const { position } = client;
        if (position) {
          void this.sendToTransport({
            type: "CLIENT_MOVED",
            version: MESSAGE_VERSION,
            client: { ...client, position },
          });
        }
      }
    }
  }

  private onClientLeft(message: ClientLeftMessage) {
    if (message.clientId !== this.clientId) {
      delete this.clients[message.clientId];
    }
  }

  private sendUpdateMessage(message: StateUpdateMessage) {
    this.pendingMessages.push(message);
    if (this.waitingAck) {
      return;
    }
    this.sendPendingMessage();
  }

  private async sendToTransport(message: CollaborationMessage) {
    // wrap in an async function to ensure it returns a promise
    return this.transportService.sendMessage(message);
  }

  /**
   * Send the next pending message
   */
  private sendPendingMessage() {
    let message = this.pendingMessages[0];
    if (!message) {
      return;
    }
    if (message.type === "REMOTE_REVISION") {
      let revision = this.revisions.get(message.nextRevisionId);
      if (revision.commands.length === 0) {
        /**
         * The command is empty, we have to rebase all the next local revisions
         * to avoid issues with undo/redo
         */
        this.revisions.rebase(revision.id);
        revision = this.revisions.get(message.nextRevisionId);
      }

      const squishedCommands = this.commandSquisher.squish(revision.commands);
      const containsSquishedCommands = squishedCommands.some(
        (x) => x.type === "SQUISHED_UPDATE_CELL"
      );
      if (!containsSquishedCommands) {
        // if nothing was squished, safely keep the revision.commands, as the squishing process does change b.e. the casing of references
        message = {
          ...message,
          commands: revision.commands,
        };
      } else {
        const sortedCommands = [
          ...this.commandSquisher.collectConsecutiveUpdateCellCommands(revision.commands),
        ].flat();
        if (
          this.shouldVerifySquish &&
          !deepEquals(sortedCommands, this.commandSquisher.unsquish(squishedCommands))
        ) {
          message = {
            ...message,
            commands: revision.commands,
            squishedFailed: true,
            rootCommandType: revision.rootCommand?.type,
          };
        } else {
          message = {
            ...message,
            commands: squishedCommands,
          };
        }
      }
    }
    if (this.isReplayingInitialRevisions) {
      throw new Error(`Trying to send a new revision while replaying initial revision. This can lead to endless dispatches every time the spreadsheet is open.
      ${JSON.stringify(message)}`);
    }
    this.waitingAck = true;
    this.sendToTransport({
      ...message,
      serverRevisionId: this.serverRevisionId,
    }).catch((e: Error) => {
      if (!(e instanceof ClientDisconnectedError)) {
        throw e.cause || e;
      }
      this.waitingAck = false;
    });
  }

  private acknowledge(message: CollaborationMessage) {
    if (message.type === "REVISION_UNDONE" || message.type === "REVISION_REDONE") {
      this.waitingUndoRedoAck = false;
    }
    switch (message.type) {
      case "REMOTE_REVISION":
      case "REVISION_REDONE":
      case "SNAPSHOT_CREATED":
        this.waitingAck = false;
        this.pendingMessages = this.pendingMessages.filter(
          (msg) => msg.nextRevisionId !== message.nextRevisionId
        );
        this.serverRevisionId = message.nextRevisionId;
        this.processedRevisions.add(message.nextRevisionId);
        this.lastRevisionMessage = message;
        this.sendPendingMessage();
        break;
      case "REVISION_UNDONE": {
        this.waitingAck = false;
        this.pendingMessages = this.pendingMessages.filter(
          (msg) => msg.nextRevisionId !== message.nextRevisionId
        );
        const firstPendingRevisionId = this.pendingMessages.findIndex(
          (message): message is RemoteRevisionMessage => message.type === "REMOTE_REVISION"
        );
        if (firstPendingRevisionId !== -1) {
          /**
           * Some revisions undergo transformations that may cause issues with
           * undo/redo if the transformation is destructive (we don't get back
           * the original command by transforming it with the inverse).
           * To prevent these problems, we must rebase all subsequent local
           * revisions.
           */
          this.revisions.rebase(this.pendingMessages[firstPendingRevisionId].nextRevisionId);
        }
        this.serverRevisionId = message.nextRevisionId;
        this.processedRevisions.add(message.nextRevisionId);
        this.sendPendingMessage();
        break;
      }
    }
  }

  private isAlreadyProcessed(message: CollaborationMessage): boolean {
    if (message.type === "CLIENT_MOVED" && message.client.id === this.clientId) {
      return true;
    }
    switch (message.type) {
      case "REMOTE_REVISION":
      case "REVISION_REDONE":
      case "REVISION_UNDONE":
      case "SNAPSHOT_CREATED":
        return this.processedRevisions.has(message.nextRevisionId);
      default:
        return false;
    }
  }

  isWrongServerRevisionId(message: CollaborationMessage) {
    switch (message.type) {
      case "REMOTE_REVISION":
      case "REVISION_REDONE":
      case "REVISION_UNDONE":
      case "SNAPSHOT_CREATED":
        return message.serverRevisionId !== this.serverRevisionId;
      default:
        return false;
    }
  }

  private dropPendingHistoryMessages() {
    this.waitingUndoRedoAck = false;
    this.pendingMessages = this.pendingMessages.filter(
      ({ type }) => type !== "REVISION_REDONE" && type !== "REVISION_UNDONE"
    );
  }
}