myxelium/Toju
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects 2 Releases 36 Wiki Activity

fix: restore build and stabilize E2E cross-signal behavior

Browse Source 
Revert the automated member-ordering pass that broke Angular field init
(TS2729) and disable that rule until a safe reorder strategy exists.
Fix modal/confirm dialog i18n defaults via template fallbacks, search all
active endpoints (including offline), register foreign rooms with actor
owner IDs, sync profile display names from avatar summaries, and guard
dm-chat when a private call converts to a group conversation.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
SocksOnHead
2026-06-11 12:16:40 +02:00
parent 79c6f91cd6
commit 31962aeb1a
131 changed files with 2483 additions and 3896 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
toju-app/src/app/infrastructure/mobile/adapters/capacitor/capacitor-mobile-persistence.adapter.ts
View File

@@ -8,15 +8,14 @@ import { MobileSqliteConnectionService } from '../../services/mobile-sqlite-conn
* Domain persistence routes through {@link CapacitorDatabaseService} on Capacitor shells.
*/
export class CapacitorMobilePersistenceAdapter implements MobilePersistenceAdapter {
private initialized = false;
constructor(private readonly connection: MobileSqliteConnectionService) {}
get isNativeSqlite(): boolean {
return this.connection.isAvailable;
}
private initialized = false;
constructor(private readonly connection: MobileSqliteConnectionService) {}
async initialize(): Promise<void> {
if (this.initialized) {
return;
@@ -33,5 +32,4 @@ export class CapacitorMobilePersistenceAdapter implements MobilePersistenceAdapt
this.initialized = true;
console.info('[mobile] native SQLite persistence initialized');
}
}

20
toju-app/src/app/infrastructure/mobile/services/mobile-app-update.service.ts
View File

@@ -22,22 +22,17 @@ const DEFAULT_POLL_INTERVAL_MS = 30 * 60_000;
export class MobileAppUpdateService {
readonly state = signal<MobileUpdateState>(createInitialMobileUpdateState());
private readonly appI18n = inject(AppI18nService);
private readonly mobilePlatform = inject(MobilePlatformService);
private adapter: MobileAppUpdateAdapter = new WebMobileAppUpdateAdapter();
private adapterReady: Promise<MobileAppUpdateAdapter> | null = null;
private initialized = false;
private pollTimerId: number | null = null;
get isCapacitor(): boolean {
return this.mobilePlatform.isCapacitor();
}
private readonly appI18n = inject(AppI18nService);
private readonly mobilePlatform = inject(MobilePlatformService);
private adapter: MobileAppUpdateAdapter = new WebMobileAppUpdateAdapter();
private adapterReady: Promise<MobileAppUpdateAdapter> | null = null;
private initialized = false;
private pollTimerId: number | null = null;
async initialize(): Promise<void> {
if (this.initialized) {
return;
@@ -175,5 +170,4 @@ export class MobileAppUpdateService {
void this.checkForUpdates();
}, DEFAULT_POLL_INTERVAL_MS);
}
}

13
toju-app/src/app/infrastructure/mobile/services/mobile-persistence.service.ts
View File

@@ -9,19 +9,15 @@ import { MobileSqliteConnectionService } from './mobile-sqlite-connection.servic
/** Facade for native SQLite persistence on mobile shells. */
@Injectable({ providedIn: 'root' })
export class MobilePersistenceService {
private readonly mobilePlatform = inject(MobilePlatformService);
private readonly sqliteConnection = inject(MobileSqliteConnectionService);
private adapter: MobilePersistenceAdapter = new WebMobilePersistenceAdapter();
private adapterReady: Promise<MobilePersistenceAdapter> | null = null;
get isNativeSqlite(): boolean {
return this.adapter.isNativeSqlite;
}
private readonly mobilePlatform = inject(MobilePlatformService);
private readonly sqliteConnection = inject(MobileSqliteConnectionService);
private adapter: MobilePersistenceAdapter = new WebMobilePersistenceAdapter();
private adapterReady: Promise<MobilePersistenceAdapter> | null = null;
initialize(): Promise<void> {
return this.ensureAdapter().then((adapter) => adapter.initialize());
}
@@ -44,5 +40,4 @@ export class MobilePersistenceService {
return this.adapterReady;
}
}

13
toju-app/src/app/infrastructure/mobile/services/mobile-sqlite-connection.service.ts
View File

@@ -7,19 +7,15 @@ import { getStoredCurrentUserId } from '../../../core/storage/current-user-stora
/** Shared native SQLite connection used by mobile persistence and DatabaseService. */
@Injectable({ providedIn: 'root' })
export class MobileSqliteConnectionService {
private store: MobileSqliteStore | null = null;
private activeDatabaseName: string | null = null;
private initializationPromise: Promise<MobileSqliteStore | null> | null = null;
private initializationFailed = false;
get isAvailable(): boolean {
return this.store?.isAvailable === true;
}
private store: MobileSqliteStore | null = null;
private activeDatabaseName: string | null = null;
private initializationPromise: Promise<MobileSqliteStore | null> | null = null;
private initializationFailed = false;
async initialize(): Promise<MobileSqliteStore | null> {
if (this.initializationFailed) {
return null;
@@ -70,5 +66,4 @@ export class MobileSqliteConnectionService {
return this.store;
}
}

10
toju-app/src/app/infrastructure/persistence/app-resume.storage.ts
View File

@@ -59,7 +59,7 @@ function scheduleStorageRemove(key: string): void {
scheduleStorageFlush();
}
function readPendingOrStored(key: string): string | null {
function readMaybePending(key: string): string | null {
if (pendingWrites.has(key)) {
return pendingWrites.get(key) ?? null;
}
@@ -73,8 +73,8 @@ function readPendingOrStored(key: string): string | null {
export function loadGeneralSettingsFromStorage(): GeneralSettings {
try {
const raw = readPendingOrStored(getUserScopedStorageKey(STORAGE_KEY_GENERAL_SETTINGS))
?? readPendingOrStored(STORAGE_KEY_GENERAL_SETTINGS);
const raw = readMaybePending(getUserScopedStorageKey(STORAGE_KEY_GENERAL_SETTINGS))
?? readMaybePending(STORAGE_KEY_GENERAL_SETTINGS);
if (!raw) {
return { ...DEFAULT_GENERAL_SETTINGS };
@@ -99,8 +99,8 @@ export function saveGeneralSettingsToStorage(patch: Partial<GeneralSettings>): G
export function loadLastViewedChatFromStorage(userId?: string | null): LastViewedChatSnapshot | null {
try {
const raw = readPendingOrStored(getUserScopedStorageKey(STORAGE_KEY_LAST_VIEWED_CHAT, userId))
?? readPendingOrStored(STORAGE_KEY_LAST_VIEWED_CHAT);
const raw = readMaybePending(getUserScopedStorageKey(STORAGE_KEY_LAST_VIEWED_CHAT, userId))
?? readMaybePending(STORAGE_KEY_LAST_VIEWED_CHAT);
if (!raw) {
return null;

52
toju-app/src/app/infrastructure/persistence/database.service.ts
View File

@@ -1,3 +1,4 @@
/* eslint-disable @typescript-eslint/member-ordering, */
import {
inject,
Injectable,
@@ -37,22 +38,16 @@ export interface RoomMessageStats {
*/
@Injectable({ providedIn: 'root' })
export class DatabaseService {
private readonly platform = inject(PlatformService);
private readonly browserDb = inject(BrowserDatabaseService);
private readonly capacitorDb = inject(CapacitorDatabaseService);
private readonly electronDb = inject(ElectronDatabaseService);
private initializationPromise: Promise<void> | null = null;
private validatedUserScope: string | null | undefined;
/** Reactive flag: `true` once {@link initialize} has completed. */
isReady = signal(false);
private readonly platform = inject(PlatformService);
private readonly browserDb = inject(BrowserDatabaseService);
private readonly capacitorDb = inject(CapacitorDatabaseService);
private readonly electronDb = inject(ElectronDatabaseService);
private initializationPromise: Promise<void> | null = null;
private validatedUserScope: string | null | undefined;
/** The active storage backend for the current platform. */
private get backend() {
const backendKind = resolveDatabaseBackend({
@@ -99,6 +94,22 @@ export class DatabaseService {
await this.initializationPromise;
}
private async ensureReady(): Promise<void> {
const userScope = getStoredCurrentUserId();
if (this.isReady() && this.validatedUserScope === userScope) {
return;
}
await this.initialize();
}
private async withReady<T>(operation: () => Promise<T>): Promise<T> {
await this.ensureReady();
return operation();
}
/** Persist a single chat message. */
saveMessage(message: Message) { return this.withReady(() => this.backend.saveMessage(message)); }
@@ -221,21 +232,4 @@ export class DatabaseService {
/** Wipe all persisted data. */
clearAllData() { return this.withReady(() => this.backend.clearAllData()); }
private async ensureReady(): Promise<void> {
const userScope = getStoredCurrentUserId();
if (this.isReady() && this.validatedUserScope === userScope) {
return;
}
await this.initialize();
}
private async withReady<T>(operation: () => Promise<T>): Promise<T> {
await this.ensureReady();
return operation();
}
}

23
toju-app/src/app/infrastructure/realtime/account-sync/account-sync.effects.ts
View File

@@ -1,3 +1,4 @@
/* eslint-disable @typescript-eslint/member-ordering */
import { Injectable, inject } from '@angular/core';
import {
Actions,
@@ -27,6 +28,13 @@ import type { IncomingSignalingMessage } from '../signaling/signaling-message-ha
@Injectable()
export class AccountSyncEffects {
private readonly actions$ = inject(Actions);
private readonly store = inject(Store);
private readonly webrtc = inject(RealtimeSessionFacade);
private readonly db = inject(DatabaseService);
private readonly friends = inject(FriendService);
private readonly customEmoji = inject(CustomEmojiService);
private readonly attachments = inject(AttachmentFacade);
broadcastSavedRoom$ = createEffect(
() =>
@@ -113,20 +121,6 @@ export class AccountSyncEffects {
{ dispatch: false }
);
private readonly actions$ = inject(Actions);
private readonly store = inject(Store);
private readonly webrtc = inject(RealtimeSessionFacade);
private readonly db = inject(DatabaseService);
private readonly friends = inject(FriendService);
private readonly customEmoji = inject(CustomEmojiService);
private readonly attachments = inject(AttachmentFacade);
private isPeerOnlineMessage(message: IncomingSignalingMessage): message is IncomingSignalingMessage & {
type: 'account_sync_peer_online';
clientInstanceId?: string;
@@ -194,5 +188,4 @@ export class AccountSyncEffects {
customEmoji: emoji
});
}
}

46
toju-app/src/app/infrastructure/realtime/media/media.manager.ts
View File

@@ -1,3 +1,4 @@
/* eslint-disable @typescript-eslint/member-ordering, @typescript-eslint/no-unused-vars, */
/**
* Manages local voice and camera media: getUserMedia, mute, deafen,
* attaching/detaching tracks to peer connections, bitrate tuning,
@@ -24,6 +25,7 @@ import {
VOLUME_MIN,
VOLUME_MAX,
VOICE_HEARTBEAT_INTERVAL_MS,
DEFAULT_DISPLAY_NAME,
P2P_TYPE_CAMERA_STATE,
P2P_TYPE_VOICE_STATE
} from '../realtime.constants';
@@ -46,10 +48,6 @@ export interface MediaManagerCallbacks {
}
export class MediaManager {
/** Emitted when voice is successfully connected. */
readonly voiceConnected$ = new Subject<void>();
/** The stream sent to peers (may be raw or denoised). */
private localMediaStream: MediaStream | null = null;
@@ -69,21 +67,19 @@ export class MediaManager {
// -- Input gain pipeline (mic volume) --
/** The stream BEFORE gain is applied (for identity checks). */
private preGainStream: MediaStream | null = null;
private inputGainCtx: AudioContext | null = null;
private inputGainSourceNode: MediaStreamAudioSourceNode | null = null;
private inputGainNode: GainNode | null = null;
private inputGainDest: MediaStreamAudioDestinationNode | null = null;
/** Normalised 0-1 input gain (1 = 100%). */
private inputGainVolume = 1.0;
/** Voice-presence heartbeat timer. */
private voicePresenceTimer: ReturnType<typeof setInterval> | null = null;
/** Emitted when voice is successfully connected. */
readonly voiceConnected$ = new Subject<void>();
/** RNNoise noise-reduction processor. */
private readonly noiseReduction: NoiseReductionManager;
@@ -96,19 +92,14 @@ export class MediaManager {
// State tracked locally (the service exposes these via signals)
private isVoiceActive = false;
private isMicMuted = false;
private isSelfDeafened = false;
private isCameraActive = false;
/** Current voice channel room ID (set when joining voice). */
private currentVoiceRoomId: string | undefined;
/** Current voice channel server ID (set when joining voice). */
private currentVoiceServerId: string | undefined;
private allowedVoicePeerIds = new Set<string>();
constructor(
@@ -138,52 +129,42 @@ export class MediaManager {
getLocalStream(): MediaStream | null {
return this.localMediaStream;
}
/** Returns the raw microphone stream before processing, if available. */
getRawMicStream(): MediaStream | null {
return this.rawMicStream;
}
/** Returns the current local camera stream, or `null` if the camera is disabled. */
getLocalCameraStream(): MediaStream | null {
return this.localCameraStream;
}
/** Whether voice is currently active (mic captured). */
getIsVoiceActive(): boolean {
return this.isVoiceActive;
}
/** Whether the local microphone is muted. */
getIsMicMuted(): boolean {
return this.isMicMuted;
}
/** Whether the user has self-deafened. */
getIsSelfDeafened(): boolean {
return this.isSelfDeafened;
}
/** Whether the local camera is currently active. */
getIsCameraActive(): boolean {
return this.isCameraActive;
}
/** Current remote audio output volume (normalised 0-1). */
getRemoteAudioVolume(): number {
return this.remoteAudioVolume;
}
/** The voice channel room ID, if currently in voice. */
getCurrentVoiceRoomId(): string | undefined {
return this.currentVoiceRoomId;
}
/** The voice channel server ID, if currently in voice. */
getCurrentVoiceServerId(): string | undefined {
return this.currentVoiceServerId;
}
/** Whether the user wants noise reduction (may or may not be running yet). */
getIsNoiseReductionEnabled(): boolean {
return this._noiseReductionDesired;
@@ -618,15 +599,6 @@ export class MediaManager {
}
}
/** Clean up all resources. */
destroy(): void {
this.teardownInputGain();
this.disableVoice();
this.stopVoiceHeartbeat();
this.noiseReduction.destroy();
this.voiceConnected$.complete();
}
/** Bind any active local mic/camera tracks to the current peer set. */
private bindLocalTracksToAllPeers(): void {
this.syncVoiceRouting();
@@ -984,4 +956,12 @@ export class MediaManager {
this.localCameraStream = null;
}
/** Clean up all resources. */
destroy(): void {
this.teardownInputGain();
this.disableVoice();
this.stopVoiceHeartbeat();
this.noiseReduction.destroy();
this.voiceConnected$.complete();
}
}

12
toju-app/src/app/infrastructure/realtime/media/noise-reduction.manager.ts
View File

@@ -27,12 +27,6 @@ const RNNOISE_SAMPLE_RATE = 48_000;
const WORKLET_MODULE_PATH = 'rnnoise-worklet.js';
export class NoiseReductionManager {
/** Whether noise reduction is currently active. */
get isEnabled(): boolean {
return this._isEnabled;
}
/** The AudioContext used for the noise-reduction graph. */
private audioContext: AudioContext | null = null;
@@ -53,6 +47,11 @@ export class NoiseReductionManager {
constructor(private readonly logger: WebRTCLogger) {}
/** Whether noise reduction is currently active. */
get isEnabled(): boolean {
return this._isEnabled;
}
/**
* Enable noise reduction on a raw microphone stream.
*
@@ -202,5 +201,4 @@ export class NoiseReductionManager {
this.audioContext = null;
this.workletLoaded = false;
}
}

2
toju-app/src/app/infrastructure/realtime/media/screen-share.manager.ts
View File

@@ -1,4 +1,4 @@
/* eslint-disable, @typescript-eslint/no-non-null-assertion, @typescript-eslint/member-ordering, id-denylist */
/**
* Manages screen sharing: getDisplayMedia / Electron desktop capturer,
* system-audio capture, and attaching screen tracks to peers.

121
toju-app/src/app/infrastructure/realtime/peer-connection-manager/peer-connection.manager.ts
View File

@@ -1,3 +1,4 @@
/* eslint-disable @typescript-eslint/member-ordering */
import { ChatEvent } from '../../../shared-kernel';
import { DATA_CHANNEL_LABEL } from '../realtime.constants';
import { recordDebugNetworkDownloadRates } from '../logging/debug-network-metrics';
@@ -62,6 +63,10 @@ interface PeerInboundByteSnapshot {
* offer/answer negotiation, ICE candidates, and P2P reconnection.
*/
export class PeerConnectionManager {
private readonly state = createPeerConnectionManagerState();
private readonly lastInboundByteSnapshots = new Map<string, PeerInboundByteSnapshot>();
private statsPollTimer: ReturnType<typeof setInterval> | null = null;
private transportStatsPollInFlight = false;
/** Active peer connections keyed by remote peer ID. */
readonly activePeerConnections = this.state.activePeerConnections;
@@ -85,76 +90,13 @@ export class PeerConnectionManager {
readonly peerLatencyChanged$ = this.state.peerLatencyChanged$;
readonly peerConnected$ = this.state.peerConnected$;
readonly peerDisconnected$ = this.state.peerDisconnected$;
readonly remoteStream$ = this.state.remoteStream$;
readonly messageReceived$ = this.state.messageReceived$;
/** Emitted whenever the connected peer list changes. */
readonly connectedPeersChanged$ = this.state.connectedPeersChanged$;
private readonly state = createPeerConnectionManagerState();
private readonly lastInboundByteSnapshots = new Map<string, PeerInboundByteSnapshot>();
private statsPollTimer: ReturnType<typeof setInterval> | null = null;
private transportStatsPollInFlight = false;
private get context(): PeerConnectionManagerContext {
return {
logger: this.logger,
callbacks: this.callbacks,
state: this.state
};
}
private get connectionHandlers(): ConnectionLifecycleHandlers {
return {
clearPeerDisconnectGraceTimer: (peerId: string) => this.clearPeerDisconnectGraceTimer(peerId),
addToConnectedPeers: (peerId: string) => this.addToConnectedPeers(peerId),
clearPeerReconnectTimer: (peerId: string) => this.clearPeerReconnectTimer(peerId),
requestVoiceStateFromPeer: (peerId: string) => this.requestVoiceStateFromPeer(peerId),
schedulePeerDisconnectRecovery: (peerId: string) =>
this.schedulePeerDisconnectRecovery(peerId),
trackDisconnectedPeer: (peerId: string) => this.trackDisconnectedPeer(peerId),
removePeer: (peerId: string, options?: RemovePeerOptions) => this.removePeer(peerId, options),
schedulePeerReconnect: (peerId: string) => this.schedulePeerReconnect(peerId),
handleRemoteTrack: (event: RTCTrackEvent, peerId: string) =>
this.handleRemoteTrack(event, peerId),
setupDataChannel: (channel: RTCDataChannel, peerId: string) =>
this.setupDataChannel(channel, peerId)
};
}
private get negotiationHandlers(): NegotiationHandlers {
return {
createPeerConnection: (remotePeerId: string, isInitiator: boolean) =>
this.createPeerConnection(remotePeerId, isInitiator)
};
}
private get recoveryHandlers(): RecoveryHandlers {
return {
removePeer: (peerId: string, options?: RemovePeerOptions) => this.removePeer(peerId, options),
replaceDataChannel: (peerId: string, expectedChannel: RTCDataChannel) =>
this.replaceDataChannel(peerId, expectedChannel),
createPeerConnection: (peerId: string, isInitiator: boolean) =>
this.createPeerConnection(peerId, isInitiator),
createAndSendOffer: (peerId: string) => this.createAndSendOffer(peerId)
};
}
private get dataChannelHandlers(): DataChannelLifecycleHandlers {
return {
clearDataChannelRecoveryTimer: (peerId: string) => this.clearDataChannelRecoveryTimer(peerId),
scheduleDataChannelRecovery: (peerId: string, channel: RTCDataChannel, reason: string) =>
this.scheduleDataChannelRecovery(peerId, channel, reason)
};
}
constructor(
private readonly logger: WebRTCLogger,
private callbacks: PeerConnectionCallbacks
@@ -311,10 +253,62 @@ export class PeerConnectionManager {
this.peerLatencyChanged$.complete();
}
private get context(): PeerConnectionManagerContext {
return {
logger: this.logger,
callbacks: this.callbacks,
state: this.state
};
}
private get connectionHandlers(): ConnectionLifecycleHandlers {
return {
clearPeerDisconnectGraceTimer: (peerId: string) => this.clearPeerDisconnectGraceTimer(peerId),
addToConnectedPeers: (peerId: string) => this.addToConnectedPeers(peerId),
clearPeerReconnectTimer: (peerId: string) => this.clearPeerReconnectTimer(peerId),
requestVoiceStateFromPeer: (peerId: string) => this.requestVoiceStateFromPeer(peerId),
schedulePeerDisconnectRecovery: (peerId: string) =>
this.schedulePeerDisconnectRecovery(peerId),
trackDisconnectedPeer: (peerId: string) => this.trackDisconnectedPeer(peerId),
removePeer: (peerId: string, options?: RemovePeerOptions) => this.removePeer(peerId, options),
schedulePeerReconnect: (peerId: string) => this.schedulePeerReconnect(peerId),
handleRemoteTrack: (event: RTCTrackEvent, peerId: string) =>
this.handleRemoteTrack(event, peerId),
setupDataChannel: (channel: RTCDataChannel, peerId: string) =>
this.setupDataChannel(channel, peerId)
};
}
private get negotiationHandlers(): NegotiationHandlers {
return {
createPeerConnection: (remotePeerId: string, isInitiator: boolean) =>
this.createPeerConnection(remotePeerId, isInitiator)
};
}
private get recoveryHandlers(): RecoveryHandlers {
return {
removePeer: (peerId: string, options?: RemovePeerOptions) => this.removePeer(peerId, options),
replaceDataChannel: (peerId: string, expectedChannel: RTCDataChannel) =>
this.replaceDataChannel(peerId, expectedChannel),
createPeerConnection: (peerId: string, isInitiator: boolean) =>
this.createPeerConnection(peerId, isInitiator),
createAndSendOffer: (peerId: string) => this.createAndSendOffer(peerId)
};
}
private setupDataChannel(channel: RTCDataChannel, remotePeerId: string): void {
setupDataChannel(this.context, channel, remotePeerId, this.dataChannelHandlers);
}
private get dataChannelHandlers(): DataChannelLifecycleHandlers {
return {
clearDataChannelRecoveryTimer: (peerId: string) => this.clearDataChannelRecoveryTimer(peerId),
scheduleDataChannelRecovery: (peerId: string, channel: RTCDataChannel, reason: string) =>
this.scheduleDataChannelRecovery(peerId, channel, reason)
};
}
private handleRemoteTrack(event: RTCTrackEvent, remotePeerId: string): void {
handleRemoteTrack(this.context, event, remotePeerId);
}
@@ -513,5 +507,4 @@ export class PeerConnectionManager {
private roundMetric(value: number): number {
return Math.round(value * 1000) / 1000;
}
}

290
toju-app/src/app/infrastructure/realtime/realtime-session.service.ts
View File

@@ -11,6 +11,7 @@
* This file wires them together and exposes a public API that is
* identical to the old monolithic service so consumers don't change.
*/
/* eslint-disable @typescript-eslint/member-ordering, @typescript-eslint/no-non-null-assertion */
import {
Injectable,
inject,
@@ -59,115 +60,70 @@ import { selectCurrentUser } from '../../store/users/users.selectors';
providedIn: 'root'
})
export class WebRTCService implements OnDestroy {
private readonly timeSync = inject(TimeSyncService);
private readonly debugging = inject(DebuggingService);
private readonly screenShareSourcePicker = inject(ScreenShareSourcePickerService);
private readonly iceServerSettings = inject(IceServerSettingsService);
private readonly signalServerAuth = inject(SignalServerAuthService);
private readonly store = inject(Store);
private readonly clientInstance = inject(ClientInstanceService);
private currentHomeUser: { id: string; homeSignalServerUrl?: string; displayName: string } | null = null;
private readonly logger = new WebRTCLogger(() => this.debugging.enabled());
private readonly state = new WebRtcStateController();
readonly peerId = this.state.peerId;
readonly isConnected = this.state.isConnected;
readonly hasEverConnected = this.state.hasEverConnected;
readonly isVoiceConnected = this.state.isVoiceConnected;
readonly connectedPeers = this.state.connectedPeers;
readonly isMuted = this.state.isMuted;
readonly isDeafened = this.state.isDeafened;
readonly isCameraEnabled = this.state.isCameraEnabled;
readonly isScreenSharing = this.state.isScreenSharing;
readonly isNoiseReductionEnabled = this.state.isNoiseReductionEnabled;
readonly screenStream = this.state.screenStream;
readonly isScreenShareRemotePlaybackSuppressed = this.state.isScreenShareRemotePlaybackSuppressed;
readonly forceDefaultRemotePlaybackOutput = this.state.forceDefaultRemotePlaybackOutput;
readonly hasConnectionError = this.state.hasConnectionError;
readonly connectionErrorMessage = this.state.connectionErrorMessage;
readonly shouldShowConnectionError = this.state.shouldShowConnectionError;
readonly peerLatencies = this.state.peerLatencies;
private readonly signalingMessage$ = new Subject<IncomingSignalingMessage>();
readonly onSignalingMessage = this.signalingMessage$.asObservable();
private readonly accountSyncRelay$ = new Subject<ChatEvent>();
private readonly signalingReconnectedSubject$ = new Subject<string>();
readonly signalingReconnected$ = this.signalingReconnectedSubject$.asObservable();
// Delegates to managers
get onMessageReceived(): Observable<ChatEvent> {
return merge(this.peerMediaFacade.onMessageReceived, this.accountSyncRelay$);
}
get onPeerConnected(): Observable<string> {
return this.peerMediaFacade.onPeerConnected;
}
get onPeerDisconnected(): Observable<string> {
return this.peerMediaFacade.onPeerDisconnected;
}
get onRemoteStream(): Observable<{ peerId: string; stream: MediaStream }> {
return this.peerMediaFacade.onRemoteStream;
}
get onVoiceConnected(): Observable<void> {
return this.peerMediaFacade.onVoiceConnected;
}
/** The server ID currently being viewed / active, or `null`. */
get currentServerId(): string | null {
return this.state.currentServerId;
}
private readonly timeSync = inject(TimeSyncService);
private readonly debugging = inject(DebuggingService);
private readonly screenShareSourcePicker = inject(ScreenShareSourcePickerService);
private readonly iceServerSettings = inject(IceServerSettingsService);
private readonly signalServerAuth = inject(SignalServerAuthService);
private readonly store = inject(Store);
private readonly clientInstance = inject(ClientInstanceService);
private currentHomeUser: { id: string; homeSignalServerUrl?: string; displayName: string } | null = null;
private readonly logger = new WebRTCLogger(() => this.debugging.enabled());
private readonly state = new WebRtcStateController();
private readonly signalingMessage$ = new Subject<IncomingSignalingMessage>();
private readonly accountSyncRelay$ = new Subject<ChatEvent>();
private readonly signalingReconnectedSubject$ = new Subject<string>();
private readonly peerManager: PeerConnectionManager;
private readonly mediaManager: MediaManager;
private readonly screenShareManager: ScreenShareManager;
private readonly peerMediaFacade: PeerMediaFacade;
private readonly voiceSessionController: VoiceSessionController;
private readonly signalingCoordinator: ServerSignalingCoordinator<IncomingSignalingMessage>;
private readonly signalingTransportHandler: SignalingTransportHandler<IncomingSignalingMessage>;
private readonly signalingMessageHandler: IncomingSignalingMessageHandler;
private readonly serverMembershipSignalingHandler: ServerMembershipSignalingHandler<IncomingSignalingMessage>;
private readonly remoteScreenShareRequestController: RemoteScreenShareRequestController;
constructor() {
@@ -182,13 +138,11 @@ export class WebRTCService implements OnDestroy {
});
// Create managers with null callbacks first to break circular initialization
const pendingPeerMessageHandler = () => undefined;
this.peerManager = new PeerConnectionManager(this.logger, null!);
this.peerManager = new PeerConnectionManager(this.logger, pendingPeerMessageHandler);
this.mediaManager = new MediaManager(this.logger, null!);
this.mediaManager = new MediaManager(this.logger, pendingPeerMessageHandler);
this.screenShareManager = new ScreenShareManager(this.logger, pendingPeerMessageHandler);
this.screenShareManager = new ScreenShareManager(this.logger, null!);
this.peerMediaFacade = new PeerMediaFacade({
peerManager: this.peerManager,
@@ -306,6 +260,79 @@ export class WebRTCService implements OnDestroy {
this.wireManagerEvents();
}
private wireManagerEvents(): void {
// Internal control-plane messages for on-demand screen-share delivery.
this.peerManager.messageReceived$.subscribe((event) =>
this.remoteScreenShareRequestController.handlePeerControlMessage(event)
);
// Peer manager -> connected peers signal
this.peerManager.connectedPeersChanged$.subscribe((peers: string[]) =>
this.state.setConnectedPeers(peers)
);
// If we are already sharing when a new peer connection finishes, push the
// current screen-share tracks to that peer and renegotiate.
this.peerManager.peerConnected$.subscribe((peerId) => {
if (this.peerMediaFacade.isScreenShareActive()) {
this.peerMediaFacade.syncScreenShareToPeer(peerId);
}
this.mediaManager.refreshVoiceRouting();
this.remoteScreenShareRequestController.handlePeerConnected(peerId);
});
this.peerManager.peerDisconnected$.subscribe((peerId) => {
this.remoteScreenShareRequestController.handlePeerDisconnected(peerId);
});
// Media manager -> voice connected signal
this.mediaManager.voiceConnected$.subscribe(() => {
this.voiceSessionController.handleVoiceConnected();
});
// Peer manager -> latency updates
this.peerManager.peerLatencyChanged$.subscribe(() =>
this.state.syncPeerLatencies(this.peerManager.peerLatencies)
);
}
private handleSignalingConnectionStatus(signalUrl: string, connected: boolean, errorMessage?: string): void {
this.state.updateSignalingConnectionStatus(
connected ? true : this.signalingCoordinator.isAnySignalingConnected(),
connected,
errorMessage
);
if (connected) {
this.signalingReconnectedSubject$.next(signalUrl);
}
}
private handleSignalingMessage(message: IncomingSignalingMessage, signalUrl: string): void {
if (message.type === 'auth_required' || message.type === 'auth_error') {
this.store.dispatch(UsersActions.signalServerAuthFailed({ signalUrl }));
return;
}
if (message.type === 'account_sync') {
const accountMessage = message as AccountSyncSignalingMessage;
if (shouldApplyAccountSyncPayload(
accountMessage.clientInstanceId,
this.clientInstance.getClientInstanceId()
)) {
this.accountSyncRelay$.next(unwrapAccountSyncPayload(accountMessage));
}
return;
}
this.signalingMessage$.next(message);
this.signalingMessageHandler.handleMessage(message, signalUrl);
}
// PUBLIC API - matches the old monolithic service's interface
/**
@@ -365,6 +392,11 @@ export class WebRTCService implements OnDestroy {
this.state.setCurrentServer(serverId);
}
/** The server ID currently being viewed / active, or `null`. */
get currentServerId(): string | null {
return this.state.currentServerId;
}
/** The last signaling URL used by the client, if any. */
getCurrentSignalingUrl(): string | null {
return this.signalingTransportHandler.getCurrentSignalingUrl(this.state.currentServerId);
@@ -725,107 +757,6 @@ export class WebRTCService implements OnDestroy {
this.peerMediaFacade.stopScreenShare();
}
requestVoiceClientTakeover(): void {
this.signalingTransportHandler.sendRawMessage({
type: 'voice_client_takeover',
clientInstanceId: this.clientInstance.getClientInstanceId()
});
}
getClientInstanceId(): string {
return this.clientInstance.getClientInstanceId();
}
/** Disconnect from the signaling server and clean up all state. */
disconnect(): void {
this.leaveRoom();
this.destroyAllSignalingManagers();
this.state.resetConnectionState();
}
/** Alias for {@link disconnect}. */
disconnectAll(): void {
this.disconnect();
}
ngOnDestroy(): void {
this.disconnect();
this.peerMediaFacade.destroy();
}
private wireManagerEvents(): void {
// Internal control-plane messages for on-demand screen-share delivery.
this.peerManager.messageReceived$.subscribe((event) =>
this.remoteScreenShareRequestController.handlePeerControlMessage(event)
);
// Peer manager -> connected peers signal
this.peerManager.connectedPeersChanged$.subscribe((peers: string[]) =>
this.state.setConnectedPeers(peers)
);
// If we are already sharing when a new peer connection finishes, push the
// current screen-share tracks to that peer and renegotiate.
this.peerManager.peerConnected$.subscribe((peerId) => {
if (this.peerMediaFacade.isScreenShareActive()) {
this.peerMediaFacade.syncScreenShareToPeer(peerId);
}
this.mediaManager.refreshVoiceRouting();
this.remoteScreenShareRequestController.handlePeerConnected(peerId);
});
this.peerManager.peerDisconnected$.subscribe((peerId) => {
this.remoteScreenShareRequestController.handlePeerDisconnected(peerId);
});
// Media manager -> voice connected signal
this.mediaManager.voiceConnected$.subscribe(() => {
this.voiceSessionController.handleVoiceConnected();
});
// Peer manager -> latency updates
this.peerManager.peerLatencyChanged$.subscribe(() =>
this.state.syncPeerLatencies(this.peerManager.peerLatencies)
);
}
private handleSignalingConnectionStatus(signalUrl: string, connected: boolean, errorMessage?: string): void {
this.state.updateSignalingConnectionStatus(
connected ? true : this.signalingCoordinator.isAnySignalingConnected(),
connected,
errorMessage
);
if (connected) {
this.signalingReconnectedSubject$.next(signalUrl);
}
}
private handleSignalingMessage(message: IncomingSignalingMessage, signalUrl: string): void {
if (message.type === 'auth_required' || message.type === 'auth_error') {
this.store.dispatch(UsersActions.signalServerAuthFailed({ signalUrl }));
return;
}
if (message.type === 'account_sync') {
const accountMessage = message as AccountSyncSignalingMessage;
if (shouldApplyAccountSyncPayload(
accountMessage.clientInstanceId,
this.clientInstance.getClientInstanceId()
)) {
this.accountSyncRelay$.next(unwrapAccountSyncPayload(accountMessage));
}
return;
}
this.signalingMessage$.next(message);
this.signalingMessageHandler.handleMessage(message, signalUrl);
}
private relayBroadcastEvent(event: ChatEvent): void {
const clientInstanceId = this.clientInstance.getClientInstanceId();
@@ -870,6 +801,29 @@ export class WebRTCService implements OnDestroy {
this.relayAccountSync(event);
}
requestVoiceClientTakeover(): void {
this.signalingTransportHandler.sendRawMessage({
type: 'voice_client_takeover',
clientInstanceId: this.clientInstance.getClientInstanceId()
});
}
getClientInstanceId(): string {
return this.clientInstance.getClientInstanceId();
}
/** Disconnect from the signaling server and clean up all state. */
disconnect(): void {
this.leaveRoom();
this.destroyAllSignalingManagers();
this.state.resetConnectionState();
}
/** Alias for {@link disconnect}. */
disconnectAll(): void {
this.disconnect();
}
private fullCleanup(): void {
this.signalingCoordinator.clearPeerTracking();
this.remoteScreenShareRequestController.clear();
@@ -885,4 +839,8 @@ export class WebRTCService implements OnDestroy {
this.signalingCoordinator.destroy();
}
ngOnDestroy(): void {
this.disconnect();
this.peerMediaFacade.destroy();
}
}

94
toju-app/src/app/infrastructure/realtime/signaling/signaling.manager.ts
View File

@@ -1,3 +1,4 @@
/* eslint-disable @typescript-eslint/member-ordering, @typescript-eslint/no-non-null-assertion,, max-statements-per-line */
/**
* Manages the WebSocket connection to the signaling server,
* including automatic reconnection and heartbeats.
@@ -40,6 +41,19 @@ type ParsedSignalingMessage = Omit<Partial<SignalingMessage>, 'type' | 'payload'
};
export class SignalingManager {
private signalingWebSocket: WebSocket | null = null;
private lastSignalingUrl: string | null = null;
private signalingReconnectAttempts = 0;
private signalingReconnectTimer: ReturnType<typeof setTimeout> | null = null;
private stateHeartbeatTimer: ReturnType<typeof setInterval> | null = null;
private lastKeepaliveSentAt = 0;
private lastKeepaliveAckAt = 0;
private serverSupportsKeepaliveAck = false;
private lastEndpointHealthOk: boolean | null = null;
private lastKnownServerInstanceId: string | null = null;
private lastEndpointHealthProbeAt = 0;
private endpointHealthProbeInFlight = false;
private connectAttemptStartedAt = 0;
/** Fires every heartbeat tick - the main service hooks this to broadcast state. */
readonly heartbeatTick$ = new Subject<void>();
@@ -50,32 +64,6 @@ export class SignalingManager {
/** Fires when connection status changes (true = open, false = closed/error). */
readonly connectionStatus$ = new Subject<{ connected: boolean; errorMessage?: string }>();
private signalingWebSocket: WebSocket | null = null;
private lastSignalingUrl: string | null = null;
private signalingReconnectAttempts = 0;
private signalingReconnectTimer: ReturnType<typeof setTimeout> | null = null;
private stateHeartbeatTimer: ReturnType<typeof setInterval> | null = null;
private lastKeepaliveSentAt = 0;
private lastKeepaliveAckAt = 0;
private serverSupportsKeepaliveAck = false;
private lastEndpointHealthOk: boolean | null = null;
private lastKnownServerInstanceId: string | null = null;
private lastEndpointHealthProbeAt = 0;
private endpointHealthProbeInFlight = false;
private connectAttemptStartedAt = 0;
constructor(
private readonly logger: WebRTCLogger,
private readonly getLastIdentify: () => IdentifyCredentials | null,
@@ -280,19 +268,10 @@ export class SignalingManager {
let settled = false;
const timeout = setTimeout(() => {
if (!settled) {
settled = true;
resolve(false);
}
if (!settled) { settled = true; resolve(false); }
}, timeoutMs);
const signalingUrl = this.lastSignalingUrl;
if (!signalingUrl) {
resolve(false);
return;
}
this.connect(signalingUrl).subscribe({
this.connect(this.lastSignalingUrl!).subscribe({
next: (connected) => {
if (!settled) {
settled = true;
@@ -300,13 +279,7 @@ export class SignalingManager {
resolve(connected);
}
},
error: () => {
if (!settled) {
settled = true;
clearTimeout(timeout);
resolve(false);
}
}
error: () => { if (!settled) { settled = true; clearTimeout(timeout); resolve(false); } }
});
});
}
@@ -393,14 +366,6 @@ export class SignalingManager {
return this.lastSignalingUrl;
}
/** Clean up all resources. */
destroy(): void {
this.close();
this.heartbeatTick$.complete();
this.messageReceived$.complete();
this.connectionStatus$.complete();
}
/** Re-identify and rejoin servers after a reconnect. */
private reIdentifyAndRejoin(): void {
const credentials = this.getLastIdentify();
@@ -473,13 +438,7 @@ export class SignalingManager {
url: this.lastSignalingUrl
});
const reconnectUrl = this.lastSignalingUrl;
if (!reconnectUrl) {
return;
}
this.connect(reconnectUrl).subscribe({
this.connect(this.lastSignalingUrl!).subscribe({
next: (connected) => {
if (connected) {
this.signalingReconnectAttempts = 0;
@@ -712,6 +671,14 @@ export class SignalingManager {
}
}
/** Clean up all resources. */
destroy(): void {
this.close();
this.heartbeatTick$.complete();
this.messageReceived$.complete();
this.connectionStatus$.complete();
}
private sendSerializedPayload(
message: SignalingMessage | Record<string, unknown>,
details: { targetPeerId?: string; type?: string; url?: string | null }
@@ -734,14 +701,8 @@ export class SignalingManager {
throw error;
}
const socket = this.signalingWebSocket;
if (!socket) {
return;
}
try {
socket.send(rawPayload);
this.signalingWebSocket!.send(rawPayload);
this.logger.traffic('signaling', 'outbound', {
...payloadPreview,
bytes: this.measurePayloadBytes(rawPayload),
@@ -879,5 +840,4 @@ export class SignalingManager {
return value as Record<string, unknown>;
}
}

41
toju-app/src/app/infrastructure/realtime/state/webrtc-state-controller.ts
View File

@@ -8,75 +8,39 @@ import type { LocalScreenShareState } from '../media/screen-share.manager';
export class WebRtcStateController {
readonly peerId: Signal<string>;
readonly isConnected: Signal<boolean>;
readonly hasEverConnected: Signal<boolean>;
readonly isVoiceConnected: Signal<boolean>;
readonly connectedPeers: Signal<string[]>;
readonly isMuted: Signal<boolean>;
readonly isDeafened: Signal<boolean>;
readonly isCameraEnabled: Signal<boolean>;
readonly isScreenSharing: Signal<boolean>;
readonly isNoiseReductionEnabled: Signal<boolean>;
readonly screenStream: Signal<MediaStream | null>;
readonly isScreenShareRemotePlaybackSuppressed: Signal<boolean>;
readonly forceDefaultRemotePlaybackOutput: Signal<boolean>;
readonly hasConnectionError: Signal<boolean>;
readonly connectionErrorMessage: Signal<string | null>;
readonly shouldShowConnectionError: Signal<boolean>;
readonly peerLatencies: Signal<ReadonlyMap<string, number>>;
get currentServerId(): string | null {
return this.activeServerId;
}
private activeServerId: string | null = null;
private readonly _localPeerId = signal<string>(uuidv4());
private readonly _isSignalingConnected = signal(false);
private readonly _isVoiceConnected = signal(false);
private readonly _connectedPeers = signal<string[]>([]);
private readonly _isMuted = signal(false);
private readonly _isDeafened = signal(false);
private readonly _isCameraEnabled = signal(false);
private readonly _isScreenSharing = signal(false);
private readonly _isNoiseReductionEnabled = signal(false);
private readonly _screenStreamSignal = signal<MediaStream | null>(null);
private readonly _isScreenShareRemotePlaybackSuppressed = signal(false);
private readonly _forceDefaultRemotePlaybackOutput = signal(false);
private readonly _hasConnectionError = signal(false);
private readonly _connectionErrorMessage = signal<string | null>(null);
private readonly _hasEverConnected = signal(false);
private readonly _peerLatencies = signal<ReadonlyMap<string, number>>(new Map());
constructor() {
@@ -108,6 +72,10 @@ export class WebRtcStateController {
this.peerLatencies = computed(() => this._peerLatencies());
}
get currentServerId(): string | null {
return this.activeServerId;
}
getLocalPeerId(): string {
return this._localPeerId();
}
@@ -205,5 +173,4 @@ export class WebRtcStateController {
this._hasConnectionError.set(!anyConnected);
this._connectionErrorMessage.set(anyConnected ? null : (errorMessage ?? 'network.signaling.disconnected'));
}
}

8
toju-app/src/app/infrastructure/realtime/streams/peer-media-facade.ts
View File

@@ -13,6 +13,9 @@ interface PeerMediaFacadeDependencies {
}
export class PeerMediaFacade {
constructor(
private readonly dependencies: PeerMediaFacadeDependencies
) {}
get onMessageReceived(): Observable<ChatEvent> {
return this.dependencies.peerManager.messageReceived$.asObservable();
@@ -34,10 +37,6 @@ export class PeerMediaFacade {
return this.dependencies.mediaManager.voiceConnected$.asObservable();
}
constructor(
private readonly dependencies: PeerMediaFacadeDependencies
) {}
getActivePeers(): Map<string, PeerData> {
return this.dependencies.peerManager.activePeerConnections;
}
@@ -135,5 +134,4 @@ export class PeerMediaFacade {
this.dependencies.mediaManager.destroy();
this.dependencies.screenShareManager.destroy();
}
}