import {db} from '@db/instance';
import {
  calculateTextHeight,
  mapReactionToStore,
  getTextWithHighlights,
  getMessagesWithLocalData,
  mapMembershipsToDB,
  prefetchMessageImages,
} from '@modules/chats/utils';
import {
  reduceToLatestMessages,
  deleteOrphanedMemberships,
} from '@modules/chats/db';
import {fileHandler} from '@modules/chats/instances/FileHandler';
import {Conversation, Membership, Message} from '@modules/chats/interfaces';
import {sentryLogs, wait, BlockingLatch} from '@shared/utils';

import {ConversationsUpdaterService} from './ConversationsUpdaterService';
import {PendingAssets} from '../PendingAssets';
import {UnreadsUpdaterService} from './UnreadsUpdaterService';
import {PendingMessagesTracker} from '../PendingMessagesTracker';
import {ReactionsUpdaterService} from './ReactionsUpdaterService';
import {MembershipUpdaterService} from './MembershipUpdaterService';
import {UsersUpdaterService} from './UsersUpdaterService';
import {AssetsUpdaterService} from './AssetsUpdaterService';
import {MessagesUpdaterService} from './MessagesUpdaterService';
import {PinnedMessagesUpdaterService} from './PinnedMessagesUpdaterService';

class OrchestratorUpdaterServiceClass {
  private static readonly DB_FLUSH_DELAY_MS = 300;

  /**
   * Run the non-DB hydration steps that every bulk "latest messages"
   * path needs after the DB write: reaction store, attachment store.
   *
   * Shared by `saveRecentConversations` (top-N recents) and
   * `RecentsHandler.fetchAllConversations` (entire conversation list).
   * Keeping it here as a single source of truth so future post-DB hooks
   * (e.g. pinned, pending assets) only need to be added once.
   */
  processLatestsPostDB(
    messages: Message[],
    {processReactions = true}: {processReactions?: boolean} = {},
  ): void {
    if (!messages.length) return;
    if (processReactions) {
      ReactionsUpdaterService.processWithMessages(messages);
    }
    AssetsUpdaterService.processAttachments(messages);
  }

  //-------- Conversations -------------

  async markConversationAsRead(
    id: string,
    latestMessageTs: string,
    version?: number,
  ) {
    UnreadsUpdaterService.markAsRead(id, latestMessageTs, version);
    await ConversationsUpdaterService.markAsRead(id);
  }

  /**
   * Save recent conversations to the store and the DB.
   */
  async saveRecentConversations(
    data: {
      group_conversation: Conversation;
      latest_messages: Message[];
    }[],
  ) {
    const conversationsPromises: Promise<void>[] = [];
    const messagesPromises: Promise<void>[] = [];
    const dataWithHeight: {
      group_conversation: Conversation;
      latest_messages: Message[];
    }[] = [];

    for (const item of data) {
      try {
        // From recents, messages are in descending order
        const messagesAscending = item.latest_messages.reverse();
        const messages = await getMessagesWithLocalData(messagesAscending);

        dataWithHeight.push({
          group_conversation: item.group_conversation,
          latest_messages: messages,
        });

        // Save conversation to the DB
        conversationsPromises.push(
          ConversationsUpdaterService.saveNewConversation(
            item.group_conversation,
            {updateStore: false},
          ),
        );

        // Save messages to the DB
        messagesPromises.push(
          this.saveMessages(messages, {withLatestUpdate: false}),
        );

        // Hydrate non-DB stores (reactions, attachments)
        this.processLatestsPostDB(messages);
      } catch (error) {
        sentryLogs.error({
          message: 'Error saving recent conversations',
          module: 'CHATS',
          component: 'OrchestratorUpdaterService.saveRecentConversations',
          error,
          data: {
            channel: item.group_conversation.id,
            messagesRange: `from ${item.latest_messages.at(0)?.ts} to ${
              item.latest_messages.at(-1)?.ts
            }`,
          },
        });
      }
    }

    // Save in the store
    ConversationsUpdaterService.saveRecentConversations(dataWithHeight);

    try {
      await Promise.all([...conversationsPromises, ...messagesPromises]);
    } catch (error) {
      sentryLogs.error({
        message: 'Failed to persist recents to DB',
        module: 'CHATS',
        component: 'OrchestratorUpdaterService.saveRecentConversations',
        error,
      });
      throw error;
    }
  }
  // ----------------

  //-------- Messages -------------

  private messagesToProcess: Message[] = [];
  private flushingPromise: Promise<void> | null = null;
  private retryCount = new Map<string, number>();
  private static readonly MAX_DB_RETRIES = 2;

  async processMessage(messageInput: Message): Promise<void> {
    BlockingLatch.block(messageInput.channel);

    // Register message as pending before any processing
    // This prevents the range from being extended until the message is confirmed saved
    PendingMessagesTracker.registerPending(
      messageInput.channel,
      messageInput.ts,
    );

    if (messageInput.attachments) {
      fileHandler.processAttachmentsBatch([
        {
          attachments: messageInput.attachments,
          messageId: messageInput.client_msg_id,
          messageMetadata: {
            hidden: messageInput.hidden,
            channel: messageInput.channel,
          },
        },
      ]);
      // Warm ExpoImage's cache for the message's thumbnails
      prefetchMessageImages(messageInput.attachments);
    }

    PendingAssets.addOrRemoveFromMessage(messageInput);

    const formattedText = messageInput.formatted_text
      ? messageInput.formatted_text
      : messageInput.highlighted_spans?.length
      ? await getTextWithHighlights({
          message: messageInput.text,
          highlightedSpans: messageInput.highlighted_spans,
        })
      : messageInput.text;
    const message: Message = {
      ...messageInput,
      formatted_text: formattedText,
      height: await calculateTextHeight({
        text: formattedText,
        blocks: messageInput.blocks,
      }),
    };

    MessagesUpdaterService.saveMessages([message], {
      updateStore: true,
      updateDb: false,
    });

    if (message.reactions?.length) {
      ReactionsUpdaterService.addMessageReactionsToStore(
        message.client_msg_id,
        mapReactionToStore({
          reactions: message.reactions,
        }),
      );
    }

    if (PinnedMessagesUpdaterService.isPinned(message.channel, message.ts)) {
      PinnedMessagesUpdaterService.updateMessage(message.channel, message);
    }

    this.addToSave(message);
  }

  private addToSave(message: Message) {
    this.messagesToProcess.push(message);
    this.scheduleFlush();
  }

  private scheduleFlush() {
    if (!this.flushingPromise) {
      this.flushingPromise = this.flushMessagesToProcess();
    }
  }

  /**
   * Waits for the batch window, then atomically swaps the queue and persists
   * all accumulated messages to the DB. Messages arriving during the flush
   * go into a fresh queue and will be picked up by the next scheduled flush.
   */
  private async flushMessagesToProcess(): Promise<void> {
    await wait(OrchestratorUpdaterServiceClass.DB_FLUSH_DELAY_MS);

    // Atomic swap: messages arriving after this point go to the new empty queue
    const batch = this.messagesToProcess;
    this.messagesToProcess = [];

    try {
      const latestMessages = await reduceToLatestMessages(batch);

      // Update store with latest messages
      MessagesUpdaterService.setMessagesAsLatest(latestMessages);

      // Save messages to the DB and update the latest message
      await Promise.all([
        MessagesUpdaterService.saveMessages(batch, {
          //Update DB only
          updateDb: true,
          updateStore: false,
        }),
        ConversationsUpdaterService.updateConversationsLatest(latestMessages),
      ]);

      // Confirm all messages as saved and update ranges
      PendingMessagesTracker.confirmMessagesAndUpdateRanges(batch);
      ReactionsUpdaterService.processWithMessages(batch);
      // Clear retry counters for successfully saved messages
      batch.forEach(msg => this.retryCount.delete(msg.client_msg_id));
    } catch (error) {
      const toRetry: Message[] = [];
      const permanentlyFailed: Message[] = [];

      for (const msg of batch) {
        const attempts = (this.retryCount.get(msg.client_msg_id) ?? 0) + 1;
        if (attempts <= OrchestratorUpdaterServiceClass.MAX_DB_RETRIES) {
          this.retryCount.set(msg.client_msg_id, attempts);
          toRetry.push(msg);
        } else {
          this.retryCount.delete(msg.client_msg_id);
          permanentlyFailed.push(msg);
        }
      }

      if (toRetry.length > 0) {
        // Re-insert at the front so they are retried in the next flush
        this.messagesToProcess = [...toRetry, ...this.messagesToProcess];
      }

      if (permanentlyFailed.length > 0) {
        PendingMessagesTracker.markMessagesFailed(permanentlyFailed);
        sentryLogs.error({
          message: 'Failed to save messages to DB after retries',
          module: 'CHATS',
          component: 'OrchestratorUpdaterService.flushMessagesToProcess',
          error,
          data: {messageCount: permanentlyFailed.length},
        });
      }
    } finally {
      // Unblock all channels related to the saved messages
      batch.forEach(msg => {
        BlockingLatch.unblock(msg.channel);
      });

      this.flushingPromise = null;

      // If new messages arrived during the flush, schedule another batch
      if (this.messagesToProcess.length > 0) {
        this.scheduleFlush();
      }
    }
  }

  /**
   * Save new messages to the DB and update the stores. Before call this, you need to calculate the height of the messages,
   * calling `getMessagesWithHeight`.
   *
   * @param options.skipStoreUpdate - If true, skip updating the store (useful for pagination where the caller handles the window update)
   */
  async saveNewMessages(
    messages: Message[],
    options?: {skipStoreUpdate?: boolean},
  ): Promise<void> {
    if (!messages.length) {
      return;
    }

    // unique channel IDs to block/unblock
    const uniqueIds = [...new Set(messages.map(m => m.channel))];

    uniqueIds.forEach(id => {
      BlockingLatch.block(id);
    });

    try {
      await this.saveMessages(messages, {
        skipStoreUpdate: options?.skipStoreUpdate,
      });

      ReactionsUpdaterService.processWithMessages(messages);
      AssetsUpdaterService.processAttachments(messages);
    } finally {
      uniqueIds.forEach(id => {
        BlockingLatch.unblock(id);
      });
    }
  }

  private async saveMessages(
    messages: Message[],
    options?: {withLatestUpdate?: boolean; skipStoreUpdate?: boolean},
  ) {
    const {withLatestUpdate = true, skipStoreUpdate} = options || {};
    await Promise.all([
      MessagesUpdaterService.saveMessages(messages, {
        updateStore: !skipStoreUpdate,
      }),
      withLatestUpdate
        ? ConversationsUpdaterService.updateConversationsLatest(messages, {
            withMapper: true,
          })
        : null,
    ]);
  }

  // ------ Users and Memberships ------

  async saveUsersAndMemberships(memberships: Membership[], channelId: string) {
    await UsersUpdaterService.saveUsers(
      memberships.map(membership => membership.user),
    );

    await MembershipUpdaterService.saveGroupMemberships({
      channelId,
      memberships: mapMembershipsToDB(channelId, memberships),
    });
  }

  async processUsersAndMemberships(
    memberships: Membership[],
    channelId: string,
  ) {
    const validUserIds = memberships.map(m => m.user.id);

    await db.instance.transaction(async tx => {
      await UsersUpdaterService.saveUsers(
        memberships.map(membership => membership.user),
        {tx},
      );
      await MembershipUpdaterService.saveGroupMemberships({
        channelId,
        memberships: mapMembershipsToDB(channelId, memberships),
        tx,
      });
      if (validUserIds.length > 0) {
        await deleteOrphanedMemberships(channelId, validUserIds, tx);
      }
    });
  }
}

export const OrchestratorUpdaterService = new OrchestratorUpdaterServiceClass();
