Toju/toju-app/src/app/domains/direct-message/application/services/direct-message.service.ts

/* eslint-disable @typescript-eslint/member-ordering */
import {
  Injectable,
  computed,
  effect,
  inject,
  signal
} from '@angular/core';
import { Router } from '@angular/router';
import { Store } from '@ngrx/store';
import { v4 as uuidv4 } from 'uuid';
import { DirectMessageRepository } from '../../infrastructure/direct-message.repository';
import { OfflineMessageQueueService } from './offline-message-queue.service';
import { PeerDeliveryService } from './peer-delivery.service';
import { AttachmentFacade } from '../../../attachment';
import { CustomEmojiService } from '../../../custom-emoji';
import { NotificationsFacade } from '../../../notifications';
import { SignalServerCredentialStoreService } from '../../../authentication/application/services/signal-server-credential-store.service';
import {
  advanceDirectMessageStatus,
  createDirectConversation,
  createGroupConversation,
  createDirectCallStartedMessage,
  directMessageConversationIncludesUser,
  directMessageEventIncludesUser,
  directMessageSyncIncludesUser,
  getDirectConversationId,
  isGroupDirectConversation,
  updateMessageStatusInConversation,
  upsertDirectMessage
} from '../../domain/logic/direct-message.logic';
import { collectDirectMessageSelfUserIds, isSelfDirectMessageSender } from '../../domain/logic/direct-message-identity.rules';
import {
  DirectMessage,
  DirectMessageConversation,
  DirectMessageEventPayload,
  DirectMessageMutationEventPayload,
  DirectMessageParticipant,
  DirectMessageSyncEventPayload,
  DirectMessageSyncRequestEventPayload,
  DirectMessageStatus,
  DirectMessageStatusEventPayload,
  DirectMessageTypingEventPayload,
  toDirectMessageParticipant
} from '../../domain/models/direct-message.model';
import type {
  ChatEvent,
  Reaction,
  User
} from '../../../../shared-kernel';
import { selectCurrentUser } from '../../../../store/users/users.selectors';

const DIRECT_MESSAGE_SYNC_LIMIT = 1000;
const DIRECT_MESSAGE_SYNC_REQUEST_COOLDOWN_MS = 5000;
const DIRECT_MESSAGE_TYPING_TTL_MS = 3000;
const DIRECT_MESSAGE_TYPING_PURGE_MS = 1000;
const DIRECT_MESSAGE_ATTACHMENT_STORAGE_PREFIX = 'direct-message:';

interface DirectMessageTypingEntry {
  conversationId: string;
  userId: string;
  displayName: string;
  expiresAt: number;
}

@Injectable({ providedIn: 'root' })
export class DirectMessageService {
  private readonly repository = inject(DirectMessageRepository);
  private readonly offlineQueue = inject(OfflineMessageQueueService);
  private readonly delivery = inject(PeerDeliveryService);
  private readonly attachments = inject(AttachmentFacade);
  private readonly customEmoji = inject(CustomEmojiService);
  private readonly credentialStore = inject(SignalServerCredentialStoreService);
  private readonly store = inject(Store);
  private readonly router = inject(Router);
  private readonly notifications = inject(NotificationsFacade);
  private readonly currentUser = this.store.selectSignal(selectCurrentUser);
  private readonly conversationsSignal = signal<DirectMessageConversation[]>([]);
  private readonly selectedConversationIdSignal = signal<string | null>(null);
  private readonly typingEntriesSignal = signal<DirectMessageTypingEntry[]>([]);
  private readonly lastSyncRequestAt = new Map<string, number>();
  private loadedOwnerId: string | null = null;

  readonly conversations = computed(() => [...this.conversationsSignal()].sort(
    (firstConversation, secondConversation) => secondConversation.lastMessageAt - firstConversation.lastMessageAt
  ));
  readonly selectedConversationId = this.selectedConversationIdSignal.asReadonly();
  readonly selectedConversation = computed(() => {
    const selectedId = this.selectedConversationIdSignal();

    return selectedId
      ? this.conversationsSignal().find((conversation) => conversation.id === selectedId) ?? null
      : null;
  });
  readonly totalUnreadCount = computed(() => this.conversationsSignal().reduce(
    (total, conversation) => total + conversation.unreadCount,
    0
  ));
  readonly typingEntries = this.typingEntriesSignal.asReadonly();

  constructor() {
    effect(() => {
      const ownerId = this.getCurrentUserId();

      void this.loadForOwner(ownerId);
    });

    this.delivery.directMessageEvents$.subscribe((event) => {
      void this.handlePeerEvent(event);
    });

    this.delivery.peerConnected$.subscribe(() => {
      void this.retryPending();
      void this.requestOpenConversationSync();
    });

    this.delivery.networkRestored$.subscribe(() => {
      void this.retryPending();
      void this.requestOpenConversationSync();
    });

    window.setInterval(() => this.purgeExpiredTypingEntries(), DIRECT_MESSAGE_TYPING_PURGE_MS);
  }

  async createConversation(user: User): Promise<DirectMessageConversation> {
    const currentUser = this.requireCurrentUser();
    const ownerId = this.getCurrentUserIdOrThrow();

    await this.loadForOwner(ownerId);

    const currentParticipant = toDirectMessageParticipant(currentUser);
    const peerParticipant = toDirectMessageParticipant(user);
    const conversationId = getDirectConversationId(currentParticipant.userId, peerParticipant.userId);
    const existingConversation = this.conversationsSignal().find((conversation) => conversation.id === conversationId);

    if (existingConversation) {
      this.selectedConversationIdSignal.set(existingConversation.id);
      return existingConversation;
    }

    const conversation = createDirectConversation(currentParticipant, peerParticipant, Date.now());

    await this.persistConversation(ownerId, conversation);
    this.selectedConversationIdSignal.set(conversation.id);
    return conversation;
  }

  async createGroupConversation(
    participants: DirectMessageParticipant[],
    title?: string,
    conversationId = `dm-group-${uuidv4()}`
  ): Promise<DirectMessageConversation> {
    const currentUser = this.requireCurrentUser();
    const ownerId = this.getCurrentUserIdOrThrow();
    const currentParticipant = toDirectMessageParticipant(currentUser);
    const allParticipants = this.uniqueParticipants([currentParticipant, ...participants]);

    await this.loadForOwner(ownerId);

    const existingConversation = this.conversationsSignal().find((conversation) => conversation.id === conversationId)
      ?? await this.repository.getConversation(ownerId, conversationId);

    if (existingConversation) {
      const mergedConversation = this.mergeConversationParticipants({
        ...existingConversation,
        kind: 'group',
        title: existingConversation.title || title
      }, allParticipants);

      await this.persistConversation(ownerId, mergedConversation);
      this.selectedConversationIdSignal.set(mergedConversation.id);
      return mergedConversation;
    }

    const conversation = createGroupConversation(conversationId, allParticipants, Date.now(), title);

    await this.persistConversation(ownerId, conversation);
    this.selectedConversationIdSignal.set(conversation.id);
    return conversation;
  }

  async openConversation(conversationId: string): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();

    await this.loadForOwner(ownerId);
    this.selectedConversationIdSignal.set(conversationId);
    await this.markRead(conversationId);
    this.requestConversationSync(conversationId);
  }

  closeConversationView(conversationId?: string | null): void {
    if (!conversationId || this.selectedConversationIdSignal() === conversationId) {
      this.selectedConversationIdSignal.set(null);
    }
  }

  async forgetConversation(conversationId: string): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const conversation = await this.repository.getConversation(ownerId, conversationId);

    if (!conversation) {
      return;
    }

    await this.repository.deleteConversation(ownerId, conversationId);

    for (const message of conversation.messages) {
      await this.offlineQueue.markDelivered(ownerId, message.id);
    }

    this.conversationsSignal.update((conversations) => conversations.filter((entry) => entry.id !== conversationId));

    if (this.selectedConversationIdSignal() === conversationId) {
      this.selectedConversationIdSignal.set(null);
    }
  }

  async sendMessage(
    conversationId: string,
    content: string,
    replyToId?: string,
    id?: string
  ): Promise<DirectMessage> {
    const normalizedContent = content.trim();

    if (!normalizedContent) {
      throw new Error('Cannot send an empty direct message.');
    }

    const currentUser = this.requireCurrentUser();
    const ownerId = this.getCurrentUserIdOrThrow();
    const conversation = await this.requireConversation(ownerId, conversationId);
    const senderId = currentUser.oderId || currentUser.id;
    const recipientIds = this.recipientIdsFor(conversation, senderId);
    const recipientId = recipientIds[0];

    if (!recipientId) {
      throw new Error('Direct message conversation has no recipients.');
    }

    const message: DirectMessage = {
      id: id ?? uuidv4(),
      conversationId,
      senderId,
      recipientId,
      recipientIds,
      content: normalizedContent,
      timestamp: Date.now(),
      status: 'QUEUED',
      reactions: [],
      isDeleted: false,
      replyToId
    };

    await this.persistConversation(ownerId, upsertDirectMessage(conversation, message, false));
    await this.attemptDelivery(ownerId, message, conversation);
    return message;
  }

  async recordCallStarted(
    conversationId: string,
    caller: DirectMessageParticipant,
    participants: DirectMessageParticipant[],
    timestamp = Date.now()
  ): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const currentUser = this.requireCurrentUser();
    const currentParticipant = toDirectMessageParticipant(currentUser);
    const allParticipants = this.uniqueParticipants([
      currentParticipant,
      caller,
      ...participants
    ]);

    await this.loadForOwner(ownerId);

    const existingConversation = this.conversationsSignal().find((conversation) => conversation.id === conversationId)
      ?? await this.repository.getConversation(ownerId, conversationId)
      ?? this.createConversationForSystemEvent(conversationId, currentParticipant, caller, allParticipants, timestamp);
    const conversation = this.mergeConversationParticipants(existingConversation, allParticipants);
    const message = createDirectCallStartedMessage(
      conversation.id,
      caller,
      conversation.participants.filter((participantId) => participantId !== caller.userId),
      timestamp
    );

    await this.persistConversation(ownerId, upsertDirectMessage(conversation, message, false));
  }

  async editMessage(conversationId: string, messageId: string, content: string): Promise<void> {
    const normalizedContent = content.trim();

    if (!normalizedContent) {
      return;
    }

    await this.applyAndSendMutation(conversationId, {
      conversationId,
      messageId,
      type: 'edit',
      content: normalizedContent,
      editedAt: Date.now(),
      updatedAt: Date.now()
    });
  }

  async deleteMessage(conversationId: string, messageId: string): Promise<void> {
    await this.applyAndSendMutation(conversationId, {
      conversationId,
      messageId,
      type: 'delete',
      updatedAt: Date.now()
    });
  }

  async addReaction(conversationId: string, messageId: string, emoji: string): Promise<void> {
    const userId = this.getCurrentUserIdOrThrow();
    const reaction: Reaction = {
      id: uuidv4(),
      messageId,
      oderId: userId,
      userId,
      emoji,
      timestamp: Date.now()
    };

    await this.applyAndSendMutation(conversationId, {
      conversationId,
      messageId,
      type: 'reaction-add',
      reaction,
      updatedAt: reaction.timestamp
    });
  }

  async toggleReaction(conversationId: string, messageId: string, emoji: string): Promise<void> {
    const userId = this.getCurrentUserIdOrThrow();
    const conversation = await this.requireConversation(userId, conversationId);
    const message = conversation.messages.find((entry) => entry.id === messageId);
    const existingReaction = message?.reactions?.find((reaction) =>
      reaction.emoji === emoji && (reaction.userId === userId || reaction.oderId === userId)
    );

    if (existingReaction) {
      await this.applyAndSendMutation(conversationId, {
        conversationId,
        messageId,
        type: 'reaction-remove',
        oderId: userId,
        emoji,
        updatedAt: Date.now()
      });

      return;
    }

    await this.addReaction(conversationId, messageId, emoji);
  }

  requestPeerAvatarSync(conversationId: string): void {
    const currentUserId = this.getCurrentUserId();
    const conversation = this.conversationsSignal().find((entry) => entry.id === conversationId);

    for (const peerId of this.recipientIdsFor(conversation, currentUserId)) {
      this.delivery.requestUserAvatar(peerId);
    }
  }

  currentUserId(): string | null {
    return this.getCurrentUserId();
  }

  async updateStatus(messageId: string, status: DirectMessageStatus): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const conversation = this.conversationsSignal().find((entry) => entry.messages.some((message) => message.id === messageId));

    if (!conversation) {
      return;
    }

    await this.persistConversation(ownerId, updateMessageStatusInConversation(conversation, messageId, status));
  }

  async receiveMessage(message: DirectMessage, sender: User): Promise<void> {
    await this.handleIncomingMessage({
      message,
      sender: toDirectMessageParticipant(sender)
    });
  }

  async markRead(conversationId: string): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const currentUserId = this.getCurrentUserIdOrThrow();
    const conversation = await this.requireConversation(ownerId, conversationId);
    const updatedConversation = { ...conversation, unreadCount: 0 };

    await this.persistConversation(ownerId, updatedConversation);
    await this.repository.markRead(ownerId, conversationId);

    for (const message of updatedConversation.messages) {
      if (message.recipientId !== currentUserId || message.status === 'ACKNOWLEDGED') {
        continue;
      }

      const nextStatus = advanceDirectMessageStatus(message.status, 'ACKNOWLEDGED');

      if (nextStatus !== message.status) {
        await this.persistConversation(ownerId, updateMessageStatusInConversation(updatedConversation, message.id, nextStatus));
      }

      this.sendStatusUpdate(message.senderId, {
        conversationId,
        messageId: message.id,
        status: 'ACKNOWLEDGED',
        updatedAt: Date.now()
      });
    }
  }

  async retryPending(): Promise<void> {
    const ownerId = this.getCurrentUserId();

    if (!ownerId) {
      return;
    }

    await this.loadForOwner(ownerId);

    const pendingMessageIds = await this.offlineQueue.retryPending(ownerId);
    const messages = this.conversationsSignal().flatMap((conversation) => conversation.messages);

    for (const messageId of pendingMessageIds) {
      const message = messages.find((entry) => entry.id === messageId);
      const conversation = message
        ? this.conversationsSignal().find((entry) => entry.id === message.conversationId)
        : null;

      if (message && conversation) {
        await this.attemptDelivery(ownerId, message, conversation);
      }
    }
  }

  typingUsers(conversationId: string | null | undefined): string[] {
    if (!conversationId) {
      return [];
    }

    const now = Date.now();

    return this.typingEntriesSignal()
      .filter((entry) => entry.conversationId === conversationId && entry.expiresAt > now)
      .map((entry) => entry.displayName);
  }

  sendTyping(conversationId: string, isTyping = true): void {
    const conversation = this.conversationsSignal().find((entry) => entry.id === conversationId);
    const currentUser = this.currentUser();
    const currentUserId = this.getCurrentUserId();
    const recipientIds = this.recipientIdsFor(conversation, currentUserId);

    if (!conversation || !currentUser || recipientIds.length === 0) {
      return;
    }

    for (const recipientId of recipientIds) {
      this.delivery.sendViaWebRTC(recipientId, {
        type: 'direct-message-typing',
        directMessageTyping: {
          conversationId,
          sender: toDirectMessageParticipant(currentUser),
          isTyping,
          updatedAt: Date.now()
        }
      });
    }
  }

  private async handlePeerEvent(event: ChatEvent): Promise<void> {
    if (event.type === 'direct-message' && event.directMessage) {
      await this.handleIncomingMessage(event.directMessage);
      return;
    }

    if (event.type === 'direct-message-status' && event.directMessageStatus) {
      await this.handleIncomingStatus(event.directMessageStatus);
      return;
    }

    if (event.type === 'direct-message-mutation' && event.directMessageMutation) {
      await this.handleIncomingMutation(event.directMessageMutation);
      return;
    }

    if (event.type === 'direct-message-typing' && event.directMessageTyping) {
      this.handleIncomingTyping(event.directMessageTyping);
      return;
    }

    if (event.type === 'direct-message-sync-request' && event.directMessageSyncRequest) {
      await this.handleIncomingSyncRequest(event.directMessageSyncRequest);
      return;
    }

    if (event.type === 'direct-message-sync' && event.directMessageSync) {
      await this.handleIncomingSync(event.directMessageSync);
    }
  }

  private async handleIncomingMessage(payload: DirectMessageEventPayload): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const currentUser = this.requireCurrentUser();
    const selfUserIds = this.getSelfUserIds();

    if (!directMessageEventIncludesUser(payload, selfUserIds) || isSelfDirectMessageSender(payload, selfUserIds)) {
      return;
    }

    const currentParticipant = toDirectMessageParticipant(currentUser);
    const sender = payload.sender;
    const conversationId = payload.message.conversationId
      || getDirectConversationId(currentParticipant.userId, sender.userId);
    const participants = this.uniqueParticipants([
      currentParticipant,
      sender,
      ...(payload.participants ?? [])
    ]);
    const existingConversation = this.conversationsSignal().find((conversation) => conversation.id === conversationId)
      ?? (payload.conversationKind === 'group' || participants.length > 2
        ? createGroupConversation(conversationId, participants, payload.message.timestamp, payload.conversationTitle)
        : createDirectConversation(currentParticipant, sender, payload.message.timestamp));
    const conversationWithParticipants = this.mergeConversationParticipants(existingConversation, participants);
    const incomingMessage: DirectMessage = {
      ...payload.message,
      conversationId,
      status: advanceDirectMessageStatus(payload.message.status, 'DELIVERED')
    };
    const shouldIncrementUnread = !this.isConversationVisible(conversationId);

    await this.persistConversation(ownerId, upsertDirectMessage(conversationWithParticipants, incomingMessage, shouldIncrementUnread));
    this.sendStatusUpdate(incomingMessage.senderId, {
      conversationId,
      messageId: incomingMessage.id,
      status: 'DELIVERED',
      updatedAt: Date.now()
    });

    if (incomingMessage.kind !== 'system' && !incomingMessage.isDeleted) {
      void this.notifications.handleIncomingDirectMessage({
        id: incomingMessage.id,
        senderName: sender.displayName || sender.username || sender.userId,
        content: incomingMessage.content,
        conversationVisible: !shouldIncrementUnread
      });
    }

    if (!shouldIncrementUnread) {
      await this.markRead(conversationId);
    }
  }

  private async handleIncomingStatus(payload: DirectMessageStatusEventPayload): Promise<void> {
    await this.updateStatus(payload.messageId, payload.status);

    if (payload.status === 'DELIVERED' || payload.status === 'ACKNOWLEDGED') {
      await this.offlineQueue.markDelivered(this.getCurrentUserIdOrThrow(), payload.messageId);
    }
  }

  private isConversationVisible(conversationId: string): boolean {
    const currentUrl = this.router.url.split(/[?#]/, 1)[0];

    if (!currentUrl.startsWith('/dm/') && !currentUrl.startsWith('/pm/')) {
      if (currentUrl.startsWith('/call/')) {
        return this.selectedConversationIdSignal() === conversationId;
      }

      return false;
    }

    const prefix = currentUrl.startsWith('/pm/') ? '/pm/' : '/dm/';

    try {
      return decodeURIComponent(currentUrl.slice(prefix.length)) === conversationId;
    } catch {
      return currentUrl.slice(prefix.length) === conversationId;
    }
  }

  private async handleIncomingMutation(payload: DirectMessageMutationEventPayload): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const conversation = await this.findConversation(ownerId, payload.conversationId);
    const selfUserIds = this.getSelfUserIds();

    if (!conversation || !directMessageConversationIncludesUser(conversation, selfUserIds)) {
      return;
    }

    await this.persistConversation(ownerId, this.applyMutation(conversation, payload));
  }

  private handleIncomingTyping(payload: DirectMessageTypingEventPayload): void {
    const selfUserIds = this.getSelfUserIds();

    if (selfUserIds.size === 0 || selfUserIds.has(payload.sender.userId)) {
      return;
    }

    const conversation = this.conversationsSignal().find((entry) => entry.id === payload.conversationId);

    if (!conversation
      || !directMessageConversationIncludesUser(conversation, selfUserIds)
      || !directMessageConversationIncludesUser(conversation, payload.sender.userId)) {
      return;
    }

    if (!payload.isTyping) {
      this.typingEntriesSignal.update((entries) => entries.filter((entry) =>
        !(entry.conversationId === payload.conversationId && entry.userId === payload.sender.userId)
      ));

      return;
    }

    const nextEntry: DirectMessageTypingEntry = {
      conversationId: payload.conversationId,
      userId: payload.sender.userId,
      displayName: payload.sender.displayName,
      expiresAt: Date.now() + DIRECT_MESSAGE_TYPING_TTL_MS
    };

    this.typingEntriesSignal.update((entries) => [
      ...entries.filter((entry) =>
        !(entry.conversationId === nextEntry.conversationId && entry.userId === nextEntry.userId)
      ),
      nextEntry
    ]);
  }

  private async handleIncomingSyncRequest(payload: DirectMessageSyncRequestEventPayload): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const currentUser = this.requireCurrentUser();
    const conversation = await this.findConversation(ownerId, payload.conversationId);
    const selfUserIds = this.getSelfUserIds();

    if (!conversation
      || selfUserIds.has(payload.sender.userId)
      || !directMessageConversationIncludesUser(conversation, selfUserIds)
      || !directMessageConversationIncludesUser(conversation, payload.sender.userId)) {
      return;
    }

    this.delivery.sendViaWebRTC(payload.sender.userId, {
      type: 'direct-message-sync',
      directMessageSync: {
        conversationId: conversation.id,
        sender: toDirectMessageParticipant(currentUser),
        participants: Object.values(conversation.participantProfiles),
        conversationKind: this.conversationKind(conversation),
        conversationTitle: conversation.title,
        messages: conversation.messages.slice(-DIRECT_MESSAGE_SYNC_LIMIT),
        syncedAt: Date.now()
      }
    });
  }

  private async handleIncomingSync(payload: DirectMessageSyncEventPayload): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const currentUser = this.requireCurrentUser();
    const currentParticipant = toDirectMessageParticipant(currentUser);
    const selfUserIds = this.getSelfUserIds();

    if (selfUserIds.has(payload.sender.userId)) {
      return;
    }

    if (!directMessageSyncIncludesUser(payload, selfUserIds) || !directMessageSyncIncludesUser(payload, payload.sender.userId)) {
      return;
    }

    const existingConversation = this.conversationsSignal().find((conversation) => conversation.id === payload.conversationId)
      ?? await this.repository.getConversation(ownerId, payload.conversationId)
      ?? (payload.conversationKind === 'group' || payload.participants.length > 2
        ? createGroupConversation(payload.conversationId, [currentParticipant, ...payload.participants], payload.syncedAt, payload.conversationTitle)
        : createDirectConversation(currentParticipant, payload.sender, payload.syncedAt));
    const participantProfiles = {
      ...existingConversation.participantProfiles,
      ...Object.fromEntries(payload.participants.map((participant) => [participant.userId, participant])),
      [currentParticipant.userId]: currentParticipant
    };
    const syncBaseConversation: DirectMessageConversation = {
      ...existingConversation,
      kind: payload.conversationKind ?? existingConversation.kind,
      title: payload.conversationTitle ?? existingConversation.title,
      participants: Object.keys(participantProfiles).sort(),
      participantProfiles
    };
    const mergedConversation = payload.messages.reduce<DirectMessageConversation>(
      (conversation, message) => upsertDirectMessage(conversation, message, false),
      syncBaseConversation
    );

    await this.persistConversation(ownerId, mergedConversation);

    if (this.selectedConversationIdSignal() === payload.conversationId) {
      await this.markRead(payload.conversationId);
    }
  }

  private async applyAndSendMutation(
    conversationId: string,
    payload: DirectMessageMutationEventPayload
  ): Promise<void> {
    const ownerId = this.getCurrentUserIdOrThrow();
    const conversation = await this.requireConversation(ownerId, conversationId);
    const updatedConversation = this.applyMutation(conversation, payload);
    const recipientIds = this.recipientIdsFor(conversation, ownerId);

    await this.persistConversation(ownerId, updatedConversation);

    if (payload.type === 'edit' && payload.content) {
      this.customEmoji.pushEmojisInContent(payload.content);
    } else if (payload.type === 'reaction-add' && payload.reaction?.emoji) {
      this.customEmoji.pushEmojisInContent(payload.reaction.emoji);
    }

    for (const recipientId of recipientIds) {
      this.delivery.sendViaWebRTC(recipientId, {
        type: 'direct-message-mutation',
        directMessageMutation: payload
      });
    }
  }

  private applyMutation(
    conversation: DirectMessageConversation,
    payload: DirectMessageMutationEventPayload
  ): DirectMessageConversation {
    const messages = conversation.messages.map((message) => {
      if (message.id !== payload.messageId) {
        return message;
      }

      if (payload.type === 'edit' && payload.content) {
        return {
          ...message,
          content: payload.content,
          editedAt: payload.editedAt ?? payload.updatedAt,
          isDeleted: false
        };
      }

      if (payload.type === 'delete') {
        return {
          ...message,
          content: '',
          isDeleted: true,
          editedAt: payload.updatedAt
        };
      }

      if (payload.type === 'reaction-add' && payload.reaction) {
        const reactions = (message.reactions ?? []).filter((reaction) =>
          !(reaction.emoji === payload.reaction?.emoji && reaction.userId === payload.reaction.userId)
        );

        return {
          ...message,
          reactions: [...reactions, payload.reaction]
        };
      }

      if (payload.type === 'reaction-remove' && payload.oderId && payload.emoji) {
        return {
          ...message,
          reactions: (message.reactions ?? []).filter((reaction) =>
            !(reaction.emoji === payload.emoji && (reaction.userId === payload.oderId || reaction.oderId === payload.oderId))
          )
        };
      }

      return message;
    });

    return { ...conversation, messages };
  }

  private async attemptDelivery(ownerId: string, message: DirectMessage, conversation: DirectMessageConversation): Promise<void> {
    const currentUser = this.requireCurrentUser();
    const recipientIds = this.recipientIdsFor(conversation, ownerId);

    let sentCount = 0;

    this.customEmoji.pushEmojisInContent(message.content);

    for (const recipientId of recipientIds) {
      if (this.delivery.sendViaWebRTC(recipientId, {
        type: 'direct-message',
        directMessage: {
          message,
          sender: toDirectMessageParticipant(currentUser),
          participants: Object.values(conversation.participantProfiles),
          conversationKind: this.conversationKind(conversation),
          conversationTitle: conversation.title
        }
      })) {
        sentCount += 1;
      }
    }

    if (sentCount < recipientIds.length) {
      await this.offlineQueue.enqueue(ownerId, message.id);
    }

    if (sentCount > 0) {
      await this.updateStatus(message.id, 'SENT');
    }

    if (sentCount === recipientIds.length) {
      await this.offlineQueue.markDelivered(ownerId, message.id);
    }
  }

  private sendStatusUpdate(recipientId: string, payload: DirectMessageStatusEventPayload): void {
    this.delivery.handleAck(recipientId, {
      type: 'direct-message-status',
      directMessageStatus: payload
    });
  }

  private requestOpenConversationSync(): void {
    const conversationId = this.selectedConversationIdSignal();

    if (conversationId) {
      this.requestConversationSync(conversationId);
    }
  }

  private requestConversationSync(conversationId: string): void {
    const conversation = this.conversationsSignal().find((entry) => entry.id === conversationId);
    const currentUser = this.currentUser();
    const currentUserId = this.getCurrentUserId();
    const recipientIds = this.recipientIdsFor(conversation, currentUserId);

    if (!conversation || !currentUser || recipientIds.length === 0) {
      return;
    }

    const now = Date.now();

    for (const recipientId of recipientIds) {
      const syncKey = `${conversationId}:${recipientId}`;

      if (now - (this.lastSyncRequestAt.get(syncKey) ?? 0) < DIRECT_MESSAGE_SYNC_REQUEST_COOLDOWN_MS) {
        continue;
      }

      this.lastSyncRequestAt.set(syncKey, now);

      this.delivery.sendViaWebRTC(recipientId, {
        type: 'direct-message-sync-request',
        directMessageSyncRequest: {
          conversationId,
          sender: toDirectMessageParticipant(currentUser),
          requestedAt: Date.now()
        }
      });
    }
  }

  private purgeExpiredTypingEntries(): void {
    const now = Date.now();

    this.typingEntriesSignal.update((entries) => entries.filter((entry) => entry.expiresAt > now));
  }

  private async loadForOwner(ownerId: string | null): Promise<void> {
    if (!ownerId) {
      this.loadedOwnerId = null;
      this.conversationsSignal.set([]);
      return;
    }

    if (this.loadedOwnerId === ownerId) {
      return;
    }

    this.loadedOwnerId = ownerId;
    const conversations = await this.repository.loadConversations(ownerId);

    conversations.forEach((conversation) => this.rememberConversationAttachmentStorage(conversation));
    this.conversationsSignal.set(conversations);
  }

  private async persistConversation(ownerId: string, conversation: DirectMessageConversation): Promise<void> {
    this.rememberConversationAttachmentStorage(conversation);
    await this.repository.saveConversation(ownerId, conversation);
    this.conversationsSignal.update((conversations) => {
      const nextConversations = conversations.filter((entry) => entry.id !== conversation.id);

      nextConversations.push(conversation);
      return nextConversations;
    });
  }

  private rememberConversationAttachmentStorage(conversation: DirectMessageConversation): void {
    const storageContainer = `${DIRECT_MESSAGE_ATTACHMENT_STORAGE_PREFIX}${conversation.id}`;

    for (const message of conversation.messages) {
      this.attachments.rememberMessageRoom(message.id, storageContainer);
    }
  }

  private mergeConversationParticipants(
    conversation: DirectMessageConversation,
    participants: DirectMessageParticipant[]
  ): DirectMessageConversation {
    const participantProfiles = {
      ...conversation.participantProfiles,
      ...Object.fromEntries(participants.map((participant) => [participant.userId, participant]))
    };

    return {
      ...conversation,
      participants: Object.keys(participantProfiles).sort(),
      participantProfiles
    };
  }

  private createConversationForSystemEvent(
    conversationId: string,
    currentParticipant: DirectMessageParticipant,
    caller: DirectMessageParticipant,
    participants: DirectMessageParticipant[],
    timestamp: number
  ): DirectMessageConversation {
    if (participants.length > 2) {
      return createGroupConversation(conversationId, participants, timestamp);
    }

    const peer = participants.find((participant) => participant.userId !== currentParticipant.userId) ?? caller;

    return {
      ...createDirectConversation(currentParticipant, peer, timestamp),
      id: conversationId
    };
  }

  private recipientIdsFor(conversation: DirectMessageConversation | null | undefined, currentUserId: string | null | undefined): string[] {
    if (!conversation || !currentUserId) {
      return [];
    }

    const selfUserIds = this.getSelfUserIds();

    return conversation.participants.filter((participantId) => !selfUserIds.has(participantId));
  }

  private conversationKind(conversation: DirectMessageConversation): 'direct' | 'group' {
    return isGroupDirectConversation(conversation) ? 'group' : 'direct';
  }

  private uniqueParticipants(participants: DirectMessageParticipant[]): DirectMessageParticipant[] {
    const seen = new Set<string>();

    return participants.filter((participant) => {
      if (!participant.userId || seen.has(participant.userId)) {
        return false;
      }

      seen.add(participant.userId);
      return true;
    });
  }

  private async requireConversation(ownerId: string, conversationId: string): Promise<DirectMessageConversation> {
    const conversation = await this.findConversation(ownerId, conversationId);

    if (!conversation) {
      throw new Error('Direct message conversation not found.');
    }

    return conversation;
  }

  private async findConversation(ownerId: string, conversationId: string): Promise<DirectMessageConversation | null> {
    await this.loadForOwner(ownerId);

    return this.conversationsSignal().find((entry) => entry.id === conversationId)
      ?? await this.repository.getConversation(ownerId, conversationId);
  }

  private requireCurrentUser(): User {
    const currentUser = this.currentUser();

    if (!currentUser) {
      throw new Error('Cannot use direct messages without a current user.');
    }

    return currentUser;
  }

  private getCurrentUserId(): string | null {
    const user = this.currentUser();

    return user?.oderId || user?.id || null;
  }

  private getCurrentUserIdOrThrow(): string {
    const ownerId = this.getCurrentUserId();

    if (!ownerId) {
      throw new Error('Cannot use direct messages without a current user.');
    }

    return ownerId;
  }

  private getSelfUserIds(): ReadonlySet<string> {
    const currentUser = this.currentUser();

    if (!currentUser) {
      return new Set();
    }

    const actorUserIds = this.credentialStore.listValidCredentials().map((credential) => credential.userId);

    return collectDirectMessageSelfUserIds(currentUser, actorUserIds);
  }
}