feat(presence): live stream badges, honest ring delivery, cross-signal DMs
- 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.
This commit is contained in:
+158
-39
@@ -1,4 +1,4 @@
|
||||
/* eslint-disable @typescript-eslint/member-ordering */
|
||||
|
||||
import {
|
||||
Injectable,
|
||||
computed,
|
||||
@@ -24,12 +24,20 @@ import {
|
||||
directMessageConversationIncludesUser,
|
||||
directMessageEventIncludesUser,
|
||||
directMessageSyncIncludesUser,
|
||||
getDirectConversationId,
|
||||
isGroupDirectConversation,
|
||||
updateMessageStatusInConversation,
|
||||
upsertDirectMessage
|
||||
} from '../../domain/logic/direct-message.logic';
|
||||
import { collectDirectMessageSelfUserIds, isSelfDirectMessageSender } from '../../domain/logic/direct-message-identity.rules';
|
||||
import {
|
||||
buildDirectParticipantAliasIndex,
|
||||
canonicalizeDirectConversationId,
|
||||
canonicalizeDirectParticipantId,
|
||||
collectDirectMessageSelfUserIds,
|
||||
getCanonicalDirectConversationId,
|
||||
isSelfDirectMessageSender,
|
||||
mergeAliasDirectConversations,
|
||||
type DirectParticipantAliasIndex
|
||||
} from '../../domain/logic/direct-message-identity.rules';
|
||||
import {
|
||||
DirectMessage,
|
||||
DirectMessageConversation,
|
||||
@@ -48,7 +56,7 @@ import type {
|
||||
Reaction,
|
||||
User
|
||||
} from '../../../../shared-kernel';
|
||||
import { selectCurrentUser } from '../../../../store/users/users.selectors';
|
||||
import { selectAllUsers, selectCurrentUser } from '../../../../store/users/users.selectors';
|
||||
|
||||
const DIRECT_MESSAGE_SYNC_LIMIT = 1000;
|
||||
const DIRECT_MESSAGE_SYNC_REQUEST_COOLDOWN_MS = 5000;
|
||||
@@ -75,6 +83,7 @@ export class DirectMessageService {
|
||||
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[]>([]);
|
||||
@@ -128,10 +137,11 @@ export class DirectMessageService {
|
||||
|
||||
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);
|
||||
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);
|
||||
@@ -258,30 +268,32 @@ export class DirectMessageService {
|
||||
}
|
||||
|
||||
async recordCallStarted(
|
||||
conversationId: string,
|
||||
rawConversationId: string,
|
||||
caller: DirectMessageParticipant,
|
||||
participants: DirectMessageParticipant[],
|
||||
timestamp = Date.now()
|
||||
): Promise<void> {
|
||||
const ownerId = this.getCurrentUserIdOrThrow();
|
||||
const currentUser = this.requireCurrentUser();
|
||||
const currentParticipant = toDirectMessageParticipant(currentUser);
|
||||
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,
|
||||
caller,
|
||||
...participants
|
||||
canonicalCaller,
|
||||
...participants.map((participant) => this.canonicalParticipant(aliasIndex, participant))
|
||||
]);
|
||||
|
||||
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 existingConversation = await this.collapseAliasConversations(ownerId, conversationId)
|
||||
?? this.createConversationForSystemEvent(conversationId, currentParticipant, canonicalCaller, allParticipants, timestamp);
|
||||
const conversation = this.mergeConversationParticipants(existingConversation, allParticipants);
|
||||
const message = createDirectCallStartedMessage(
|
||||
conversation.id,
|
||||
caller,
|
||||
conversation.participants.filter((participantId) => participantId !== caller.userId),
|
||||
canonicalCaller,
|
||||
conversation.participants.filter((participantId) => participantId !== canonicalCaller.userId),
|
||||
timestamp
|
||||
);
|
||||
|
||||
@@ -517,23 +529,32 @@ export class DirectMessageService {
|
||||
return;
|
||||
}
|
||||
|
||||
const currentParticipant = toDirectMessageParticipant(currentUser);
|
||||
const sender = payload.sender;
|
||||
const aliasIndex = this.participantAliasIndex();
|
||||
const currentParticipant = this.canonicalParticipant(aliasIndex, toDirectMessageParticipant(currentUser));
|
||||
const sender = this.canonicalParticipant(aliasIndex, payload.sender);
|
||||
const conversationId = payload.message.conversationId
|
||||
|| getDirectConversationId(currentParticipant.userId, sender.userId);
|
||||
? canonicalizeDirectConversationId(aliasIndex, payload.message.conversationId)
|
||||
: getCanonicalDirectConversationId(aliasIndex, currentParticipant.userId, sender.userId);
|
||||
const participants = this.uniqueParticipants([
|
||||
currentParticipant,
|
||||
sender,
|
||||
...(payload.participants ?? [])
|
||||
...(payload.participants ?? []).map((participant) => this.canonicalParticipant(aliasIndex, participant))
|
||||
]);
|
||||
const existingConversation = this.conversationsSignal().find((conversation) => conversation.id === conversationId)
|
||||
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));
|
||||
: {
|
||||
...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);
|
||||
@@ -590,7 +611,10 @@ export class DirectMessageService {
|
||||
|
||||
private async handleIncomingMutation(payload: DirectMessageMutationEventPayload): Promise<void> {
|
||||
const ownerId = this.getCurrentUserIdOrThrow();
|
||||
const conversation = await this.findConversation(ownerId, payload.conversationId);
|
||||
const conversation = await this.findConversation(
|
||||
ownerId,
|
||||
canonicalizeDirectConversationId(this.participantAliasIndex(), payload.conversationId)
|
||||
);
|
||||
const selfUserIds = this.getSelfUserIds();
|
||||
|
||||
if (!conversation || !directMessageConversationIncludesUser(conversation, selfUserIds)) {
|
||||
@@ -607,25 +631,28 @@ export class DirectMessageService {
|
||||
return;
|
||||
}
|
||||
|
||||
const conversation = this.conversationsSignal().find((entry) => entry.id === payload.conversationId);
|
||||
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, payload.sender.userId)) {
|
||||
|| !directMessageConversationIncludesUser(conversation, senderId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (!payload.isTyping) {
|
||||
this.typingEntriesSignal.update((entries) => entries.filter((entry) =>
|
||||
!(entry.conversationId === payload.conversationId && entry.userId === payload.sender.userId)
|
||||
!(entry.conversationId === conversationId && entry.userId === senderId)
|
||||
));
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
const nextEntry: DirectMessageTypingEntry = {
|
||||
conversationId: payload.conversationId,
|
||||
userId: payload.sender.userId,
|
||||
conversationId,
|
||||
userId: senderId,
|
||||
displayName: payload.sender.displayName,
|
||||
expiresAt: Date.now() + DIRECT_MESSAGE_TYPING_TTL_MS
|
||||
};
|
||||
@@ -641,7 +668,10 @@ export class DirectMessageService {
|
||||
private async handleIncomingSyncRequest(payload: DirectMessageSyncRequestEventPayload): Promise<void> {
|
||||
const ownerId = this.getCurrentUserIdOrThrow();
|
||||
const currentUser = this.requireCurrentUser();
|
||||
const conversation = await this.findConversation(ownerId, payload.conversationId);
|
||||
const conversation = await this.findConversation(
|
||||
ownerId,
|
||||
canonicalizeDirectConversationId(this.participantAliasIndex(), payload.conversationId)
|
||||
);
|
||||
const selfUserIds = this.getSelfUserIds();
|
||||
|
||||
if (!conversation
|
||||
@@ -668,7 +698,9 @@ export class DirectMessageService {
|
||||
private async handleIncomingSync(payload: DirectMessageSyncEventPayload): Promise<void> {
|
||||
const ownerId = this.getCurrentUserIdOrThrow();
|
||||
const currentUser = this.requireCurrentUser();
|
||||
const currentParticipant = toDirectMessageParticipant(currentUser);
|
||||
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)) {
|
||||
@@ -679,14 +711,18 @@ export class DirectMessageService {
|
||||
return;
|
||||
}
|
||||
|
||||
const existingConversation = this.conversationsSignal().find((conversation) => conversation.id === payload.conversationId)
|
||||
?? await this.repository.getConversation(ownerId, payload.conversationId)
|
||||
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(payload.conversationId, [currentParticipant, ...payload.participants], payload.syncedAt, payload.conversationTitle)
|
||||
: createDirectConversation(currentParticipant, payload.sender, payload.syncedAt));
|
||||
? createGroupConversation(conversationId, [currentParticipant, ...syncParticipants], payload.syncedAt, payload.conversationTitle)
|
||||
: {
|
||||
...createDirectConversation(currentParticipant, sender, payload.syncedAt),
|
||||
id: conversationId
|
||||
});
|
||||
const participantProfiles = {
|
||||
...existingConversation.participantProfiles,
|
||||
...Object.fromEntries(payload.participants.map((participant) => [participant.userId, participant])),
|
||||
...Object.fromEntries(syncParticipants.map((participant) => [participant.userId, participant])),
|
||||
[currentParticipant.userId]: currentParticipant
|
||||
};
|
||||
const syncBaseConversation: DirectMessageConversation = {
|
||||
@@ -697,14 +733,20 @@ export class DirectMessageService {
|
||||
participantProfiles
|
||||
};
|
||||
const mergedConversation = payload.messages.reduce<DirectMessageConversation>(
|
||||
(conversation, message) => upsertDirectMessage(conversation, message, false),
|
||||
(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() === payload.conversationId) {
|
||||
await this.markRead(payload.conversationId);
|
||||
if (this.selectedConversationIdSignal() === conversationId) {
|
||||
await this.markRead(conversationId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -912,6 +954,83 @@ export class DirectMessageService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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[]
|
||||
|
||||
Reference in New Issue
Block a user