Fix private calls

2026-05-17 15:14:52 +02:00
parent 0f6cb3ee77
commit e769a6ee4a
71 changed files with 5821 additions and 349 deletions
@@ -12,10 +12,13 @@ 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 {
  advanceDirectMessageStatus,
  createDirectConversation,
+  createGroupConversation,
  getDirectConversationId,
+  isGroupDirectConversation,
  updateMessageStatusInConversation,
  upsertDirectMessage
 } from '../../domain/logic/direct-message.logic';
@@ -24,8 +27,12 @@ import {
  DirectMessageConversation,
  DirectMessageEventPayload,
  DirectMessageMutationEventPayload,
+  DirectMessageParticipant,
+  DirectMessageSyncEventPayload,
+  DirectMessageSyncRequestEventPayload,
  DirectMessageStatus,
  DirectMessageStatusEventPayload,
+  DirectMessageTypingEventPayload,
  toDirectMessageParticipant
 } from '../../domain/models/direct-message.model';
 import type {
@@ -35,16 +42,32 @@ import type {
 } 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 store = inject(Store);
  private readonly router = inject(Router);
  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(
@@ -62,6 +85,7 @@ export class DirectMessageService {
    (total, conversation) => total + conversation.unreadCount,
    0
  ));
+  readonly typingEntries = this.typingEntriesSignal.asReadonly();

  constructor() {
    effect(() => {
@@ -76,11 +100,15 @@ export class DirectMessageService {

    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> {
@@ -106,12 +134,47 @@ export class DirectMessageService {
    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 {
@@ -152,10 +215,11 @@ export class DirectMessageService {
    const ownerId = this.getCurrentUserIdOrThrow();
    const conversation = await this.requireConversation(ownerId, conversationId);
    const senderId = currentUser.oderId || currentUser.id;
-    const recipientId = conversation.participants.find((participantId) => participantId !== senderId);
+    const recipientIds = this.recipientIdsFor(conversation, senderId);
+    const recipientId = recipientIds[0];

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

    const message: DirectMessage = {
@@ -163,6 +227,7 @@ export class DirectMessageService {
      conversationId,
      senderId,
      recipientId,
+      recipientIds,
      content: normalizedContent,
      timestamp: Date.now(),
      status: 'QUEUED',
@@ -172,7 +237,7 @@ export class DirectMessageService {
    };

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

@@ -249,9 +314,8 @@ export class DirectMessageService {
  requestPeerAvatarSync(conversationId: string): void {
    const currentUserId = this.getCurrentUserId();
    const conversation = this.conversationsSignal().find((entry) => entry.id === conversationId);
-    const peerId = conversation?.participants.find((participantId) => participantId !== currentUserId);

-    if (peerId) {
+    for (const peerId of this.recipientIdsFor(conversation, currentUserId)) {
      this.delivery.requestUserAvatar(peerId);
    }
  }
@@ -321,13 +385,51 @@ export class DirectMessageService {

    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) {
-        await this.attemptDelivery(ownerId, message);
+      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);
@@ -341,6 +443,21 @@ export class DirectMessageService {

    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);
    }
  }

@@ -351,8 +468,16 @@ export class DirectMessageService {
    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)
-      ?? createDirectConversation(currentParticipant, sender, payload.message.timestamp);
+      ?? (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,
@@ -360,7 +485,7 @@ export class DirectMessageService {
    };
    const shouldIncrementUnread = !this.isConversationVisible(conversationId);

-    await this.persistConversation(ownerId, upsertDirectMessage(existingConversation, incomingMessage, shouldIncrementUnread));
+    await this.persistConversation(ownerId, upsertDirectMessage(conversationWithParticipants, incomingMessage, shouldIncrementUnread));
    this.sendStatusUpdate(incomingMessage.senderId, {
      conversationId,
      messageId: incomingMessage.id,
@@ -384,14 +509,20 @@ export class DirectMessageService {
  private isConversationVisible(conversationId: string): boolean {
    const currentUrl = this.router.url.split(/[?#]/, 1)[0];

-    if (!currentUrl.startsWith('/dm/')) {
+    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('/dm/'.length)) === conversationId;
+      return decodeURIComponent(currentUrl.slice(prefix.length)) === conversationId;
    } catch {
-      return currentUrl.slice('/dm/'.length) === conversationId;
+      return currentUrl.slice(prefix.length) === conversationId;
    }
  }

@@ -402,6 +533,98 @@ export class DirectMessageService {
    await this.persistConversation(ownerId, this.applyMutation(conversation, payload));
  }

+  private handleIncomingTyping(payload: DirectMessageTypingEventPayload): void {
+    const currentUserId = this.getCurrentUserId();
+
+    if (!currentUserId || payload.sender.userId === currentUserId) {
+      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 = this.conversationsSignal().find((entry) => entry.id === payload.conversationId)
+      ?? await this.repository.getConversation(ownerId, payload.conversationId);
+
+    if (!conversation || payload.sender.userId === ownerId) {
+      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);
+
+    if (payload.sender.userId === ownerId) {
+      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
@@ -409,11 +632,11 @@ export class DirectMessageService {
    const ownerId = this.getCurrentUserIdOrThrow();
    const conversation = await this.requireConversation(ownerId, conversationId);
    const updatedConversation = this.applyMutation(conversation, payload);
-    const recipientId = conversation.participants.find((participantId) => participantId !== ownerId);
+    const recipientIds = this.recipientIdsFor(conversation, ownerId);

    await this.persistConversation(ownerId, updatedConversation);

-    if (recipientId) {
+    for (const recipientId of recipientIds) {
      this.delivery.sendViaWebRTC(recipientId, {
        type: 'direct-message-mutation',
        directMessageMutation: payload
@@ -474,23 +697,38 @@ export class DirectMessageService {
    return { ...conversation, messages };
  }

-  private async attemptDelivery(ownerId: string, message: DirectMessage): Promise<void> {
+  private async attemptDelivery(ownerId: string, message: DirectMessage, conversation: DirectMessageConversation): Promise<void> {
    const currentUser = this.requireCurrentUser();
-    const sent = this.delivery.sendViaWebRTC(message.recipientId, {
-      type: 'direct-message',
-      directMessage: {
-        message,
-        sender: toDirectMessageParticipant(currentUser)
-      }
-    });
+    const recipientIds = this.recipientIdsFor(conversation, ownerId);

-    if (!sent) {
-      await this.offlineQueue.enqueue(ownerId, message.id);
-      return;
+    let sentCount = 0;
+
+    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;
+      }
    }

-    await this.offlineQueue.markDelivered(ownerId, message.id);
-    await this.updateStatus(message.id, 'SENT');
+    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 {
@@ -500,6 +738,52 @@ export class DirectMessageService {
    });
  }

+  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;
@@ -512,10 +796,14 @@ export class DirectMessageService {
    }

    this.loadedOwnerId = ownerId;
-    this.conversationsSignal.set(await this.repository.loadConversations(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);
@@ -525,6 +813,55 @@ export class DirectMessageService {
    });
  }

+  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 recipientIdsFor(conversation: DirectMessageConversation | null | undefined, currentUserId: string | null | undefined): string[] {
+    if (!conversation || !currentUserId) {
+      return [];
+    }
+
+    return conversation.participants.filter((participantId) => participantId !== currentUserId);
+  }
+
+  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> {
    await this.loadForOwner(ownerId);