myxelium/Toju
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects 1 Releases 17 Wiki Activity

fix: Broken voice states and connectivity drops

Browse Source
This commit is contained in:
SocksOnHead
2026-04-11 12:32:22 +02:00
parent 0865c2fe33
commit ef1182d46f
28 changed files with 1244 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
toju-app/src/app/domains/server-directory/README.md
View File

@@ -90,10 +90,15 @@ stateDiagram-v2
The facade exposes `testServer(endpointId)` and `testAllServers()`. Both delegate to `ServerEndpointHealthService.probeEndpoint()`, which:
1. Sends `GET /api/health` with a 5-second timeout
2. On success, checks the response's `serverVersion` against the client version via `ServerEndpointCompatibilityService`
3. If versions are incompatible, the endpoint is marked `incompatible` and deactivated
4. If `/api/health` fails, falls back to `GET /api/servers` as a basic liveness check
5. Updates the endpoint's status, latency, and version info in the state service
2. Reads the response's `serverVersion` and stable `serverInstanceId`
3. Checks the reported version against the client version via `ServerEndpointCompatibilityService`
4. If versions are incompatible, the endpoint is marked `incompatible` and deactivated
5. If `/api/health` fails, falls back to `GET /api/servers` as a basic liveness check
6. Updates the endpoint's status, latency, and version info in the state service
`serverInstanceId` lets the client detect when multiple configured URLs point at the same backend. `ServerEndpointStateService.resolveCanonicalEndpoint()` prefers one canonical endpoint per backend instance so REST calls, WebSocket routing, and room fallback logic do not treat same-instance aliases as different signaling clusters.
Room signaling now waits for that initial health sweep before the first saved-room reconnect attempt. That avoids a cold-start race where alias endpoints could open separate WebSocket managers before `serverInstanceId` had been learned.
```mermaid
sequenceDiagram
@@ -106,7 +111,7 @@ sequenceDiagram
Health->>API: GET /api/health (5s timeout)
alt 200 OK
API-->>Health: { serverVersion }
API-->>Health: { serverVersion, serverInstanceId }
Health->>Compat: evaluateServerVersion(serverVersion, clientVersion)
Compat-->>Health: { isCompatible, serverVersion }
Health-->>Facade: online / incompatible + latency + versions
@@ -132,6 +137,10 @@ The facade's `searchServers(query)` method supports two modes controlled by a `s
The API service normalises every `ServerInfo` response, filling in `sourceId`, `sourceName`, and `sourceUrl` so the UI knows which endpoint each server came from.
That search fan-out is discovery only. Once a room is created or joined, the room keeps an authoritative signal-server affinity via its `sourceId` / `sourceUrl`. The join response can repair stale saved metadata, and reconnect logic now retries that authoritative endpoint first before probing any other configured endpoints.
Fallback stays temporary. If the authoritative endpoint is unavailable, the client can probe other active compatible endpoints as a last resort for the current session, but it does not rewrite the room's saved affinity to that fallback endpoint.
## Server-owned room metadata
`ServerInfo` also carries the server-owned `channels` list for each room. Register and update calls persist this channel metadata on the server, and search or hydration responses return the normalised channel list so text and voice channel topology survives reloads, reconnects, and fresh joins.
@@ -147,6 +156,8 @@ Default servers are configured in the environment file. The state service builds
- `restoreDefaultServers()` re-adds any removed defaults and clears the removal tracking
- The primary default URL is used as a fallback when no endpoint is resolved
Saved rooms can also self-heal their endpoint metadata. If a room has missing or stale source information, the client now searches the configured endpoints for that room, restores the correct source mapping, and persists the repair locally.
URL sanitisation strips trailing slashes and `/api` suffixes. Protocol-less URLs get `http` or `https` based on the current page protocol.
## Server administration

92
toju-app/src/app/domains/server-directory/application/server-directory.facade.ts
View File

@@ -21,6 +21,12 @@ import type {
ServerSourceSelector,
UnbanServerMemberRequest
} from '../domain/server-directory.models';
import {
buildRoomSignalSelector,
buildRoomSignalSource,
type RoomSignalSource,
type RoomSignalSourceInput
} from '../domain/room-signal-source';
import { ServerEndpointCompatibilityService } from '../infrastructure/server-endpoint-compatibility.service';
import { ServerEndpointHealthService } from '../infrastructure/server-endpoint-health.service';
import { ServerEndpointStateService } from './server-endpoint-state.service';
@@ -38,6 +44,7 @@ export class ServerDirectoryFacade {
private readonly endpointCompatibility = inject(ServerEndpointCompatibilityService);
private readonly endpointHealth = inject(ServerEndpointHealthService);
private readonly api = inject(ServerDirectoryApiService);
private readonly initialServerHealthCheck: Promise<void>;
private shouldSearchAllServers = true;
constructor() {
@@ -47,7 +54,11 @@ export class ServerDirectoryFacade {
this.activeServer = this.endpointState.activeServer;
this.loadConnectionSettings();
void this.testAllServers();
this.initialServerHealthCheck = this.testAllServers().catch(() => undefined);
}
async awaitInitialServerHealthCheck(): Promise<void> {
await this.initialServerHealthCheck;
}
addServer(server: { name: string; url: string }): ServerEndpoint {
@@ -110,6 +121,81 @@ export class ServerDirectoryFacade {
return !!refreshedEndpoint && refreshedEndpoint.status !== 'incompatible';
}
resolveRoomEndpoint(
source?: RoomSignalSourceInput,
options?: { ensureEndpoint?: boolean; setActive?: boolean }
): ServerEndpoint | null {
const normalizedSource = buildRoomSignalSource(source);
if (normalizedSource.sourceId) {
const endpointById = this.endpointState.resolveCanonicalEndpoint(
this.servers().find((endpoint) => endpoint.id === normalizedSource.sourceId) ?? null
);
if (endpointById) {
return endpointById;
}
}
if (!normalizedSource.sourceUrl) {
return null;
}
const endpointByUrl = this.endpointState.resolveCanonicalEndpoint(
this.findServerByUrl(normalizedSource.sourceUrl) ?? null
);
if (endpointByUrl) {
return endpointByUrl;
}
if (!options?.ensureEndpoint) {
return null;
}
return this.ensureServerEndpoint({
name: normalizedSource.sourceName ?? 'Signal Server',
url: normalizedSource.sourceUrl
}, {
setActive: options.setActive
});
}
normaliseRoomSignalSource(
source?: RoomSignalSourceInput,
options?: { ensureEndpoint?: boolean; setActive?: boolean }
): RoomSignalSource {
const endpoint = this.resolveRoomEndpoint(source, options);
return buildRoomSignalSource(source, endpoint);
}
buildRoomSignalSelector(
source?: RoomSignalSourceInput,
options?: { ensureEndpoint?: boolean; setActive?: boolean }
): ServerSourceSelector | undefined {
return buildRoomSignalSelector(this.normaliseRoomSignalSource(source, options));
}
getFallbackRoomEndpoints(source?: RoomSignalSourceInput): ServerEndpoint[] {
const primaryEndpoint = this.resolveRoomEndpoint(source);
const primarySource = this.normaliseRoomSignalSource(source);
const primaryUrl = primarySource.sourceUrl ? this.endpointState.sanitiseUrl(primarySource.sourceUrl) : null;
const primaryInstanceId = primaryEndpoint?.instanceId ?? null;
return this.activeServers().filter((endpoint) => {
if (endpoint.status === 'incompatible') {
return false;
}
if (primaryInstanceId && endpoint.instanceId === primaryInstanceId) {
return false;
}
return !primaryUrl || this.endpointState.sanitiseUrl(endpoint.url) !== primaryUrl;
});
}
setSearchAllServers(enabled: boolean): void {
this.shouldSearchAllServers = enabled;
}
@@ -159,6 +245,10 @@ export class ServerDirectoryFacade {
return this.api.getServer(serverId, selector);
}
findServerAcrossActiveEndpoints(serverId: string, source?: RoomSignalSourceInput): Observable<ServerInfo | null> {
return this.api.findServerAcrossActiveEndpoints(serverId, source);
}
registerServer(
server: Omit<ServerInfo, 'createdAt'> & { id?: string; password?: string | null },
selector?: ServerSourceSelector

61
toju-app/src/app/domains/server-directory/application/server-endpoint-state.service.ts
View File

@@ -121,6 +121,20 @@ export class ServerEndpointStateService {
return this._servers().find((endpoint) => this.sanitiseUrl(endpoint.url) === sanitisedUrl);
}
resolveCanonicalEndpoint(endpoint: ServerEndpoint | null | undefined): ServerEndpoint | null {
if (!endpoint?.instanceId) {
return endpoint ?? null;
}
const equivalentEndpoints = this._servers().filter((candidate) => candidate.instanceId === endpoint.instanceId);
if (equivalentEndpoints.length <= 1) {
return endpoint;
}
return [...equivalentEndpoints].sort((left, right) => this.compareEndpointPreference(left, right))[0] ?? endpoint;
}
removeServer(endpointId: string): void {
const endpoints = this._servers();
const target = endpoints.find((endpoint) => endpoint.id === endpointId);
@@ -208,6 +222,7 @@ export class ServerEndpointStateService {
return {
...endpoint,
instanceId: versions?.serverInstanceId ?? endpoint.instanceId,
status,
latency,
isActive: status === 'incompatible' ? false : endpoint.isActive,
@@ -312,4 +327,50 @@ export class ServerEndpointStateService {
private saveEndpoints(): void {
this.storage.saveEndpoints(this._servers());
}
private compareEndpointPreference(left: ServerEndpoint, right: ServerEndpoint): number {
const scoreDifference = this.endpointPreferenceScore(right) - this.endpointPreferenceScore(left);
if (scoreDifference !== 0) {
return scoreDifference;
}
return left.url.localeCompare(right.url);
}
private endpointPreferenceScore(endpoint: ServerEndpoint): number {
let score = 0;
if (endpoint.isDefault) {
score += 4;
}
if (this.usesSecureProtocol(endpoint.url)) {
score += 2;
}
if (!this.isIpHost(endpoint.url)) {
score += 1;
}
return score;
}
private usesSecureProtocol(url: string): boolean {
try {
return new URL(url).protocol === 'https:';
} catch {
return false;
}
}
private isIpHost(url: string): boolean {
try {
const hostname = new URL(url).hostname;
return /^\d{1,3}(?:\.\d{1,3}){3}$/.test(hostname) || hostname.includes(':');
} catch {
return false;
}
}
}

46
toju-app/src/app/domains/server-directory/domain/room-signal-source.spec.ts Normal file
View File

@@ -0,0 +1,46 @@
import {
areRoomSignalSourcesEqual,
buildRoomSignalSelector,
buildRoomSignalSource,
getSourceUrlFromSignalingUrl
} from './room-signal-source';
describe('room-signal-source helpers', () => {
it('converts signaling urls back to normalized source urls', () => {
expect(getSourceUrlFromSignalingUrl('wss://signal.toju.app')).toBe('https://signal.toju.app');
expect(getSourceUrlFromSignalingUrl('ws://46.59.68.77:3001')).toBe('http://46.59.68.77:3001');
});
it('prefers the resolved endpoint when normalizing a room source', () => {
expect(buildRoomSignalSource({
sourceId: 'stale-id',
sourceName: 'Stale Source',
sourceUrl: 'https://old.example.com',
signalingUrl: 'wss://signal.toju.app'
}, {
id: 'primary-id',
name: 'Primary Signal',
url: 'https://signal.toju.app/'
})).toEqual({
sourceId: 'primary-id',
sourceName: 'Primary Signal',
sourceUrl: 'https://signal.toju.app'
});
});
it('builds selectors from signaling urls when no source url is persisted yet', () => {
expect(buildRoomSignalSelector({
signalingUrl: 'wss://signal-sweden.toju.app',
fallbackName: 'Toju Signal Sweden'
})).toEqual({
sourceUrl: 'https://signal-sweden.toju.app'
});
});
it('treats equivalent persisted and signaling-derived sources as equal', () => {
expect(areRoomSignalSourcesEqual(
{ sourceUrl: 'https://signal.toju.app/' },
{ signalingUrl: 'wss://signal.toju.app' }
)).toBeTrue();
});
});

95
toju-app/src/app/domains/server-directory/domain/room-signal-source.ts Normal file
View File

@@ -0,0 +1,95 @@
import type { ServerEndpoint, ServerSourceSelector } from './server-directory.models';
import { normaliseConfiguredServerUrl, sanitiseServerBaseUrl } from './server-endpoint-defaults';
export interface RoomSignalSource {
sourceId?: string;
sourceName?: string;
sourceUrl?: string;
}
export interface RoomSignalSourceInput extends RoomSignalSource {
fallbackName?: string;
signalingUrl?: string;
}
const DEFAULT_SIGNAL_SOURCE_NAME = 'Signal Server';
export function getSourceUrlFromSignalingUrl(signalingUrl?: string): string | undefined {
const normalizedUrl = normalizeString(signalingUrl);
if (!normalizedUrl) {
return undefined;
}
const resolvedUrl = normaliseConfiguredServerUrl(normalizedUrl, 'https');
return resolvedUrl || undefined;
}
export function buildRoomSignalSource(
source?: RoomSignalSourceInput | null,
endpoint?: Pick<ServerEndpoint, 'id' | 'name' | 'url'> | null
): RoomSignalSource {
const sourceId = normalizeString(endpoint?.id) ?? normalizeString(source?.sourceId);
const sourceUrl = endpoint
? sanitiseServerBaseUrl(endpoint.url)
: (normalizeUrl(source?.sourceUrl) ?? getSourceUrlFromSignalingUrl(source?.signalingUrl));
const sourceName = normalizeString(endpoint?.name)
?? normalizeString(source?.sourceName)
?? normalizeString(source?.fallbackName)
?? (sourceUrl ? DEFAULT_SIGNAL_SOURCE_NAME : undefined);
return {
sourceId,
sourceName,
sourceUrl
};
}
export function buildRoomSignalSelector(
source?: RoomSignalSourceInput | null
): ServerSourceSelector | undefined {
const normalizedSource = buildRoomSignalSource(source);
if (normalizedSource.sourceId) {
return { sourceId: normalizedSource.sourceId };
}
if (normalizedSource.sourceUrl) {
return { sourceUrl: normalizedSource.sourceUrl };
}
return undefined;
}
export function hasRoomSignalSource(source?: RoomSignalSourceInput | null): boolean {
return !!buildRoomSignalSelector(source);
}
export function areRoomSignalSourcesEqual(
left?: RoomSignalSourceInput | null,
right?: RoomSignalSourceInput | null
): boolean {
const normalizedLeft = buildRoomSignalSource(left);
const normalizedRight = buildRoomSignalSource(right);
return normalizedLeft.sourceId === normalizedRight.sourceId
&& normalizedLeft.sourceName === normalizedRight.sourceName
&& normalizedLeft.sourceUrl === normalizedRight.sourceUrl;
}
function normalizeString(value: string | undefined | null): string | undefined {
if (typeof value !== 'string') {
return undefined;
}
const trimmed = value.trim();
return trimmed || undefined;
}
function normalizeUrl(value: string | undefined | null): string | undefined {
const normalizedValue = normalizeString(value);
return normalizedValue ? sanitiseServerBaseUrl(normalizedValue) : undefined;
}

3
toju-app/src/app/domains/server-directory/domain/server-directory.models.ts
View File

@@ -45,12 +45,14 @@ export interface DefaultServerDefinition {
}
export interface ServerEndpointVersions {
serverInstanceId?: string | null;
serverVersion?: string | null;
clientVersion?: string | null;
}
export interface ServerEndpoint {
id: string;
instanceId?: string;
name: string;
url: string;
isActive: boolean;
@@ -135,6 +137,7 @@ export interface ServerVersionCompatibilityResult {
}
export interface ServerHealthCheckPayload {
serverInstanceId?: unknown;
serverVersion?: unknown;
}

14
toju-app/src/app/domains/server-directory/feature/invite/invite.component.ts
View File

@@ -134,6 +134,15 @@ export class InviteComponent implements OnInit {
sourceId: context.endpoint.id,
sourceUrl: context.sourceUrl
}));
const resolvedSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: joinResponse.server.sourceId ?? context.endpoint.id,
sourceName: joinResponse.server.sourceName ?? context.endpoint.name,
sourceUrl: joinResponse.server.sourceUrl ?? context.sourceUrl,
signalingUrl: joinResponse.signalingUrl,
fallbackName: joinResponse.server.sourceName ?? context.endpoint.name ?? invite.server.name
}, {
ensureEndpoint: true
});
this.store.dispatch(
RoomsActions.joinRoom({
@@ -145,9 +154,8 @@ export class InviteComponent implements OnInit {
Array.isArray(joinResponse.server.channels) && joinResponse.server.channels.length > 0
? joinResponse.server.channels
: invite.server.channels,
sourceId: context.endpoint.id,
sourceName: context.endpoint.name,
sourceUrl: context.sourceUrl
...resolvedSource,
signalingUrl: joinResponse.signalingUrl
}
})
);

13
toju-app/src/app/domains/server-directory/feature/server-search/server-search.component.ts
View File

@@ -273,13 +273,24 @@ export class ServerSearchComponent implements OnInit {
sourceId: server.sourceId,
sourceUrl: server.sourceUrl
}));
const resolvedSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: response.server.sourceId ?? server.sourceId,
sourceName: response.server.sourceName ?? server.sourceName,
sourceUrl: response.server.sourceUrl ?? server.sourceUrl,
signalingUrl: response.signalingUrl,
fallbackName: response.server.sourceName ?? server.sourceName ?? server.name
}, {
ensureEndpoint: true
});
const resolvedServer = {
...server,
...response.server,
channels:
Array.isArray(response.server.channels) && response.server.channels.length > 0
? response.server.channels
: server.channels
: server.channels,
...resolvedSource,
signalingUrl: response.signalingUrl
};
this.closePasswordDialog();

1
toju-app/src/app/domains/server-directory/index.ts
View File

@@ -1,3 +1,4 @@
export * from './application/server-directory.facade';
export * from './domain/server-directory.constants';
export * from './domain/server-directory.models';
export * from './domain/room-signal-source';

97
toju-app/src/app/domains/server-directory/infrastructure/server-directory-api.service.ts
View File

@@ -29,6 +29,7 @@ import type {
ServerSourceSelector,
UnbanServerMemberRequest
} from '../domain/server-directory.models';
import type { RoomSignalSourceInput } from '../domain/room-signal-source';
@Injectable({ providedIn: 'root' })
export class ServerDirectoryApiService {
@@ -45,14 +46,16 @@ export class ServerDirectoryApiService {
resolveEndpoint(selector?: ServerSourceSelector): ServerEndpoint | null {
if (selector?.sourceId) {
return this.endpointState.servers().find((endpoint) => endpoint.id === selector.sourceId) ?? null;
const endpoint = this.endpointState.servers().find((candidate) => candidate.id === selector.sourceId) ?? null;
return this.endpointState.resolveCanonicalEndpoint(endpoint);
}
if (selector?.sourceUrl) {
return this.endpointState.findServerByUrl(selector.sourceUrl) ?? null;
return this.endpointState.resolveCanonicalEndpoint(this.endpointState.findServerByUrl(selector.sourceUrl) ?? null);
}
return (
return this.endpointState.resolveCanonicalEndpoint(
this.endpointState.activeServer() ??
this.endpointState.servers().find((endpoint) => endpoint.status !== 'incompatible') ??
this.endpointState.servers()[0] ??
@@ -92,6 +95,45 @@ export class ServerDirectoryApiService {
);
}
findServerAcrossActiveEndpoints(serverId: string, source?: RoomSignalSourceInput): Observable<ServerInfo | null> {
const candidateEndpoints = this.getSearchableEndpoints();
const preferredSourceUrl = source?.sourceUrl ? this.endpointState.sanitiseUrl(source.sourceUrl) : undefined;
const prioritizedEndpoints = [...candidateEndpoints].sort((left, right) => {
if (preferredSourceUrl) {
const leftMatches = this.endpointState.sanitiseUrl(left.url) === preferredSourceUrl;
const rightMatches = this.endpointState.sanitiseUrl(right.url) === preferredSourceUrl;
if (leftMatches !== rightMatches) {
return leftMatches ? -1 : 1;
}
}
if (source?.sourceId) {
const leftMatches = left.id === source.sourceId;
const rightMatches = right.id === source.sourceId;
if (leftMatches !== rightMatches) {
return leftMatches ? -1 : 1;
}
}
return 0;
});
if (prioritizedEndpoints.length === 0) {
return this.getServer(serverId, this.resolveSelector(source));
}
return forkJoin(
prioritizedEndpoints.map((endpoint) => this.getServer(serverId, {
sourceId: endpoint.id,
sourceUrl: endpoint.url
}))
).pipe(
map((servers) => servers.find((server): server is ServerInfo => !!server) ?? null)
);
}
registerServer(
server: Omit<ServerInfo, 'createdAt'> & { id?: string; password?: string | null },
selector?: ServerSourceSelector
@@ -221,11 +263,17 @@ export class ServerDirectoryApiService {
}
private resolveBaseServerUrl(selector?: ServerSourceSelector): string {
const resolvedEndpoint = this.resolveEndpoint(selector);
if (resolvedEndpoint) {
return resolvedEndpoint.url;
}
if (selector?.sourceUrl) {
return this.endpointState.sanitiseUrl(selector.sourceUrl);
}
return this.resolveEndpoint(selector)?.url ?? this.endpointState.getPrimaryDefaultServerUrl();
return this.endpointState.getPrimaryDefaultServerUrl();
}
private unwrapServersResponse(response: { servers: ServerInfo[]; total: number } | ServerInfo[]): ServerInfo[] {
@@ -245,7 +293,7 @@ export class ServerDirectoryApiService {
}
private searchAllEndpoints(query: string): Observable<ServerInfo[]> {
const onlineEndpoints = this.endpointState.activeServers().filter((endpoint) => endpoint.status !== 'offline');
const onlineEndpoints = this.getSearchableEndpoints();
if (onlineEndpoints.length === 0) {
return this.searchSingleEndpoint(query, this.getApiBaseUrl(), this.endpointState.activeServer());
@@ -258,7 +306,7 @@ export class ServerDirectoryApiService {
}
private getAllServersFromAllEndpoints(): Observable<ServerInfo[]> {
const onlineEndpoints = this.endpointState.activeServers().filter((endpoint) => endpoint.status !== 'offline');
const onlineEndpoints = this.getSearchableEndpoints();
if (onlineEndpoints.length === 0) {
return this.http.get<{ servers: ServerInfo[]; total: number }>(`${this.getApiBaseUrl()}/servers`).pipe(
@@ -277,6 +325,30 @@ export class ServerDirectoryApiService {
).pipe(map((resultArrays) => resultArrays.flat()));
}
private getSearchableEndpoints(): ServerEndpoint[] {
const activeEndpoints = this.endpointState.activeServers().filter((endpoint) => endpoint.status !== 'offline');
if (activeEndpoints.length > 0) {
return activeEndpoints;
}
const activeServer = this.endpointState.activeServer();
return activeServer ? [activeServer] : [];
}
private resolveSelector(source?: RoomSignalSourceInput): ServerSourceSelector | undefined {
if (source?.sourceId) {
return { sourceId: source.sourceId };
}
if (source?.sourceUrl) {
return { sourceUrl: source.sourceUrl };
}
return undefined;
}
private deduplicateById<T extends { id: string }>(items: T[]): T[] {
const seen = new Set<string>();
@@ -298,13 +370,14 @@ export class ServerDirectoryApiService {
const candidate = server as Record<string, unknown>;
const sourceName = this.getStringValue(candidate['sourceName']);
const sourceUrl = this.getStringValue(candidate['sourceUrl']);
const resolvedSource = this.endpointState.resolveCanonicalEndpoint(source);
return {
id: this.getStringValue(candidate['id']) ?? '',
name: this.getStringValue(candidate['name']) ?? 'Unnamed server',
description: this.getStringValue(candidate['description']),
topic: this.getStringValue(candidate['topic']),
hostName: this.getStringValue(candidate['hostName']) ?? sourceName ?? source?.name ?? 'Unknown API',
hostName: this.getStringValue(candidate['hostName']) ?? sourceName ?? resolvedSource?.name ?? 'Unknown API',
ownerId: this.getStringValue(candidate['ownerId']),
ownerName: this.getStringValue(candidate['ownerName']),
ownerPublicKey: this.getStringValue(candidate['ownerPublicKey']),
@@ -319,9 +392,13 @@ export class ServerDirectoryApiService {
roleAssignments: this.getRoleAssignmentsValue(candidate['roleAssignments']),
channelPermissions: this.getChannelPermissionOverridesValue(candidate['channelPermissions']),
createdAt: this.getNumberValue(candidate['createdAt'], Date.now()),
sourceId: this.getStringValue(candidate['sourceId']) ?? source?.id,
sourceName: sourceName ?? source?.name,
sourceUrl: sourceUrl ? this.endpointState.sanitiseUrl(sourceUrl) : source ? this.endpointState.sanitiseUrl(source.url) : undefined
sourceId: this.getStringValue(candidate['sourceId']) ?? resolvedSource?.id,
sourceName: sourceName ?? resolvedSource?.name,
sourceUrl: sourceUrl
? this.endpointState.sanitiseUrl(sourceUrl)
: resolvedSource
? this.endpointState.sanitiseUrl(resolvedSource.url)
: undefined
};
}

5
toju-app/src/app/domains/server-directory/infrastructure/server-endpoint-health.service.ts
View File

@@ -30,12 +30,16 @@ export class ServerEndpointHealthService {
payload.serverVersion,
clientVersion
);
const serverInstanceId = typeof payload.serverInstanceId === 'string' && payload.serverInstanceId.trim().length > 0
? payload.serverInstanceId.trim()
: undefined;
if (!versionCompatibility.isCompatible) {
return {
status: 'incompatible',
latency,
versions: {
serverInstanceId,
serverVersion: versionCompatibility.serverVersion,
clientVersion
}
@@ -46,6 +50,7 @@ export class ServerEndpointHealthService {
status: 'online',
latency,
versions: {
serverInstanceId,
serverVersion: versionCompatibility.serverVersion,
clientVersion
}

22
toju-app/src/app/domains/voice-connection/application/voice-activity.service.ts
View File

@@ -75,6 +75,12 @@ export class VoiceActivityService implements OnDestroy {
this.untrackStream(peerId);
})
);
this.subs.push(
this.voiceConnection.onVoiceConnected.subscribe(() => {
this.ensureAllRemoteStreamsTracked();
})
);
}
trackLocalMic(userId: string, stream: MediaStream): void {
@@ -124,6 +130,10 @@ export class VoiceActivityService implements OnDestroy {
}
const ctx = new AudioContext();
if (ctx.state === 'suspended')
void ctx.resume();
const analyser = ctx.createAnalyser();
const sources = audioTracks.map((track) => ctx.createMediaStreamSource(new MediaStream([track])));
@@ -166,6 +176,18 @@ export class VoiceActivityService implements OnDestroy {
this.stopPolling();
}
private ensureAllRemoteStreamsTracked(): void {
const peers = this.voiceConnection.getConnectedPeers();
for (const peerId of peers) {
const stream = this.voiceConnection.getRemoteVoiceStream(peerId);
if (stream) {
this.trackStream(peerId, stream);
}
}
}
private ensurePolling(): void {
if (this.animFrameId !== null)
return;

173
toju-app/src/app/features/servers/servers-rail.component.ts
View File

@@ -18,6 +18,8 @@ import {
EMPTY,
Subject,
catchError,
firstValueFrom,
from,
switchMap,
tap
} from 'rxjs';
@@ -337,35 +339,44 @@ export class ServersRailComponent {
this.joinPasswordError.set(null);
return this.serverDirectory.requestJoin({
roomId: room.id,
userId: currentUserId,
userPublicKey: currentUser?.oderId || currentUserId,
displayName: currentUser?.displayName || 'Anonymous',
password: password?.trim() || undefined
}, {
sourceId: room.sourceId,
sourceUrl: room.sourceUrl
})
.pipe(
tap((response) => {
this.closePasswordDialog();
this.store.dispatch(
RoomsActions.updateRoom({
roomId: room.id,
changes: this.toRoomRefreshChanges(room, response.server)
return from(this.resolveRoomJoinTarget(room)).pipe(
switchMap((joinTarget) => {
if (!joinTarget.selector) {
if (this.currentRoom()?.id === room.id) {
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: true }));
}
return EMPTY;
}
return this.serverDirectory.requestJoin({
roomId: room.id,
userId: currentUserId,
userPublicKey: currentUser?.oderId || currentUserId,
displayName: currentUser?.displayName || 'Anonymous',
password: password?.trim() || undefined
}, joinTarget.selector)
.pipe(
tap((response) => {
this.closePasswordDialog();
this.store.dispatch(
RoomsActions.updateRoom({
roomId: room.id,
changes: this.toRoomRefreshChanges(joinTarget.room, response.server, response.signalingUrl)
})
);
if (this.currentRoom()?.id === room.id) {
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
}
})
);
if (this.currentRoom()?.id === room.id) {
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
}
}),
catchError((error: unknown) => {
this.handleBackgroundJoinError(room, error);
return EMPTY;
})
);
}),
catchError((error: unknown) => {
this.handleBackgroundJoinError(room, error);
return EMPTY;
})
);
}
private handleBackgroundJoinError(room: Room, error: unknown): void {
@@ -414,7 +425,17 @@ export class ServersRailComponent {
|| (typeof status === 'number' && status >= 500);
}
private toRoomRefreshChanges(room: Room, server: ServerInfo): Partial<Room> {
private toRoomRefreshChanges(room: Room, server: ServerInfo, signalingUrl?: string): Partial<Room> {
const resolvedSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: server.sourceId ?? room.sourceId,
sourceName: server.sourceName ?? room.sourceName,
sourceUrl: server.sourceUrl ?? room.sourceUrl,
signalingUrl,
fallbackName: server.sourceName ?? room.sourceName ?? room.name
}, {
ensureEndpoint: true
});
return {
name: server.name,
description: server.description,
@@ -432,9 +453,99 @@ export class ServersRailComponent {
Array.isArray(server.channels) && server.channels.length > 0
? server.channels
: room.channels,
sourceId: server.sourceId ?? room.sourceId,
sourceName: server.sourceName ?? room.sourceName,
sourceUrl: server.sourceUrl ?? room.sourceUrl
...resolvedSource
};
}
private async resolveRoomJoinTarget(room: Room): Promise<{
room: Room;
selector: ReturnType<ServerDirectoryFacade['buildRoomSignalSelector']>;
}> {
let resolvedRoom = this.applyResolvedRoomSource(room, this.serverDirectory.normaliseRoomSignalSource({
sourceId: room.sourceId,
sourceName: room.sourceName,
sourceUrl: room.sourceUrl,
fallbackName: room.sourceName ?? room.name
}, {
ensureEndpoint: !!room.sourceUrl
}));
let selector = this.serverDirectory.buildRoomSignalSelector({
sourceId: resolvedRoom.sourceId,
sourceName: resolvedRoom.sourceName,
sourceUrl: resolvedRoom.sourceUrl,
fallbackName: resolvedRoom.sourceName ?? resolvedRoom.name
}, {
ensureEndpoint: !!resolvedRoom.sourceUrl
});
const authoritativeServer = (
selector
? await firstValueFrom(this.serverDirectory.getServer(room.id, selector))
: null
) ?? await firstValueFrom(this.serverDirectory.findServerAcrossActiveEndpoints(room.id, {
sourceId: resolvedRoom.sourceId,
sourceName: resolvedRoom.sourceName,
sourceUrl: resolvedRoom.sourceUrl,
fallbackName: resolvedRoom.sourceName ?? resolvedRoom.name
}));
if (!authoritativeServer) {
return {
room: resolvedRoom,
selector
};
}
const authoritativeSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: authoritativeServer.sourceId ?? resolvedRoom.sourceId,
sourceName: authoritativeServer.sourceName ?? resolvedRoom.sourceName,
sourceUrl: authoritativeServer.sourceUrl ?? resolvedRoom.sourceUrl,
fallbackName: authoritativeServer.sourceName ?? resolvedRoom.sourceName ?? resolvedRoom.name
}, {
ensureEndpoint: !!(authoritativeServer.sourceUrl ?? resolvedRoom.sourceUrl)
});
resolvedRoom = this.applyResolvedRoomSource(resolvedRoom, authoritativeSource);
selector = this.serverDirectory.buildRoomSignalSelector({
sourceId: resolvedRoom.sourceId,
sourceName: resolvedRoom.sourceName,
sourceUrl: resolvedRoom.sourceUrl,
fallbackName: resolvedRoom.sourceName ?? resolvedRoom.name
}, {
ensureEndpoint: !!resolvedRoom.sourceUrl
});
return {
room: resolvedRoom,
selector
};
}
private applyResolvedRoomSource(room: Room, source: Pick<Room, 'sourceId' | 'sourceName' | 'sourceUrl'>): Room {
const nextRoom: Room = {
...room,
sourceId: source.sourceId,
sourceName: source.sourceName,
sourceUrl: source.sourceUrl
};
if (
room.sourceId === nextRoom.sourceId
&& room.sourceName === nextRoom.sourceName
&& room.sourceUrl === nextRoom.sourceUrl
) {
return room;
}
this.store.dispatch(RoomsActions.updateRoom({
roomId: room.id,
changes: {
sourceId: nextRoom.sourceId,
sourceName: nextRoom.sourceName,
sourceUrl: nextRoom.sourceUrl
}
}));
return nextRoom;
}
}

6
toju-app/src/app/infrastructure/realtime/README.md
View File

@@ -117,6 +117,12 @@ The signaling layer's only job is getting two peers to exchange SDP offers/answe
Each signaling URL gets its own `SignalingManager` (one WebSocket each). `SignalingTransportHandler` picks the right socket based on which server the message is for. `ServerSignalingCoordinator` tracks which peers belong to which servers and which signaling URLs, so we know when it is safe to tear down a peer connection after leaving a server.
Room affinity is authoritative at this layer as well. The renderer repairs each room's saved `sourceId` / `sourceUrl` from server-directory responses and routes `join_server`, `view_server`, and room-scoped signaling traffic to that room's signaling URL first. If that route fails, alternate endpoints can be tried temporarily, but server-scoped raw messages are no longer broadcast to every connected signaling manager when the route is unknown.
Cold-start routing now waits for the initial server-directory health probes so same-backend aliases can collapse to one canonical signaling endpoint before any saved rooms reconnect. When a room is reconnected on a chosen socket, its background rooms are re-joined on that same socket as well so stale per-signal memberships do not keep orphan managers alive, and reconnect replay only sends `view_server` for rooms that manager still has joined.
This is still a non-federated model. Different signaling servers do not share peer registries or relay WebRTC offers for each other, so users in the same room must converge on the same signaling endpoint to discover one another reliably.
```mermaid
sequenceDiagram
participant UI as App

1
toju-app/src/app/infrastructure/realtime/realtime.constants.ts
View File

@@ -84,6 +84,7 @@ export const SIGNALING_TYPE_CONNECTED = 'connected';
export const SIGNALING_TYPE_SERVER_USERS = 'server_users';
export const SIGNALING_TYPE_USER_JOINED = 'user_joined';
export const SIGNALING_TYPE_USER_LEFT = 'user_left';
export const SIGNALING_TYPE_ACCESS_DENIED = 'access_denied';
export const P2P_TYPE_STATE_REQUEST = 'state-request';
export const P2P_TYPE_VOICE_STATE_REQUEST = 'voice-state-request';

39
toju-app/src/app/infrastructure/realtime/signaling/server-membership-signaling-handler.ts
View File

@@ -33,6 +33,8 @@ export class ServerMembershipSignalingHandler<TMessage> {
return;
}
this.migrateServerSignalUrl(roomId, resolvedSignalUrl);
this.dependencies.signalingCoordinator.setServerSignalUrl(roomId, resolvedSignalUrl);
this.dependencies.signalingCoordinator.setLastJoinedServer(resolvedSignalUrl, {
serverId: roomId,
@@ -55,6 +57,8 @@ export class ServerMembershipSignalingHandler<TMessage> {
return;
}
this.migrateServerSignalUrl(serverId, resolvedSignalUrl);
this.dependencies.signalingCoordinator.setServerSignalUrl(serverId, resolvedSignalUrl);
this.dependencies.signalingCoordinator.setLastJoinedServer(resolvedSignalUrl, {
serverId,
@@ -99,7 +103,9 @@ export class ServerMembershipSignalingHandler<TMessage> {
return;
}
for (const { signalUrl, serverIds } of this.dependencies.signalingCoordinator.getJoinedServerEntries()) {
const joinedEntries = this.dependencies.signalingCoordinator.getJoinedServerEntries();
for (const { signalUrl, serverIds } of joinedEntries) {
for (const joinedServerId of serverIds) {
this.dependencies.signalingTransport.sendRawMessageToSignalUrl(signalUrl, {
type: SIGNALING_TYPE_LEAVE_SERVER,
@@ -109,6 +115,11 @@ export class ServerMembershipSignalingHandler<TMessage> {
}
this.dependencies.signalingCoordinator.clearJoinedServers();
for (const { signalUrl } of joinedEntries) {
this.dependencies.signalingCoordinator.pruneUnusedSignalUrl(signalUrl);
}
this.dependencies.runFullCleanup();
}
@@ -116,11 +127,13 @@ export class ServerMembershipSignalingHandler<TMessage> {
const resolvedSignalUrl = this.dependencies.signalingCoordinator.getServerSignalUrl(serverId);
if (resolvedSignalUrl) {
this.dependencies.signalingCoordinator.removeJoinedServer(resolvedSignalUrl, serverId);
this.dependencies.signalingTransport.sendRawMessageToSignalUrl(resolvedSignalUrl, {
type: SIGNALING_TYPE_LEAVE_SERVER,
serverId
});
this.dependencies.signalingCoordinator.removeJoinedServer(resolvedSignalUrl, serverId);
this.dependencies.signalingCoordinator.pruneUnusedSignalUrl(resolvedSignalUrl);
} else {
this.dependencies.signalingTransport.sendRawMessage({
type: SIGNALING_TYPE_LEAVE_SERVER,
@@ -143,4 +156,26 @@ export class ServerMembershipSignalingHandler<TMessage> {
?? this.dependencies.signalingCoordinator.getServerSignalUrl(serverId)
?? this.getCurrentSignalingUrl();
}
private migrateServerSignalUrl(serverId: string, nextSignalUrl: string): void {
const previousSignalUrl = this.dependencies.signalingCoordinator.getServerSignalUrl(serverId);
if (!previousSignalUrl || previousSignalUrl === nextSignalUrl) {
return;
}
this.dependencies.signalingTransport.sendRawMessageToSignalUrl(previousSignalUrl, {
type: SIGNALING_TYPE_LEAVE_SERVER,
serverId
});
this.dependencies.signalingCoordinator.removeJoinedServer(previousSignalUrl, serverId);
this.dependencies.signalingCoordinator.pruneUnusedSignalUrl(previousSignalUrl);
this.dependencies.logger.info('Migrated server to a new signaling route', {
previousSignalUrl,
serverId,
signalUrl: nextSignalUrl
});
}
}

91
toju-app/src/app/infrastructure/realtime/signaling/server-signaling-coordinator.ts
View File

@@ -110,6 +110,10 @@ export class ServerSignalingCoordinator<TMessage> {
this.lastJoinedServerBySignalUrl.set(signalUrl, joinedServer);
}
getLastJoinedServer(signalUrl: string): JoinedServerInfo | null {
return this.lastJoinedServerBySignalUrl.get(signalUrl) ?? null;
}
clearLastJoinedServers(): void {
this.lastJoinedServerBySignalUrl.clear();
}
@@ -156,15 +160,57 @@ export class ServerSignalingCoordinator<TMessage> {
}
removeJoinedServer(signalUrl: string, serverId: string): void {
this.getOrCreateMemberServerSet(signalUrl).delete(serverId);
const memberServerIds = this.memberServerIdsBySignalUrl.get(signalUrl);
if (!memberServerIds) {
this.repairLastJoinedServer(signalUrl, serverId);
return;
}
memberServerIds.delete(serverId);
if (memberServerIds.size === 0) {
this.memberServerIdsBySignalUrl.delete(signalUrl);
}
this.repairLastJoinedServer(signalUrl, serverId);
}
removeJoinedServerEverywhere(serverId: string): void {
for (const memberServerIds of this.memberServerIdsBySignalUrl.values()) {
memberServerIds.delete(serverId);
for (const signalUrl of Array.from(this.memberServerIdsBySignalUrl.keys())) {
this.removeJoinedServer(signalUrl, serverId);
this.pruneUnusedSignalUrl(signalUrl);
}
}
pruneUnusedSignalUrl(signalUrl: string): void {
if (this.getMemberServerIdsForSignalUrl(signalUrl).size > 0) {
return;
}
this.memberServerIdsBySignalUrl.delete(signalUrl);
this.lastJoinedServerBySignalUrl.delete(signalUrl);
const subscriptions = this.signalingSubscriptions.get(signalUrl);
if (subscriptions) {
for (const subscription of subscriptions) {
subscription.unsubscribe();
}
this.signalingSubscriptions.delete(signalUrl);
}
const manager = this.signalingManagers.get(signalUrl);
if (manager) {
manager.destroy();
this.signalingManagers.delete(signalUrl);
}
this.removeSignalUrlFromPeerTracking(signalUrl);
}
getMemberServerIdsForSignalUrl(signalUrl: string): ReadonlySet<string> {
return this.memberServerIdsBySignalUrl.get(signalUrl) ?? new Set<string>();
}
@@ -344,6 +390,45 @@ export class ServerSignalingCoordinator<TMessage> {
return trackedServerIds;
}
private repairLastJoinedServer(signalUrl: string, removedServerId: string): void {
const lastJoined = this.lastJoinedServerBySignalUrl.get(signalUrl);
if (!lastJoined) {
return;
}
const memberServerIds = this.memberServerIdsBySignalUrl.get(signalUrl);
if (!memberServerIds || memberServerIds.size === 0) {
this.lastJoinedServerBySignalUrl.delete(signalUrl);
return;
}
if (lastJoined.serverId !== removedServerId && memberServerIds.has(lastJoined.serverId)) {
return;
}
const nextServerId = memberServerIds.values().next().value as string | undefined;
if (!nextServerId) {
this.lastJoinedServerBySignalUrl.delete(signalUrl);
return;
}
this.lastJoinedServerBySignalUrl.set(signalUrl, {
...lastJoined,
serverId: nextServerId
});
}
private removeSignalUrlFromPeerTracking(signalUrl: string): void {
const peerIds = new Set<string>([...this.peerKnownSignalUrls.keys(), ...this.peerServerMap.keys()]);
for (const peerId of peerIds) {
this.removePeerSignalScope(peerId, signalUrl);
}
}
private removePeerSignalScope(peerId: string, signalUrl: string): void {
const trackedSignalUrls = this.peerServerMap.get(peerId);

15
toju-app/src/app/infrastructure/realtime/signaling/signaling-message-handler.ts
View File

@@ -1,6 +1,7 @@
import type { SignalingMessage } from '../../../shared-kernel';
import { PeerData } from '../realtime.types';
import {
SIGNALING_TYPE_ACCESS_DENIED,
SIGNALING_TYPE_ANSWER,
SIGNALING_TYPE_CONNECTED,
SIGNALING_TYPE_ICE_CANDIDATE,
@@ -98,6 +99,10 @@ export class IncomingSignalingMessageHandler {
this.handleIceCandidateSignalingMessage(message, signalUrl);
return;
case SIGNALING_TYPE_ACCESS_DENIED:
this.handleAccessDeniedSignalingMessage(message, signalUrl);
return;
default:
return;
}
@@ -357,6 +362,16 @@ export class IncomingSignalingMessageHandler {
this.dependencies.peerManager.handleIceCandidate(fromUserId, candidate);
}
private handleAccessDeniedSignalingMessage(message: IncomingSignalingMessage, signalUrl: string): void {
if (!message.serverId) {
return;
}
// Remove the server from the coordinator for this signal URL so it won't
// be re-joined on the next reconnect cycle.
this.dependencies.signalingCoordinator.removeJoinedServer(signalUrl, message.serverId);
}
private scheduleUserJoinedFallbackOffer(peerId: string, signalUrl: string, serverId?: string): void {
this.clearUserJoinedFallbackOffer(peerId);

30
toju-app/src/app/infrastructure/realtime/signaling/signaling-transport-handler.ts
View File

@@ -118,6 +118,13 @@ export class SignalingTransportHandler<TMessage> {
if (serverSignalUrl && this.sendRawMessageToSignalUrl(serverSignalUrl, message)) {
return;
}
this.dependencies.logger.warn('[signaling] Missing server signal route for outbound raw message', {
serverId,
type: messageType
});
return;
}
const connectedManagers = this.getConnectedSignalingManagers();
@@ -161,14 +168,14 @@ export class SignalingTransportHandler<TMessage> {
displayName: normalizedDisplayName
};
const identifyMessage = {
type: SIGNALING_TYPE_IDENTIFY,
oderId,
displayName: normalizedDisplayName
};
if (signalUrl) {
this.sendRawMessageToSignalUrl(signalUrl, identifyMessage);
this.sendRawMessageToSignalUrl(signalUrl, {
type: SIGNALING_TYPE_IDENTIFY,
oderId,
displayName: normalizedDisplayName,
connectionScope: signalUrl
});
return;
}
@@ -178,8 +185,13 @@ export class SignalingTransportHandler<TMessage> {
return;
}
for (const { manager } of connectedManagers) {
manager.sendRawMessage(identifyMessage);
for (const { signalUrl: managerSignalUrl, manager } of connectedManagers) {
manager.sendRawMessage({
type: SIGNALING_TYPE_IDENTIFY,
oderId,
displayName: normalizedDisplayName,
connectionScope: managerSignalUrl
});
}
}
}

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

@@ -283,7 +283,8 @@ export class SignalingManager {
if (credentials) {
this.sendRawMessage({ type: SIGNALING_TYPE_IDENTIFY,
oderId: credentials.oderId,
displayName: credentials.displayName });
displayName: credentials.displayName,
connectionScope: this.lastSignalingUrl ?? undefined });
}
const memberIds = this.getMemberServerIds();
@@ -296,17 +297,10 @@ export class SignalingManager {
const lastJoined = this.getLastJoinedServer();
if (lastJoined) {
if (lastJoined && memberIds.has(lastJoined.serverId)) {
this.sendRawMessage({ type: SIGNALING_TYPE_VIEW_SERVER,
serverId: lastJoined.serverId });
}
} else {
const lastJoined = this.getLastJoinedServer();
if (lastJoined) {
this.sendRawMessage({ type: SIGNALING_TYPE_JOIN_SERVER,
serverId: lastJoined.serverId });
}
}
}

6
toju-app/src/app/store/rooms/rooms.actions.ts
View File

@@ -37,7 +37,11 @@ export const RoomsActions = createActionGroup({
'Create Room Success': props<{ room: Room }>(),
'Create Room Failure': props<{ error: string }>(),
'Join Room': props<{ roomId: string; password?: string; serverInfo?: Partial<ServerInfo> & { name: string } }>(),
'Join Room': props<{
roomId: string;
password?: string;
serverInfo?: Partial<ServerInfo> & { name: string; signalingUrl?: string };
}>(),
'Join Room Success': props<{ room: Room }>(),
'Join Room Failure': props<{ error: string }>(),

454
toju-app/src/app/store/rooms/rooms.effects.ts
View File

@@ -13,6 +13,7 @@ import {
of,
from,
EMPTY,
firstValueFrom,
merge
} from 'rxjs';
import {
@@ -43,7 +44,10 @@ import {
saveLastViewedChatToStorage
} from '../../infrastructure/persistence';
import {
areRoomSignalSourcesEqual,
CLIENT_UPDATE_REQUIRED_MESSAGE,
type RoomSignalSource,
type ServerInfo,
type ServerSourceSelector,
ServerDirectoryFacade
} from '../../domains/server-directory';
@@ -169,6 +173,12 @@ interface RoomPresenceSignalingMessage {
displayName?: string;
}
interface RoomSignalConnectionPlan {
fallbackSources: RoomSignalSource[];
primarySource: RoomSignalSource | null;
room: Room;
}
type BlockedRoomAccessAction =
| ReturnType<typeof RoomsActions.forgetRoom>
| ReturnType<typeof RoomsActions.joinRoomFailure>;
@@ -196,6 +206,7 @@ export class RoomsEffects {
* join/leave sound within {@link RECONNECT_SOUND_GRACE_MS}.
*/
private recentlyLeftVoiceTimestamps = new Map<string, number>();
private readonly roomSignalFallbackSources = new Map<string, RoomSignalSource>();
private roomNavigationRequestVersion = 0;
private latestNavigatedRoomId: string | null = null;
@@ -388,13 +399,27 @@ export class RoomsEffects {
return from(this.db.getRoom(roomId)).pipe(
switchMap((room) => {
const sourceSelector = serverInfo
? {
? this.serverDirectory.buildRoomSignalSelector({
sourceId: serverInfo.sourceId,
sourceUrl: serverInfo.sourceUrl
}
sourceName: serverInfo.sourceName,
sourceUrl: serverInfo.sourceUrl,
signalingUrl: serverInfo.signalingUrl,
fallbackName: serverInfo.sourceName ?? serverInfo.name
}, {
ensureEndpoint: !!(serverInfo.sourceUrl ?? serverInfo.signalingUrl)
})
: undefined;
if (room) {
const resolvedSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: serverInfo?.sourceId ?? room.sourceId,
sourceName: serverInfo?.sourceName ?? room.sourceName,
sourceUrl: serverInfo?.sourceUrl ?? room.sourceUrl,
signalingUrl: serverInfo?.signalingUrl,
fallbackName: serverInfo?.sourceName ?? room.sourceName ?? serverInfo?.name ?? room.name
}, {
ensureEndpoint: !!(serverInfo?.sourceUrl ?? room.sourceUrl ?? serverInfo?.signalingUrl)
});
const resolvedRoom: Room = {
...room,
isPrivate: typeof serverInfo?.isPrivate === 'boolean' ? serverInfo.isPrivate : room.isPrivate,
@@ -403,9 +428,7 @@ export class RoomsEffects {
roles: serverInfo?.roles ?? room.roles,
roleAssignments: serverInfo?.roleAssignments ?? room.roleAssignments,
channelPermissions: serverInfo?.channelPermissions ?? room.channelPermissions,
sourceId: serverInfo?.sourceId ?? room.sourceId,
sourceName: serverInfo?.sourceName ?? room.sourceName,
sourceUrl: serverInfo?.sourceUrl ?? room.sourceUrl,
...resolvedSource,
hasPassword:
typeof serverInfo?.hasPassword === 'boolean'
? serverInfo.hasPassword
@@ -430,6 +453,15 @@ export class RoomsEffects {
// If not in local DB but we have server info from search, create a room entry
if (serverInfo) {
const resolvedSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: serverInfo.sourceId,
sourceName: serverInfo.sourceName,
sourceUrl: serverInfo.sourceUrl,
signalingUrl: serverInfo.signalingUrl,
fallbackName: serverInfo.sourceName ?? serverInfo.name
}, {
ensureEndpoint: !!(serverInfo.sourceUrl ?? serverInfo.signalingUrl)
});
const newRoom: Room = {
id: roomId,
name: serverInfo.name,
@@ -445,9 +477,7 @@ export class RoomsEffects {
roles: serverInfo.roles,
roleAssignments: serverInfo.roleAssignments,
channelPermissions: serverInfo.channelPermissions,
sourceId: serverInfo.sourceId,
sourceName: serverInfo.sourceName,
sourceUrl: serverInfo.sourceUrl
...resolvedSource
};
// Save to local DB for future reference
@@ -459,6 +489,14 @@ export class RoomsEffects {
return this.serverDirectory.getServer(roomId, sourceSelector).pipe(
switchMap((serverData) => {
if (serverData) {
const resolvedSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: serverData.sourceId,
sourceName: serverData.sourceName,
sourceUrl: serverData.sourceUrl,
fallbackName: serverData.sourceName ?? serverData.name
}, {
ensureEndpoint: !!serverData.sourceUrl
});
const newRoom: Room = {
id: serverData.id,
name: serverData.name,
@@ -474,9 +512,7 @@ export class RoomsEffects {
roles: serverData.roles,
roleAssignments: serverData.roleAssignments,
channelPermissions: serverData.channelPermissions,
sourceId: serverData.sourceId,
sourceName: serverData.sourceName,
sourceUrl: serverData.sourceUrl
...resolvedSource
};
this.db.saveRoom(newRoom);
@@ -621,16 +657,32 @@ export class RoomsEffects {
refreshServerOwnedRoomMetadata$ = createEffect(() =>
this.actions$.pipe(
ofType(RoomsActions.joinRoomSuccess, RoomsActions.viewServerSuccess),
switchMap(({ room }) =>
this.serverDirectory.getServer(room.id, {
sourceId: room.sourceId,
sourceUrl: room.sourceUrl
}).pipe(
switchMap(({ room }) => {
const source = this.resolveRoomSignalSource(room);
const selector = this.resolveRoomSignalSelector(source, room.name);
const roomRequest$ = selector
? this.serverDirectory.getServer(room.id, selector).pipe(
switchMap((serverData) => serverData
? of(serverData)
: this.serverDirectory.findServerAcrossActiveEndpoints(room.id, source))
)
: this.serverDirectory.findServerAcrossActiveEndpoints(room.id, source);
return roomRequest$.pipe(
map((serverData) => {
if (!serverData) {
return null;
}
const resolvedSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: serverData.sourceId ?? room.sourceId,
sourceName: serverData.sourceName ?? room.sourceName,
sourceUrl: serverData.sourceUrl ?? room.sourceUrl,
fallbackName: serverData.sourceName ?? room.sourceName ?? room.name
}, {
ensureEndpoint: !!(serverData.sourceUrl ?? room.sourceUrl)
});
return RoomsActions.updateRoom({
roomId: room.id,
changes: {
@@ -645,16 +697,14 @@ export class RoomsEffects {
roles: serverData.roles ?? room.roles,
roleAssignments: serverData.roleAssignments ?? room.roleAssignments,
channelPermissions: serverData.channelPermissions ?? room.channelPermissions,
sourceId: serverData.sourceId ?? room.sourceId,
sourceName: serverData.sourceName ?? room.sourceName,
sourceUrl: serverData.sourceUrl ?? room.sourceUrl
...resolvedSource
}
});
}),
filter((action): action is ReturnType<typeof RoomsActions.updateRoom> => !!action),
catchError(() => EMPTY)
)
)
);
})
)
);
@@ -732,6 +782,7 @@ export class RoomsEffects {
),
switchMap(([{ roomId }]) => {
this.db.deleteRoom(roomId);
this.roomSignalFallbackSources.delete(roomId);
this.webrtc.broadcastMessage({ type: 'room-deleted',
roomId });
@@ -824,6 +875,7 @@ export class RoomsEffects {
// Delete from local DB
this.db.deleteRoom(roomId);
this.roomSignalFallbackSources.delete(roomId);
// Leave this specific server (doesn't affect other servers)
this.webrtc.leaveRoom(roomId);
@@ -1322,6 +1374,12 @@ export class RoomsEffects {
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 })];
}
@@ -1997,23 +2055,127 @@ export class RoomsEffects {
): Promise<void> {
const shouldShowCompatibilityError = options.showCompatibilityError ?? false;
const navigationRequestVersion = options.navigationRequestVersion;
const compatibilitySelector = this.resolveCompatibilitySelector(room);
const isCompatible = compatibilitySelector === null
? true
: await this.serverDirectory.ensureEndpointVersionCompatibility(compatibilitySelector);
const isViewedRoom = () => room.id === this.latestNavigatedRoomId;
await this.serverDirectory.awaitInitialServerHealthCheck();
if (!this.isCurrentRoomNavigation(room.id, navigationRequestVersion)) {
return;
}
if (!isCompatible) {
if (shouldShowCompatibilityError) {
this.store.dispatch(
RoomsActions.setSignalServerCompatibilityError({ message: CLIENT_UPDATE_REQUIRED_MESSAGE })
);
const connectionPlan = await this.resolveRoomSignalConnectionPlan(room);
if (!this.isCurrentRoomNavigation(room.id, navigationRequestVersion)) {
return;
}
const sessionFallbackSource = this.roomSignalFallbackSources.get(room.id);
const connectionCandidates: {
isExistingFallback?: boolean;
isFallback?: boolean;
isPrimary?: boolean;
source: RoomSignalSource;
}[] = [];
const pushConnectionCandidate = (
source: RoomSignalSource | null | undefined,
flags: { isExistingFallback?: boolean; isFallback?: boolean; isPrimary?: boolean } = {}
) => {
if (!source || !this.resolveRoomSignalSelector(source, room.name)) {
return;
}
if (connectionCandidates.some((candidate) => areRoomSignalSourcesEqual(candidate.source, source))) {
return;
}
connectionCandidates.push({
...flags,
source
});
};
if (sessionFallbackSource && this.webrtc.hasJoinedServer(room.id)) {
pushConnectionCandidate(sessionFallbackSource, { isExistingFallback: true, isFallback: true });
}
pushConnectionCandidate(connectionPlan.primarySource, { isPrimary: true });
for (const fallbackSource of connectionPlan.fallbackSources) {
pushConnectionCandidate(fallbackSource, { isFallback: true });
}
let attemptedFallback = false;
for (const candidate of connectionCandidates) {
const selector = this.resolveRoomSignalSelector(candidate.source, connectionPlan.room.name);
if (!selector) {
continue;
}
const isCompatible = await this.serverDirectory.ensureEndpointVersionCompatibility(selector);
if (!this.isCurrentRoomNavigation(room.id, navigationRequestVersion)) {
return;
}
if (!isCompatible) {
if (candidate.isPrimary) {
if (shouldShowCompatibilityError) {
this.store.dispatch(
RoomsActions.setSignalServerCompatibilityError({ message: CLIENT_UPDATE_REQUIRED_MESSAGE })
);
}
if (isViewedRoom()) {
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
}
return;
}
continue;
}
if (candidate.isFallback && !candidate.isExistingFallback && !attemptedFallback) {
attemptedFallback = true;
if (isViewedRoom()) {
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: true }));
}
}
const connected = await this.connectRoomToSignalSource(
connectionPlan.room,
candidate.source,
user,
resolvedOderId,
savedRooms,
navigationRequestVersion
);
if (!this.isCurrentRoomNavigation(room.id, navigationRequestVersion)) {
return;
}
if (!connected) {
continue;
}
if (candidate.isFallback) {
this.roomSignalFallbackSources.set(room.id, candidate.source);
} else {
this.roomSignalFallbackSources.delete(room.id);
}
if (shouldShowCompatibilityError) {
this.store.dispatch(RoomsActions.setSignalServerCompatibilityError({ message: null }));
}
if (isViewedRoom()) {
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
}
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: false }));
return;
}
@@ -2021,49 +2183,9 @@ export class RoomsEffects {
this.store.dispatch(RoomsActions.setSignalServerCompatibilityError({ message: null }));
}
const wsUrl = this.serverDirectory.getWebSocketUrl({
sourceId: room.sourceId,
sourceUrl: room.sourceUrl
});
const oderId = resolvedOderId || user?.oderId || this.webrtc.peerId();
const displayName = resolveUserDisplayName(user);
const sameSignalRooms = this.getRoomsForSignalingUrl(this.includeRoom(savedRooms, room), wsUrl);
const backgroundRooms = sameSignalRooms.filter((candidate) => candidate.id !== room.id);
const joinCurrentEndpointRooms = () => {
if (!this.isCurrentRoomNavigation(room.id, navigationRequestVersion)) {
return;
}
this.webrtc.setCurrentServer(room.id);
this.webrtc.identify(oderId, displayName, wsUrl);
for (const backgroundRoom of backgroundRooms) {
if (!this.webrtc.hasJoinedServer(backgroundRoom.id)) {
this.webrtc.joinRoom(backgroundRoom.id, oderId, wsUrl);
}
}
if (this.webrtc.hasJoinedServer(room.id)) {
this.webrtc.switchServer(room.id, oderId, wsUrl);
} else {
this.webrtc.joinRoom(room.id, oderId, wsUrl);
}
};
if (this.webrtc.isSignalingConnectedTo(wsUrl)) {
joinCurrentEndpointRooms();
return;
if (isViewedRoom()) {
this.store.dispatch(RoomsActions.setSignalServerReconnecting({ isReconnecting: true }));
}
this.webrtc.connectToSignalingServer(wsUrl).subscribe({
next: (connected) => {
if (!connected || !this.isCurrentRoomNavigation(room.id, navigationRequestVersion))
return;
joinCurrentEndpointRooms();
},
error: () => {}
});
}
private syncSavedRoomConnections(user: User | null, currentRoom: Room | null, savedRooms: Room[]): void {
@@ -2076,10 +2198,12 @@ export class RoomsEffects {
const roomsBySignalingUrl = new Map<string, Room[]>();
for (const room of roomsToSync) {
const wsUrl = this.serverDirectory.getWebSocketUrl({
sourceId: room.sourceId,
sourceUrl: room.sourceUrl
});
const wsUrl = this.resolveRoomSignalingUrl(room);
if (!wsUrl) {
continue;
}
const groupedRooms = roomsBySignalingUrl.get(wsUrl) ?? [];
if (!groupedRooms.some((groupedRoom) => groupedRoom.id === room.id)) {
@@ -2149,10 +2273,7 @@ export class RoomsEffects {
continue;
}
if (this.serverDirectory.getWebSocketUrl({
sourceId: room.sourceId,
sourceUrl: room.sourceUrl
}) !== wsUrl) {
if (this.resolveRoomSignalingUrl(room) !== wsUrl) {
continue;
}
@@ -2202,6 +2323,175 @@ export class RoomsEffects {
return localStorage.getItem('metoyou_currentUserId');
}
private async resolveRoomSignalConnectionPlan(room: Room): Promise<RoomSignalConnectionPlan> {
let resolvedRoom = this.repairRoomSignalSource(room, this.resolveRoomSignalSource(room));
let primarySource = this.resolveRoomSignalSource(resolvedRoom);
if (!this.webrtc.hasJoinedServer(room.id)) {
const selector = this.resolveRoomSignalSelector(primarySource, resolvedRoom.name);
const authoritativeServer = (
selector
? await firstValueFrom(this.serverDirectory.getServer(room.id, selector))
: null
) ?? await firstValueFrom(this.serverDirectory.findServerAcrossActiveEndpoints(room.id, primarySource));
if (authoritativeServer) {
const authoritativeSource = this.serverDirectory.normaliseRoomSignalSource({
sourceId: authoritativeServer.sourceId ?? primarySource.sourceId,
sourceName: authoritativeServer.sourceName ?? primarySource.sourceName,
sourceUrl: authoritativeServer.sourceUrl ?? primarySource.sourceUrl,
fallbackName: authoritativeServer.sourceName ?? primarySource.sourceName ?? resolvedRoom.name
}, {
ensureEndpoint: !!(authoritativeServer.sourceUrl ?? primarySource.sourceUrl)
});
resolvedRoom = this.repairRoomSignalSource(resolvedRoom, authoritativeSource);
primarySource = authoritativeSource;
}
}
const fallbackSources = this.serverDirectory.getFallbackRoomEndpoints(primarySource)
.map((endpoint) => this.serverDirectory.normaliseRoomSignalSource({
sourceId: endpoint.id,
sourceName: endpoint.name,
sourceUrl: endpoint.url,
fallbackName: endpoint.name
}))
.filter((source, index, sources) =>
sources.findIndex((candidate) => areRoomSignalSourcesEqual(candidate, source)) === index
);
return {
fallbackSources,
primarySource: this.resolveRoomSignalSelector(primarySource, resolvedRoom.name) ? primarySource : null,
room: resolvedRoom
};
}
private async connectRoomToSignalSource(
room: Room,
source: RoomSignalSource,
user: User | null,
resolvedOderId: string | undefined,
savedRooms: Room[],
navigationRequestVersion?: number
): Promise<boolean> {
const selector = this.resolveRoomSignalSelector(source, room.name);
if (!selector) {
return false;
}
const wsUrl = this.serverDirectory.getWebSocketUrl(selector);
const oderId = resolvedOderId || user?.oderId || this.webrtc.peerId();
const displayName = resolveUserDisplayName(user);
const sameSignalRooms = this.getRoomsForSignalingUrl(this.includeRoom(savedRooms, room), wsUrl);
const backgroundRooms = sameSignalRooms.filter((candidate) => candidate.id !== room.id);
const joinCurrentEndpointRooms = () => {
if (!this.isCurrentRoomNavigation(room.id, navigationRequestVersion)) {
return;
}
this.webrtc.setCurrentServer(room.id);
this.webrtc.identify(oderId, displayName, wsUrl);
for (const backgroundRoom of backgroundRooms) {
this.webrtc.joinRoom(backgroundRoom.id, oderId, wsUrl);
}
if (this.webrtc.hasJoinedServer(room.id)) {
this.webrtc.switchServer(room.id, oderId, wsUrl);
} else {
this.webrtc.joinRoom(room.id, oderId, wsUrl);
}
};
if (this.webrtc.isSignalingConnectedTo(wsUrl)) {
joinCurrentEndpointRooms();
return true;
}
try {
const connected = await firstValueFrom(this.webrtc.connectToSignalingServer(wsUrl));
if (!connected || !this.isCurrentRoomNavigation(room.id, navigationRequestVersion)) {
return false;
}
joinCurrentEndpointRooms();
return true;
} catch {
return false;
}
}
private resolveRoomSignalSource(
room: Pick<Room, 'name' | 'sourceId' | 'sourceName' | 'sourceUrl'>
): RoomSignalSource {
return this.serverDirectory.normaliseRoomSignalSource({
sourceId: room.sourceId,
sourceName: room.sourceName,
sourceUrl: room.sourceUrl,
fallbackName: room.sourceName ?? room.name
}, {
ensureEndpoint: !!room.sourceUrl
});
}
private repairRoomSignalSource(room: Room, source: RoomSignalSource | null): Room {
if (!source || areRoomSignalSourcesEqual(room, source)) {
return room;
}
const changes: Partial<Room> = {
sourceId: source.sourceId,
sourceName: source.sourceName,
sourceUrl: source.sourceUrl
};
this.store.dispatch(RoomsActions.updateRoom({
roomId: room.id,
changes
}));
return {
...room,
...changes
};
}
private resolveRoomSignalSelector(
source: RoomSignalSource | null | undefined,
fallbackName: string
): ServerSourceSelector | undefined {
if (!source) {
return undefined;
}
return this.serverDirectory.buildRoomSignalSelector({
...source,
fallbackName: source.sourceName ?? fallbackName
}, {
ensureEndpoint: !!source.sourceUrl
});
}
private getPreferredRoomSignalSource(room: Room): RoomSignalSource {
const fallbackSource = this.roomSignalFallbackSources.get(room.id);
if (fallbackSource && this.webrtc.hasJoinedServer(room.id)) {
return fallbackSource;
}
return this.resolveRoomSignalSource(room);
}
private resolveRoomSignalingUrl(room: Room): string {
const selector = this.resolveRoomSignalSelector(this.getPreferredRoomSignalSource(room), room.name);
return selector ? this.serverDirectory.getWebSocketUrl(selector) : '';
}
private async getBlockedRoomAccessActions(
roomId: string,
currentUser: { id: string; oderId: string } | null