Files
Toju/toju-app/src/app/store/rooms/room-state-sync.effects.ts
2026-04-11 13:07:46 +02:00

815 lines
27 KiB
TypeScript

/* eslint-disable @typescript-eslint/member-ordering */
import { Injectable, inject } from '@angular/core';
import {
Actions,
createEffect,
ofType
} from '@ngrx/effects';
import { Action, Store } from '@ngrx/store';
import {
of,
from,
EMPTY
} from 'rxjs';
import {
map,
mergeMap,
withLatestFrom,
tap,
switchMap,
catchError
} from 'rxjs/operators';
import { RoomsActions } from './rooms.actions';
import { UsersActions } from '../users/users.actions';
import { selectCurrentUser, selectAllUsers } from '../users/users.selectors';
import {
selectActiveChannelId,
selectCurrentRoom,
selectSavedRooms
} from './rooms.selectors';
import { RealtimeSessionFacade } from '../../core/realtime';
import { DatabaseService } from '../../infrastructure/persistence';
import { resolveRoomPermission } from '../../domains/access-control';
import {
ChatEvent,
Room,
RoomSettings,
RoomPermissions,
BanEntry,
User,
VoiceState
} from '../../shared-kernel';
import { NotificationAudioService, AppSound } from '../../core/services/notification-audio.service';
import { hasRoomBanForUser } from '../../core/helpers/room-ban.helpers';
import { RECONNECT_SOUND_GRACE_MS } from '../../core/constants';
import { VoiceSessionFacade } from '../../domains/voice-session';
import {
buildSignalingUser,
buildKnownUserExtras,
isWrongServer,
resolveRoom,
sanitizeRoomSnapshot,
normalizeIncomingBans,
getPersistedCurrentUserId,
RoomPresenceSignalingMessage
} from './rooms.helpers';
/**
* NgRx effects for real-time state synchronisation: signaling presence
* events (server_users, user_joined, user_left, access_denied), P2P
* room-state / icon sync, and voice/screen/camera state broadcasts.
*/
@Injectable()
export class RoomStateSyncEffects {
private actions$ = inject(Actions);
private store = inject(Store);
private webrtc = inject(RealtimeSessionFacade);
private db = inject(DatabaseService);
private audioService = inject(NotificationAudioService);
private voiceSessionService = inject(VoiceSessionFacade);
/**
* Tracks user IDs we already know are in voice. Lives outside the
* NgRx store so it survives room switches and presence re-syncs,
* preventing false join/leave sounds during state refreshes.
*/
private knownVoiceUsers = new Set<string>();
/**
* When a user leaves (e.g. socket drops), record the timestamp so
* that a rapid re-join (reconnect) does not trigger a false
* join/leave sound within {@link RECONNECT_SOUND_GRACE_MS}.
*/
private recentlyLeftVoiceTimestamps = new Map<string, number>();
// ── Signaling presence ─────────────────────────────────────────
/** Handles WebRTC signaling events for user presence (join, leave, server_users). */
signalingMessages$ = createEffect(() =>
this.webrtc.onSignalingMessage.pipe(
withLatestFrom(
this.store.select(selectCurrentUser),
this.store.select(selectCurrentRoom),
this.store.select(selectSavedRooms)
),
mergeMap(([
message,
currentUser,
currentRoom,
savedRooms
]) => {
const signalingMessage: RoomPresenceSignalingMessage = message;
const myId = currentUser?.oderId || currentUser?.id;
const viewedServerId = currentRoom?.id;
const room = resolveRoom(signalingMessage.serverId, currentRoom, savedRooms);
const shouldClearReconnectFlag = !isWrongServer(signalingMessage.serverId, viewedServerId);
switch (signalingMessage.type) {
case 'server_users': {
if (!Array.isArray(signalingMessage.users) || !signalingMessage.serverId)
return EMPTY;
const syncedUsers = signalingMessage.users
.filter((user) => user.oderId !== myId)
.map((user) =>
buildSignalingUser(user, {
...buildKnownUserExtras(room, user.oderId),
presenceServerIds: [signalingMessage.serverId]
})
);
const actions: Action[] = [
UsersActions.syncServerPresence({
roomId: signalingMessage.serverId,
users: syncedUsers
})
];
if (shouldClearReconnectFlag) {
actions.unshift(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
}
return actions;
}
case 'user_joined': {
if (!signalingMessage.serverId || signalingMessage.oderId === myId)
return EMPTY;
if (!signalingMessage.oderId)
return EMPTY;
const joinedUser = {
oderId: signalingMessage.oderId,
displayName: signalingMessage.displayName
};
const actions: Action[] = [
UsersActions.userJoined({
user: buildSignalingUser(joinedUser, {
...buildKnownUserExtras(room, joinedUser.oderId),
presenceServerIds: [signalingMessage.serverId]
})
})
];
if (shouldClearReconnectFlag) {
actions.unshift(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
}
return actions;
}
case 'user_left': {
if (!signalingMessage.oderId)
return EMPTY;
const remainingServerIds = Array.isArray(signalingMessage.serverIds)
? signalingMessage.serverIds
: undefined;
if (!remainingServerIds || remainingServerIds.length === 0) {
if (this.knownVoiceUsers.has(signalingMessage.oderId)) {
this.recentlyLeftVoiceTimestamps.set(signalingMessage.oderId, Date.now());
}
this.knownVoiceUsers.delete(signalingMessage.oderId);
}
const actions: Action[] = [
UsersActions.userLeft({
userId: signalingMessage.oderId,
serverId: signalingMessage.serverId,
serverIds: remainingServerIds
})
];
if (shouldClearReconnectFlag) {
actions.unshift(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
}
return actions;
}
case 'access_denied': {
if (isWrongServer(signalingMessage.serverId, viewedServerId))
return EMPTY;
if (signalingMessage.reason !== 'SERVER_NOT_FOUND')
return EMPTY;
// When multiple signal URLs are configured, the room may already
// be successfully joined on a different signal server. Only show
// the reconnect notice when the room is not reachable at all.
if (signalingMessage.serverId && this.webrtc.hasJoinedServer(signalingMessage.serverId))
return EMPTY;
return [RoomsActions.setSignalServerReconnecting({ isReconnecting: true })];
}
default:
return EMPTY;
}
})
)
);
// ── P2P state sync ─────────────────────────────────────────────
/** Request a full room-state snapshot whenever a peer data channel opens. */
peerConnectedServerStateSync$ = createEffect(
() =>
this.webrtc.onPeerConnected.pipe(
withLatestFrom(this.store.select(selectCurrentRoom)),
tap(([peerId, room]) => {
if (!room)
return;
this.webrtc.sendToPeer(peerId, {
type: 'server-state-request',
roomId: room.id
});
})
),
{ dispatch: false }
);
/** Re-request the latest room-state snapshot whenever the user enters or views a server. */
roomEntryServerStateSync$ = createEffect(
() =>
this.actions$.pipe(
ofType(
RoomsActions.createRoomSuccess,
RoomsActions.joinRoomSuccess,
RoomsActions.viewServerSuccess
),
tap(({ room }) => {
for (const peerId of this.webrtc.getConnectedPeers()) {
try {
this.webrtc.sendToPeer(peerId, {
type: 'server-state-request',
roomId: room.id
});
} catch {
/* peer may have disconnected */
}
}
})
),
{ dispatch: false }
);
/** Processes incoming P2P room-state, room-sync, and icon-sync events. */
incomingRoomEvents$ = createEffect(() =>
this.webrtc.onMessageReceived.pipe(
withLatestFrom(
this.store.select(selectCurrentRoom),
this.store.select(selectSavedRooms),
this.store.select(selectAllUsers),
this.store.select(selectCurrentUser),
this.store.select(selectActiveChannelId)
),
mergeMap(([
event,
currentRoom,
savedRooms,
allUsers,
currentUser,
activeChannelId
]) => {
switch (event.type) {
case 'voice-state':
return this.handleVoiceOrScreenState(event, allUsers, currentUser ?? null, 'voice');
case 'voice-channel-move':
return this.handleVoiceChannelMove(event, currentRoom, savedRooms, currentUser ?? null);
case 'screen-state':
return this.handleVoiceOrScreenState(event, allUsers, currentUser ?? null, 'screen');
case 'camera-state':
return this.handleVoiceOrScreenState(event, allUsers, currentUser ?? null, 'camera');
case 'server-state-request':
return this.handleServerStateRequest(event, currentRoom, savedRooms);
case 'server-state-full':
return this.handleServerStateFull(event, currentRoom, savedRooms, currentUser ?? null);
case 'room-settings-update':
return this.handleRoomSettingsUpdate(event, currentRoom, savedRooms);
case 'room-permissions-update':
return this.handleRoomPermissionsUpdate(event, currentRoom, savedRooms);
case 'channels-update':
return this.handleChannelsUpdate(event, currentRoom, savedRooms, activeChannelId);
case 'server-icon-summary':
return this.handleIconSummary(event, currentRoom, savedRooms);
case 'server-icon-request':
return this.handleIconRequest(event, currentRoom, savedRooms);
case 'server-icon-full':
case 'server-icon-update':
return this.handleIconData(event, currentRoom, savedRooms);
default:
return EMPTY;
}
})
)
);
/** Broadcasts the local server icon summary to peers when a new peer connects. */
peerConnectedIconSync$ = createEffect(
() =>
this.webrtc.onPeerConnected.pipe(
withLatestFrom(this.store.select(selectCurrentRoom)),
tap(([_peerId, room]) => {
if (!room)
return;
const iconUpdatedAt = room.iconUpdatedAt || 0;
this.webrtc.broadcastMessage({
type: 'server-icon-summary',
roomId: room.id,
iconUpdatedAt
});
})
),
{ dispatch: false }
);
// ── Voice / Screen / Camera handlers ───────────────────────────
private handleVoiceOrScreenState(
event: ChatEvent,
allUsers: User[],
currentUser: User | null,
kind: 'voice' | 'screen' | 'camera'
) {
const userId: string | undefined = event.fromPeerId ?? event.oderId;
if (!userId)
return EMPTY;
const existingUser = allUsers.find((user) => user.id === userId || user.oderId === userId);
const userExists = !!existingUser;
if (kind === 'voice') {
const vs = event.voiceState as Partial<VoiceState> | undefined;
if (!vs)
return EMPTY;
const presenceRefreshAction = vs.serverId && !existingUser?.presenceServerIds?.includes(vs.serverId)
? UsersActions.userJoined({
user: buildSignalingUser(
{ oderId: userId,
displayName: event.displayName || existingUser?.displayName || 'User' },
{ presenceServerIds: [vs.serverId] }
)
})
: null;
// Detect voice-connection transitions to play join/leave sounds.
const weAreInVoice = this.webrtc.isVoiceConnected();
const nowConnected = vs.isConnected ?? false;
const wasKnown = this.knownVoiceUsers.has(userId);
const wasInCurrentVoiceRoom = this.isSameVoiceRoom(existingUser?.voiceState, currentUser?.voiceState);
const mergedVoiceState = { ...existingUser?.voiceState, ...vs };
const isInCurrentVoiceRoom = this.isSameVoiceRoom(mergedVoiceState, currentUser?.voiceState);
if (weAreInVoice) {
const isReconnect = this.consumeRecentLeave(userId);
if (!isReconnect) {
if (((!wasKnown && isInCurrentVoiceRoom) || (userExists && !wasInCurrentVoiceRoom && isInCurrentVoiceRoom)) && nowConnected) {
this.audioService.play(AppSound.Joining);
} else if (wasInCurrentVoiceRoom && !isInCurrentVoiceRoom) {
this.audioService.play(AppSound.Leave);
}
}
}
if (nowConnected) {
this.knownVoiceUsers.add(userId);
} else {
this.knownVoiceUsers.delete(userId);
}
if (!userExists) {
return of(
UsersActions.userJoined({
user: buildSignalingUser(
{ oderId: userId,
displayName: event.displayName || 'User' },
{
presenceServerIds: vs.serverId ? [vs.serverId] : undefined,
voiceState: {
isConnected: vs.isConnected ?? false,
isMuted: vs.isMuted ?? false,
isDeafened: vs.isDeafened ?? false,
isSpeaking: vs.isSpeaking ?? false,
isMutedByAdmin: vs.isMutedByAdmin,
volume: vs.volume,
roomId: vs.roomId,
serverId: vs.serverId
}
}
)
})
);
}
const actions: Action[] = [];
if (presenceRefreshAction) {
actions.push(presenceRefreshAction);
}
actions.push(UsersActions.updateVoiceState({ userId,
voiceState: vs }));
return actions;
}
if (kind === 'screen') {
const isSharing = event.isScreenSharing as boolean | undefined;
if (isSharing === undefined)
return EMPTY;
if (!userExists) {
return of(
UsersActions.userJoined({
user: buildSignalingUser(
{ oderId: userId,
displayName: event.displayName || 'User' },
{ screenShareState: { isSharing } }
)
})
);
}
return of(
UsersActions.updateScreenShareState({
userId,
screenShareState: { isSharing }
})
);
}
const isCameraEnabled = event.isCameraEnabled as boolean | undefined;
if (isCameraEnabled === undefined)
return EMPTY;
if (!userExists) {
return of(
UsersActions.userJoined({
user: buildSignalingUser(
{ oderId: userId,
displayName: event.displayName || 'User' },
{ cameraState: { isEnabled: isCameraEnabled } }
)
})
);
}
return of(
UsersActions.updateCameraState({
userId,
cameraState: { isEnabled: isCameraEnabled }
})
);
}
private handleVoiceChannelMove(
event: ChatEvent,
currentRoom: Room | null,
savedRooms: Room[],
currentUser: User | null
) {
const targetUserId = typeof event.targetUserId === 'string' ? event.targetUserId : null;
const serverId = typeof event.roomId === 'string' ? event.roomId : currentUser?.voiceState?.serverId;
const nextVoiceState = event.voiceState as Partial<VoiceState> | undefined;
if (!currentUser || !targetUserId || !serverId || !nextVoiceState?.roomId) {
return EMPTY;
}
if (targetUserId !== currentUser.id && targetUserId !== currentUser.oderId) {
return EMPTY;
}
const room = resolveRoom(serverId, currentRoom, savedRooms);
const movedChannel = room?.channels?.find((channel) => channel.id === nextVoiceState.roomId && channel.type === 'voice');
if (!room || !movedChannel) {
return EMPTY;
}
const updatedVoiceState: Partial<VoiceState> = {
isConnected: true,
isMuted: currentUser.voiceState?.isMuted ?? false,
isDeafened: currentUser.voiceState?.isDeafened ?? false,
isSpeaking: currentUser.voiceState?.isSpeaking ?? false,
isMutedByAdmin: currentUser.voiceState?.isMutedByAdmin,
volume: currentUser.voiceState?.volume,
roomId: movedChannel.id,
serverId: room.id
};
const wasViewingVoiceServer = this.voiceSessionService.isViewingVoiceServer();
this.webrtc.startVoiceHeartbeat(movedChannel.id, room.id);
this.voiceSessionService.startSession({
serverId: room.id,
serverName: room.name,
roomId: movedChannel.id,
roomName: `🔊 ${movedChannel.name}`,
serverIcon: room.icon,
serverDescription: room.description,
serverRoute: `/room/${room.id}`
});
this.voiceSessionService.setViewingVoiceServer(wasViewingVoiceServer);
this.webrtc.broadcastMessage({
type: 'voice-state',
oderId: currentUser.oderId || currentUser.id,
displayName: currentUser.displayName || 'User',
voiceState: updatedVoiceState
});
return of(UsersActions.updateVoiceState({
userId: currentUser.id,
voiceState: updatedVoiceState
}));
}
private isSameVoiceRoom(
voiceState: Partial<VoiceState> | undefined,
currentUserVoiceState: Partial<VoiceState> | undefined
): boolean {
return !!voiceState?.isConnected
&& !!currentUserVoiceState?.isConnected
&& !!voiceState.roomId
&& !!voiceState.serverId
&& voiceState.roomId === currentUserVoiceState.roomId
&& voiceState.serverId === currentUserVoiceState.serverId;
}
/**
* Returns `true` and cleans up the entry if the given user left
* recently enough to be considered a reconnect.
*/
private consumeRecentLeave(userId: string): boolean {
const now = Date.now();
for (const [id, ts] of this.recentlyLeftVoiceTimestamps) {
if (now - ts > RECONNECT_SOUND_GRACE_MS) {
this.recentlyLeftVoiceTimestamps.delete(id);
}
}
const leaveTs = this.recentlyLeftVoiceTimestamps.get(userId);
if (leaveTs !== undefined && now - leaveTs <= RECONNECT_SOUND_GRACE_MS) {
this.recentlyLeftVoiceTimestamps.delete(userId);
return true;
}
return false;
}
// ── Server-state sync handlers ─────────────────────────────────
private handleServerStateRequest(event: ChatEvent, currentRoom: Room | null, savedRooms: Room[]) {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
const fromPeerId = event.fromPeerId;
if (!room || !fromPeerId)
return EMPTY;
return from(this.db.getBansForRoom(room.id)).pipe(
tap((bans) => {
this.webrtc.sendToPeer(fromPeerId, {
type: 'server-state-full',
roomId: room.id,
room: sanitizeRoomSnapshot(room),
bans
});
}),
mergeMap(() => EMPTY)
);
}
private handleServerStateFull(
event: ChatEvent,
currentRoom: Room | null,
savedRooms: Room[],
currentUser: { id: string; oderId: string } | null
) {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
const incomingRoom = event.room as Partial<Room> | undefined;
if (!room || !incomingRoom)
return EMPTY;
const roomChanges = sanitizeRoomSnapshot(incomingRoom);
const bans = normalizeIncomingBans(room.id, event.bans);
return this.syncBansToLocalRoom(room.id, bans).pipe(
mergeMap(() => {
const actions: (ReturnType<typeof RoomsActions.updateRoom>
| ReturnType<typeof UsersActions.loadBansSuccess>
| ReturnType<typeof RoomsActions.forgetRoom>)[] = [
RoomsActions.updateRoom({
roomId: room.id,
changes: roomChanges
})
];
const isCurrentUserBanned = hasRoomBanForUser(
bans,
currentUser,
getPersistedCurrentUserId()
);
if (currentRoom?.id === room.id) {
actions.push(UsersActions.loadBansSuccess({ bans }));
}
if (isCurrentUserBanned) {
actions.push(RoomsActions.forgetRoom({ roomId: room.id }));
}
return actions;
}),
catchError(() => EMPTY)
);
}
private handleRoomSettingsUpdate(event: ChatEvent, currentRoom: Room | null, savedRooms: Room[]) {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
const settings = event.settings as Partial<RoomSettings> | undefined;
if (!room || !settings)
return EMPTY;
return of(
RoomsActions.updateRoom({
roomId: room.id,
changes: {
name: settings.name ?? room.name,
description: settings.description ?? room.description,
topic: settings.topic ?? room.topic,
isPrivate: settings.isPrivate ?? room.isPrivate,
password: settings.password === '' ? undefined : room.password,
hasPassword:
typeof settings.hasPassword === 'boolean'
? settings.hasPassword
: (typeof room.hasPassword === 'boolean' ? room.hasPassword : !!room.password),
maxUsers: settings.maxUsers ?? room.maxUsers
}
})
);
}
private handleRoomPermissionsUpdate(event: ChatEvent, currentRoom: Room | null, savedRooms: Room[]) {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
const permissions = event.permissions as Partial<RoomPermissions> | undefined;
const incomingRoom = event.room as Partial<Room> | undefined;
if (!room || (!permissions && !incomingRoom))
return EMPTY;
return of(
RoomsActions.updateRoom({
roomId: room.id,
changes: {
permissions: permissions
? { ...(room.permissions || {}),
...permissions } as RoomPermissions
: room.permissions,
roles: Array.isArray(incomingRoom?.roles) ? incomingRoom.roles : room.roles,
roleAssignments: Array.isArray(incomingRoom?.roleAssignments) ? incomingRoom.roleAssignments : room.roleAssignments,
channelPermissions: Array.isArray(incomingRoom?.channelPermissions) ? incomingRoom.channelPermissions : room.channelPermissions,
slowModeInterval: typeof incomingRoom?.slowModeInterval === 'number' ? incomingRoom.slowModeInterval : room.slowModeInterval
}
})
);
}
private handleChannelsUpdate(
event: ChatEvent,
currentRoom: Room | null,
savedRooms: Room[],
activeChannelId: string
): Action[] {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
const channels = Array.isArray(event.channels) ? event.channels : null;
if (!room || !channels) {
return [];
}
const actions: Action[] = [
RoomsActions.updateRoom({
roomId: room.id,
changes: { channels }
})
];
if (!channels.some((channel) => channel.id === activeChannelId)) {
const fallbackChannelId = channels.find((channel) => channel.type === 'text')?.id
?? 'general';
actions.push(RoomsActions.selectChannel({ channelId: fallbackChannelId }));
}
return actions;
}
// ── Icon sync handlers ─────────────────────────────────────────
private handleIconSummary(event: ChatEvent, currentRoom: Room | null, savedRooms: Room[]) {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
if (!room)
return EMPTY;
const remoteUpdated = event.iconUpdatedAt || 0;
const localUpdated = room.iconUpdatedAt || 0;
if (remoteUpdated > localUpdated && event.fromPeerId) {
this.webrtc.sendToPeer(event.fromPeerId, {
type: 'server-icon-request',
roomId: room.id
});
}
return EMPTY;
}
private handleIconRequest(event: ChatEvent, currentRoom: Room | null, savedRooms: Room[]) {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
if (!room)
return EMPTY;
if (event.fromPeerId) {
this.webrtc.sendToPeer(event.fromPeerId, {
type: 'server-icon-full',
roomId: room.id,
icon: room.icon,
iconUpdatedAt: room.iconUpdatedAt || 0
});
}
return EMPTY;
}
private handleIconData(event: ChatEvent, currentRoom: Room | null, savedRooms: Room[]) {
const roomId = typeof event.roomId === 'string' ? event.roomId : currentRoom?.id;
const room = resolveRoom(roomId, currentRoom, savedRooms);
const senderId = event.fromPeerId;
if (!room || typeof event.icon !== 'string' || !senderId)
return EMPTY;
return this.store.select(selectAllUsers).pipe(
map((users) => users.find((user) => user.id === senderId)),
mergeMap((sender) => {
if (!sender)
return EMPTY;
const isOwner = room.hostId === sender.id;
const canByRole = resolveRoomPermission(room, sender, 'manageIcon');
if (!isOwner && !canByRole)
return EMPTY;
const updates: Partial<Room> = {
icon: event.icon,
iconUpdatedAt: event.iconUpdatedAt || Date.now()
};
this.db.updateRoom(room.id, updates);
return of(RoomsActions.updateRoom({ roomId: room.id,
changes: updates }));
})
);
}
// ── Internal helpers ───────────────────────────────────────────
private syncBansToLocalRoom(roomId: string, bans: BanEntry[]) {
return from(this.db.getBansForRoom(roomId)).pipe(
switchMap((localBans) => {
const nextIds = new Set(bans.map((ban) => ban.oderId));
const removals = localBans
.filter((ban) => !nextIds.has(ban.oderId))
.map((ban) => this.db.removeBan(ban.oderId));
const saves = bans.map((ban) => this.db.saveBan({ ...ban,
roomId }));
return from(Promise.all([...removals, ...saves]));
})
);
}
}