myxelium
3266581d3c
- Camera and screen-share badges are decided by `shouldShowStreamIndicator`, so a live stream is visible from outside the voice channel and clicking the badge joins the streamer's channel before focusing the stream. - `ringParticipants` reports whether a `direct-call` ring reached anyone; a call that reached nobody shows `call.errors.ringUndelivered` instead of sitting in "calling" as if it were live. - Direct messages resolve every local identity alias, so a conversation opened from a foreign roster entry lands in the same thread.
1151 lines
40 KiB
TypeScript
1151 lines
40 KiB
TypeScript
|
|
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,
|
|
isGroupDirectConversation,
|
|
updateMessageStatusInConversation,
|
|
upsertDirectMessage
|
|
} from '../../domain/logic/direct-message.logic';
|
|
import {
|
|
buildDirectParticipantAliasIndex,
|
|
canonicalizeDirectConversationId,
|
|
canonicalizeDirectParticipantId,
|
|
collectDirectMessageSelfUserIds,
|
|
getCanonicalDirectConversationId,
|
|
isSelfDirectMessageSender,
|
|
mergeAliasDirectConversations,
|
|
type DirectParticipantAliasIndex
|
|
} 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 { selectAllUsers, 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 users = this.store.selectSignal(selectAllUsers);
|
|
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 aliasIndex = this.participantAliasIndex();
|
|
const currentParticipant = this.canonicalParticipant(aliasIndex, toDirectMessageParticipant(currentUser));
|
|
const peerParticipant = this.canonicalParticipant(aliasIndex, toDirectMessageParticipant(user));
|
|
const conversationId = getCanonicalDirectConversationId(aliasIndex, currentParticipant.userId, peerParticipant.userId);
|
|
const existingConversation = await this.collapseAliasConversations(ownerId, 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(
|
|
rawConversationId: string,
|
|
caller: DirectMessageParticipant,
|
|
participants: DirectMessageParticipant[],
|
|
timestamp = Date.now()
|
|
): Promise<void> {
|
|
const ownerId = this.getCurrentUserIdOrThrow();
|
|
const currentUser = this.requireCurrentUser();
|
|
const aliasIndex = this.participantAliasIndex();
|
|
const conversationId = canonicalizeDirectConversationId(aliasIndex, rawConversationId);
|
|
const currentParticipant = this.canonicalParticipant(aliasIndex, toDirectMessageParticipant(currentUser));
|
|
const canonicalCaller = this.canonicalParticipant(aliasIndex, caller);
|
|
const allParticipants = this.uniqueParticipants([
|
|
currentParticipant,
|
|
canonicalCaller,
|
|
...participants.map((participant) => this.canonicalParticipant(aliasIndex, participant))
|
|
]);
|
|
|
|
await this.loadForOwner(ownerId);
|
|
|
|
const existingConversation = await this.collapseAliasConversations(ownerId, conversationId)
|
|
?? this.createConversationForSystemEvent(conversationId, currentParticipant, canonicalCaller, allParticipants, timestamp);
|
|
const conversation = this.mergeConversationParticipants(existingConversation, allParticipants);
|
|
const message = createDirectCallStartedMessage(
|
|
conversation.id,
|
|
canonicalCaller,
|
|
conversation.participants.filter((participantId) => participantId !== canonicalCaller.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 aliasIndex = this.participantAliasIndex();
|
|
const currentParticipant = this.canonicalParticipant(aliasIndex, toDirectMessageParticipant(currentUser));
|
|
const sender = this.canonicalParticipant(aliasIndex, payload.sender);
|
|
const conversationId = payload.message.conversationId
|
|
? canonicalizeDirectConversationId(aliasIndex, payload.message.conversationId)
|
|
: getCanonicalDirectConversationId(aliasIndex, currentParticipant.userId, sender.userId);
|
|
const participants = this.uniqueParticipants([
|
|
currentParticipant,
|
|
sender,
|
|
...(payload.participants ?? []).map((participant) => this.canonicalParticipant(aliasIndex, participant))
|
|
]);
|
|
const existingConversation = await this.collapseAliasConversations(ownerId, conversationId)
|
|
?? (payload.conversationKind === 'group' || participants.length > 2
|
|
? createGroupConversation(conversationId, participants, payload.message.timestamp, payload.conversationTitle)
|
|
: {
|
|
...createDirectConversation(currentParticipant, sender, payload.message.timestamp),
|
|
id: conversationId
|
|
});
|
|
const conversationWithParticipants = this.mergeConversationParticipants(existingConversation, participants);
|
|
const incomingMessage: DirectMessage = {
|
|
...payload.message,
|
|
conversationId,
|
|
senderId: canonicalizeDirectParticipantId(aliasIndex, payload.message.senderId),
|
|
recipientId: canonicalizeDirectParticipantId(aliasIndex, payload.message.recipientId),
|
|
recipientIds: payload.message.recipientIds?.map((recipientId) =>
|
|
canonicalizeDirectParticipantId(aliasIndex, recipientId)),
|
|
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,
|
|
canonicalizeDirectConversationId(this.participantAliasIndex(), 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 aliasIndex = this.participantAliasIndex();
|
|
const conversationId = canonicalizeDirectConversationId(aliasIndex, payload.conversationId);
|
|
const senderId = canonicalizeDirectParticipantId(aliasIndex, payload.sender.userId);
|
|
const conversation = this.conversationsSignal().find((entry) => entry.id === conversationId);
|
|
|
|
if (!conversation
|
|
|| !directMessageConversationIncludesUser(conversation, selfUserIds)
|
|
|| !directMessageConversationIncludesUser(conversation, senderId)) {
|
|
return;
|
|
}
|
|
|
|
if (!payload.isTyping) {
|
|
this.typingEntriesSignal.update((entries) => entries.filter((entry) =>
|
|
!(entry.conversationId === conversationId && entry.userId === senderId)
|
|
));
|
|
|
|
return;
|
|
}
|
|
|
|
const nextEntry: DirectMessageTypingEntry = {
|
|
conversationId,
|
|
userId: senderId,
|
|
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,
|
|
canonicalizeDirectConversationId(this.participantAliasIndex(), 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 aliasIndex = this.participantAliasIndex();
|
|
const currentParticipant = this.canonicalParticipant(aliasIndex, toDirectMessageParticipant(currentUser));
|
|
const sender = this.canonicalParticipant(aliasIndex, payload.sender);
|
|
const selfUserIds = this.getSelfUserIds();
|
|
|
|
if (selfUserIds.has(payload.sender.userId)) {
|
|
return;
|
|
}
|
|
|
|
if (!directMessageSyncIncludesUser(payload, selfUserIds) || !directMessageSyncIncludesUser(payload, payload.sender.userId)) {
|
|
return;
|
|
}
|
|
|
|
const conversationId = canonicalizeDirectConversationId(aliasIndex, payload.conversationId);
|
|
const syncParticipants = payload.participants.map((participant) => this.canonicalParticipant(aliasIndex, participant));
|
|
const existingConversation = await this.collapseAliasConversations(ownerId, conversationId)
|
|
?? (payload.conversationKind === 'group' || payload.participants.length > 2
|
|
? createGroupConversation(conversationId, [currentParticipant, ...syncParticipants], payload.syncedAt, payload.conversationTitle)
|
|
: {
|
|
...createDirectConversation(currentParticipant, sender, payload.syncedAt),
|
|
id: conversationId
|
|
});
|
|
const participantProfiles = {
|
|
...existingConversation.participantProfiles,
|
|
...Object.fromEntries(syncParticipants.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,
|
|
conversationId,
|
|
senderId: canonicalizeDirectParticipantId(aliasIndex, message.senderId),
|
|
recipientId: canonicalizeDirectParticipantId(aliasIndex, message.recipientId),
|
|
recipientIds: message.recipientIds?.map((recipientId) => canonicalizeDirectParticipantId(aliasIndex, recipientId))
|
|
}, false),
|
|
syncBaseConversation
|
|
);
|
|
|
|
await this.persistConversation(ownerId, mergedConversation);
|
|
|
|
if (this.selectedConversationIdSignal() === conversationId) {
|
|
await this.markRead(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);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* One human can be addressed by several ids: their home id and one
|
|
* provisioned actor id per foreign signal server. Threads are stored under
|
|
* the home id, so every inbound id is resolved through this index first.
|
|
*/
|
|
private participantAliasIndex(): DirectParticipantAliasIndex {
|
|
const currentUser = this.currentUser();
|
|
const peerGroups = this.users().map((user) => ({
|
|
canonicalId: user.oderId || user.id,
|
|
aliasIds: [
|
|
user.id,
|
|
user.oderId,
|
|
user.peerId
|
|
].filter((aliasId): aliasId is string => !!aliasId)
|
|
}));
|
|
const selfGroups = currentUser
|
|
? [
|
|
{
|
|
canonicalId: currentUser.oderId || currentUser.id,
|
|
aliasIds: [...this.getSelfUserIds()]
|
|
}
|
|
]
|
|
: [];
|
|
|
|
return buildDirectParticipantAliasIndex([...peerGroups, ...selfGroups]);
|
|
}
|
|
|
|
private canonicalParticipant(
|
|
aliasIndex: DirectParticipantAliasIndex,
|
|
participant: DirectMessageParticipant
|
|
): DirectMessageParticipant {
|
|
const canonicalId = canonicalizeDirectParticipantId(aliasIndex, participant.userId);
|
|
|
|
return canonicalId === participant.userId ? participant : { ...participant, userId: canonicalId };
|
|
}
|
|
|
|
/**
|
|
* Fold any thread that resolves to `canonicalId` into a single stored
|
|
* conversation. Returns null when this human pair has no thread yet.
|
|
*/
|
|
private async collapseAliasConversations(
|
|
ownerId: string,
|
|
canonicalId: string
|
|
): Promise<DirectMessageConversation | null> {
|
|
await this.loadForOwner(ownerId);
|
|
|
|
const aliasIndex = this.participantAliasIndex();
|
|
const matching = this.conversationsSignal().filter((conversation) =>
|
|
canonicalizeDirectConversationId(aliasIndex, conversation.id) === canonicalId);
|
|
|
|
if (matching.length === 0) {
|
|
return null;
|
|
}
|
|
|
|
if (matching.length === 1 && matching[0].id === canonicalId) {
|
|
return matching[0];
|
|
}
|
|
|
|
const merged = mergeAliasDirectConversations(aliasIndex, matching);
|
|
const staleIds = matching.map((conversation) => conversation.id).filter((id) => id !== merged.id);
|
|
|
|
await this.persistConversation(ownerId, merged);
|
|
|
|
for (const staleId of staleIds) {
|
|
await this.repository.deleteConversation(ownerId, staleId);
|
|
}
|
|
|
|
this.conversationsSignal.update((conversations) => conversations.filter((conversation) =>
|
|
!staleIds.includes(conversation.id)));
|
|
|
|
if (staleIds.includes(this.selectedConversationIdSignal() ?? '')) {
|
|
this.selectedConversationIdSignal.set(merged.id);
|
|
}
|
|
|
|
return merged;
|
|
}
|
|
|
|
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);
|
|
}
|
|
}
|