diff --git a/docs/systems/federation.md b/docs/systems/federation.md index a663b850..9a91b57f 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -403,7 +403,7 @@ The `(source_instance, source_message_id)` pair is checked before insertion. Dup **`sendTypingRelay()` (`federationOutbox.ts`):** - Fetches channel's `federatedId` and `getDmParticipants()` for target resolution - Builds `FederationRelayEvent` with `typing: { homeUserId, homeInstance, username }` -- Reuses `sendCallRelay()` for the actual POST to each remote peer origin +- Calls `sendCallRelay(origin, [event], { peeringTimeoutMs: 0 })` for each remote peer origin — non-active peers are skipped and a background `ensurePeered` warm-up is kicked off instead **Inbound (`federation.ts`):** - `processDmTypingStartEvent` → look up channel by `federatedId`, resolve user via `resolveLocalUser()` (no stub creation for ephemeral events), broadcast `dm_typing` to local members @@ -1024,7 +1024,20 @@ All events carry standard relay fields: `eventType`, `messageId`, `encryptionVer ### Direct Delivery (No Outbox) -Call signaling is time-critical and bypasses the outbox entirely. `sendCallRelay()` sends a synchronous HTTP POST to the peer's `/api/federation/relay` endpoint using existing HMAC signing (`buildFederationHeaders`). If delivery fails, the call operation fails — there is no retry. +**`sendCallRelay(targetPeerOrigin, events, opts?)`** (`federationOutbox.ts`): + +- Latency-sensitive: returns `CallRelayResult = { ok: true } | { ok: false; reason: CallRelayFailureReason; error: string }`. +- Peering resolution: + 1. If the peer row is `active` or `unreachable`, POST directly (the health check restores `unreachable` peers; re-handshaking is wasteful). + 2. Otherwise race `ensurePeered` against `opts.peeringTimeoutMs` (default `CALL_PEERING_TIMEOUT_MS = 3_000` ms). The background handshake is **not** aborted on race loss — a warn-logged catch is attached so a late-rejecting background promise does not emit `unhandledRejection`. +- Peer-state → reason mapping is exhaustive over the `EnsurePeeredResult` union (`active` / `rejected` / `pending` / `failed`) plus the external `timeout` branch. TypeScript `never` check in the switch default catches future additions. Note: the `livekit_unavailable` reason in `DmCallUndeliverableReason` is emitted separately from `sendFederatedCallStart`'s LiveKit pre-flight in `ws/events.ts`, not from this switch — `sendCallRelay` only produces `CallRelayFailureReason` values (`peer_rejected` / `peer_awaiting_approval` / `peer_transient_failure` / `post_failed`). +- Non-blocking mode: `peeringTimeoutMs: 0` (used by typing) skips the POST for non-active peers, kicks off `ensurePeered` as a background warm-up, returns `peer_transient_failure` silently. + +**`sendTypingRelay(dmChannelId, eventType, userId)`**: + +- Fire-and-forget to each remote DM participant's home instance via `sendCallRelay(origin, [event], { peeringTimeoutMs: 0 })`. Typing is an ephemeral hint — lost packets are acceptable and there is no user-facing failure surface. + +**Call-start failure surfacing.** `sendFederatedCallStart` aggregates targeted-peer results and emits `dm_call_undeliverable` to the caller for failed targeted peers. See `docs/systems/voice.md` and `docs/systems/websocket.md` for the event contract. ### Call Flows @@ -1157,4 +1170,8 @@ If the `federation_mutation_log` table exists but is empty, populates it with `c ## Known Issues -No critical known issues. See `docs/federation-production-roadmap.md` for open items (FED-001 through FED-013). +See `docs/federation-production-roadmap.md` for open items (FED-001 through FED-013). + +- **Accept-relay failure dead end.** If Bob on B accepts a call from Alice on A and the `dm_call_accept` S2S relay back to A fails, Alice's client does not exit the `outgoingCall` state until the 60 s ring timeout fires `dm_call_ended`. The client clears `outgoingCall` only on `dm_call_accepted | rejected | ended` (`useWebSocket.ts`), with no LiveKit participant-join fallback. Surfacing this requires a B-side event and call-state rollback; deferred. +- **End-relay failure dead end.** Similar mechanism, lower severity because LiveKit `ParticipantDisconnected` typically unwinds the voice UI on the host side; local DM call state still lingers to the 60 s timeout. Deferred. +- **Path-B reject-relay failure dead end.** Third-instance user (Carol on C) rings for a call hosted on A via Path B; if her `dm_call_reject` C→A relay fails, A never deducts her from the pending ringees, so her name persists in the caller's "still ringing" set until the 60 s timeout. Same class as accept-failure; deferred. diff --git a/docs/systems/voice.md b/docs/systems/voice.md index f9fe3417..155171b0 100644 --- a/docs/systems/voice.md +++ b/docs/systems/voice.md @@ -56,7 +56,13 @@ DM calls work across federated instances. The caller's instance hosts the LiveKi ### Universal Relay -All `dm_call_*` signaling events (`start`, `accept`, `reject`, `end`) are relayed to every active federation peer in parallel via `Promise.all`. Each `sendCallRelay` call has a 10-second timeout. This is a synchronous HTTP POST to the peer's federation endpoint — it bypasses the outbox worker entirely because call signaling is latency-sensitive and must not be queued. +All `dm_call_*` signaling events (`start`, `accept`, `reject`, `end`) are relayed to every active federation peer in parallel. Each `sendCallRelay` call has a 10-second HTTP timeout. This bypasses the outbox worker — call signaling is latency-sensitive. + +**Auto-peering at send time.** If the target origin has no active peer record, `sendCallRelay` races an `ensurePeered` handshake against a 3 s deadline (`CALL_PEERING_TIMEOUT_MS`). On success the relay POSTs normally; on timeout it returns `peer_transient_failure` without aborting the background handshake, so a subsequent attempt typically succeeds. Typing (`sendTypingRelay`) passes `peeringTimeoutMs: 0` — the POST is skipped for non-active peers and a warm-up `ensurePeered` runs in the background. + +**Call-start failure surface.** `sendFederatedCallStart` aggregates results from its targeted-peer relays (peers whose origin matches a DM member's `homeInstance`) and emits a single `dm_call_undeliverable` event to the caller when any targeted peer fails. `terminal: true` (no successful targeted relay AND no connected local non-caller member) also destroys the local ring room so the caller's UI clears immediately; `terminal: false` leaves the call ringing for reachable recipients. The all-peers broadcast (Path-B fallback to non-member-hosting peers) logs failures only — they are not surfaced. + +Accept and end relay failures are NOT surfaced today; see the federation doc's "Known issues" section for the deferred accept-failure dead-end. ### Dual-Path Processing diff --git a/docs/systems/websocket.md b/docs/systems/websocket.md index 5a51a7bf..9f160208 100644 --- a/docs/systems/websocket.md +++ b/docs/systems/websocket.md @@ -170,6 +170,7 @@ reason: `'displaced'` (new tab) | `'session_closed'` | `dm_call_accepted` | dmChannelId?, federatedCallId? | DM members | | `dm_call_rejected` | dmChannelId?, federatedCallId? | DM members | | `dm_call_ended` | dmChannelId?, federatedCallId? | DM members | +| `dm_call_undeliverable` | Sent to caller when a call-start relay to one or more targeted peers fails. `failures[]` enumerates each failed peer with a `reason` (`peer_rejected` / `peer_awaiting_approval` / `peer_transient_failure` / `livekit_unavailable`). `terminal: true` means the local ring was destroyed; `false` means the call continues for reachable recipients. | caller only | ### Social | type | fields | scope | diff --git a/packages/server/src/utils/federationOutbox.ts b/packages/server/src/utils/federationOutbox.ts index 9fe2bec2..36bbb37c 100644 --- a/packages/server/src/utils/federationOutbox.ts +++ b/packages/server/src/utils/federationOutbox.ts @@ -6,6 +6,7 @@ import crypto from 'node:crypto'; import type { FederationRelayEvent, FederationRelayParticipant, FederationRelayAttachment, DmMessageWithUser, FederationRelayRequest } from '@backspace/shared'; import { getOurOrigin, buildFederationHeaders, generateHmacSecret } from './federationAuth.js'; import { extractDomain } from '../routes/federation.js'; +import { racePeering, ensurePeered } from './federationPeering.js'; // ─── Settings Cache ────────────────────────────────────────────────────────── @@ -469,28 +470,87 @@ export function buildRelayPayload( }; } +/** 3s budget for the on-demand handshake before a call relay POST. */ +export const CALL_PEERING_TIMEOUT_MS = 3_000; + +export type CallRelayFailureReason = + | 'peer_rejected' + | 'peer_awaiting_approval' + | 'peer_transient_failure' + | 'post_failed'; + +export type CallRelayResult = + | { ok: true } + | { ok: false; reason: CallRelayFailureReason; error: string }; + /** * Send call signaling events directly to a remote peer (bypasses outbox). - * Used for time-critical call events where latency matters. - * If the HTTP POST fails, the call operation fails — no retry. + * Latency-sensitive: if no active peer exists, race an ensurePeered handshake + * against `opts.peeringTimeoutMs` (default CALL_PEERING_TIMEOUT_MS). + * + * `peeringTimeoutMs: 0` = non-blocking mode (used by sendTypingRelay): + * - If the peer is currently active, POST. Otherwise skip the POST, kick off + * ensurePeered() in the background as a warm-up, and return + * { ok:false, reason:'peer_transient_failure' }. */ export async function sendCallRelay( targetPeerOrigin: string, events: FederationRelayEvent[], -): Promise<{ ok: boolean; error?: string }> { + opts: { peeringTimeoutMs?: number } = {}, +): Promise { + const timeoutMs = opts.peeringTimeoutMs ?? CALL_PEERING_TIMEOUT_MS; const db = getDb(); - const peer = db.select() + + // ─── Fast path: peer already active or unreachable (health check handles) ── + const existing = db.select() .from(schema.federationPeers) - .where(and( - eq(schema.federationPeers.origin, targetPeerOrigin), - eq(schema.federationPeers.status, 'active'), - )) + .where(eq(schema.federationPeers.origin, targetPeerOrigin)) .get(); + let peer = existing && (existing.status === 'active' || existing.status === 'unreachable') + ? existing + : null; + if (!peer) { - return { ok: false, error: `No active peer for origin ${targetPeerOrigin}` }; + // ─── Non-blocking mode (typing): warm up in background, do not POST ── + if (timeoutMs === 0) { + ensurePeered(targetPeerOrigin).catch(err => { + console.warn('[federation] typing-triggered background handshake:', targetPeerOrigin, err); + }); + return { ok: false, reason: 'peer_transient_failure', error: 'peer not active' }; + } + + // ─── Race ensurePeered against the deadline ── + const raced = await racePeering(targetPeerOrigin, timeoutMs); + + switch (raced.status) { + case 'active': + // Re-fetch the now-active peer row for HMAC secret. + peer = db.select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, targetPeerOrigin)) + .get() ?? null; + if (!peer) { + return { ok: false, reason: 'peer_transient_failure', error: 'peer row missing after handshake' }; + } + break; + case 'rejected': + return { ok: false, reason: 'peer_rejected', error: raced.error }; + case 'pending': + return { ok: false, reason: 'peer_awaiting_approval', error: raced.error }; + case 'failed': + return { ok: false, reason: 'peer_transient_failure', error: raced.error }; + case 'timeout': + return { ok: false, reason: 'peer_transient_failure', error: `Peering handshake did not complete within ${(timeoutMs / 1000).toFixed(1)}s` }; + default: { + // Exhaustiveness check — catches any future additions to EnsurePeeredResult. + const _exhaustive: never = raced; + return { ok: false, reason: 'peer_transient_failure', error: `unexpected peering result: ${JSON.stringify(_exhaustive)}` }; + } + } } + // ─── POST ────────────────────────────────────────────────────────────────── const ourOrigin = getOurOrigin(); const body: FederationRelayRequest = { version: 1, @@ -499,7 +559,6 @@ export async function sendCallRelay( }; const bodyStr = JSON.stringify(body); - // Use pending secret during rotation (FED-011), otherwise current const signingSecret = peer.pendingHmacSecret ?? peer.hmacSecret; const headers = buildFederationHeaders(bodyStr, signingSecret, ourOrigin); @@ -511,14 +570,27 @@ export async function sendCallRelay( signal: AbortSignal.timeout(10_000), }); - if (!res.ok) { - const text = await res.text().catch(() => ''); - return { ok: false, error: `HTTP ${res.status}: ${text}` }; - } + if (res.ok) return { ok: true }; - return { ok: true }; + const text = await res.text().catch(() => ''); + if (res.status >= 400 && res.status < 500) { + return { + ok: false, + reason: 'post_failed', + error: `HTTP ${res.status}: ${text}`, + }; + } + return { + ok: false, + reason: 'peer_transient_failure', + error: `HTTP ${res.status}: ${text}`, + }; } catch (err) { - return { ok: false, error: err instanceof Error ? err.message : 'fetch_failed' }; + return { + ok: false, + reason: 'peer_transient_failure', + error: err instanceof Error ? err.message : 'fetch_failed', + }; } } @@ -700,10 +772,16 @@ export async function sendTypingRelay( }, }; - // Fire-and-forget to each remote peer + // Fire-and-forget to each remote peer; 0ms peering timeout = non-blocking warm-up. for (const peerOrigin of remoteOrigins) { - sendCallRelay(peerOrigin, [event]).catch(err => { - console.warn(`[federation] Typing relay to ${peerOrigin} failed:`, err); - }); + sendCallRelay(peerOrigin, [event], { peeringTimeoutMs: 0 }) + .then(result => { + if (!result.ok) { + console.debug(`[federation] Typing relay to ${peerOrigin}: ${result.reason} ${result.error}`); + } + }) + .catch(err => { + console.warn(`[federation] Typing relay to ${peerOrigin} threw unexpectedly:`, err); + }); } } diff --git a/packages/server/src/utils/federationPeering.test.ts b/packages/server/src/utils/federationPeering.test.ts index e25d4495..3961dc0b 100644 --- a/packages/server/src/utils/federationPeering.test.ts +++ b/packages/server/src/utils/federationPeering.test.ts @@ -1,5 +1,6 @@ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import type { EnsurePeeredResult } from './federationPeering.js'; +import { racePeering } from './federationPeering.js'; describe('EnsurePeeredResult type', () => { it('active result has peerId', () => { @@ -43,3 +44,71 @@ describe('EnsurePeeredResult type', () => { } }); }); + +describe('racePeering', () => { + it('returns the ensurePeered result when it resolves before the timeout', async () => { + const stub = vi.fn(async (): Promise => ({ + status: 'active', + peerId: 'peer-1', + })); + const result = await racePeering('https://example.com', 1_000, stub); + expect(result).toEqual({ status: 'active', peerId: 'peer-1' }); + expect(stub).toHaveBeenCalledWith('https://example.com'); + }); + + it('returns timeout when ensurePeered takes longer than the deadline', async () => { + vi.useFakeTimers(); + const stub = vi.fn((): Promise => new Promise(() => { + // Never resolves — simulates a slow handshake. + })); + const racePromise = racePeering('https://example.com', 50, stub); + await vi.advanceTimersByTimeAsync(50); + const result = await racePromise; + expect(result).toEqual({ status: 'timeout' }); + vi.useRealTimers(); + }); + + it('returns rejected result verbatim when ensurePeered resolves with rejection', async () => { + const stub = vi.fn(async (): Promise => ({ + status: 'rejected', + error: 'peer denied', + })); + const result = await racePeering('https://example.com', 1_000, stub); + expect(result).toEqual({ status: 'rejected', error: 'peer denied' }); + }); + + it('attaches a warn-logged catch to the background handshake when the timeout wins', async () => { + vi.useFakeTimers(); + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const stub = vi.fn(() => new Promise((_, reject) => { + setTimeout(() => reject(new Error('late failure')), 30); + })); + const racePromise = racePeering('https://example.com', 10, stub); + await vi.advanceTimersByTimeAsync(10); + const result = await racePromise; + expect(result).toEqual({ status: 'timeout' }); + await vi.advanceTimersByTimeAsync(30); + // Let microtasks flush so the .catch handler runs. + await Promise.resolve(); + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining('background handshake'), + 'https://example.com', + expect.any(Error), + ); + vi.useRealTimers(); + warnSpy.mockRestore(); + }); + + it('normalizes a thrown handshake error into { status: failed } without emitting the background warn', async () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); + const stub = vi.fn(async (): Promise => { + throw new Error('immediate handshake failure'); + }); + const result = await racePeering('https://example.com', 1_000, stub); + expect(result).toEqual({ status: 'failed', error: 'immediate handshake failure' }); + // The handshake rejection was the race winner — no background warn should fire. + await Promise.resolve(); + expect(warnSpy).not.toHaveBeenCalled(); + warnSpy.mockRestore(); + }); +}); diff --git a/packages/server/src/utils/federationPeering.ts b/packages/server/src/utils/federationPeering.ts index 2bf49fcd..2b582413 100644 --- a/packages/server/src/utils/federationPeering.ts +++ b/packages/server/src/utils/federationPeering.ts @@ -210,3 +210,46 @@ async function performHandshake( export function _clearInFlightPeering(): void { inFlightPeering.clear(); } + +/** + * Race ensurePeered() against a deadline. On timeout, the background + * handshake is NOT aborted — it continues so the next attempt finds + * the peer active. A warn-logged catch is attached so a late-rejecting + * background promise does not emit an unhandledRejection. + * + * The ensurePeered implementation is injectable for testing; the default + * is the real function. + */ +export async function racePeering( + origin: string, + timeoutMs: number, + ensurePeeredFn: (origin: string) => Promise = ensurePeered, +): Promise { + const handshake = ensurePeeredFn(origin); + + let timeoutHandle: ReturnType | undefined; + const timeoutPromise = new Promise<{ status: 'timeout' }>(resolve => { + timeoutHandle = setTimeout(() => resolve({ status: 'timeout' }), timeoutMs); + }); + + let raceResult: EnsurePeeredResult | { status: 'timeout' }; + try { + raceResult = await Promise.race([handshake, timeoutPromise]); + } catch (err) { + // ensurePeeredFn rejected as the race winner. Normalize to failed. + if (timeoutHandle) clearTimeout(timeoutHandle); + const message = err instanceof Error ? err.message : 'Unknown handshake error'; + return { status: 'failed', error: message }; + } + if (timeoutHandle) clearTimeout(timeoutHandle); + + // Only when the timeout arm won is the background handshake still running. + // Guard its eventual rejection so we don't emit unhandledRejection. + if (raceResult.status === 'timeout') { + handshake.catch(err => { + console.warn('[federation] background handshake after call-relay race:', origin, err); + }); + } + + return raceResult; +} diff --git a/packages/server/src/ws/events.ts b/packages/server/src/ws/events.ts index ab8a7cba..393ae9bf 100644 --- a/packages/server/src/ws/events.ts +++ b/packages/server/src/ws/events.ts @@ -6,7 +6,8 @@ import { connectionManager } from './handler.js'; import type { VoiceRoom, DmRoomMeta, SpaceRoomMeta } from './handler.js'; import { isMember, getChannelSpaceId, isDmMember, hasPermission, computePermissions, PermissionBits } from '../utils/permissions.js'; import { broadcastDmMessage, getDmMessageWithUser } from '../routes/dm.js'; -import { MAX_MESSAGE_LENGTH, type MessageWithUser, type Attachment, type DmMessageWithUser, type Embed, type Activity, type ActivityType, type ActivityTimestamps, type ActivityAssets, type ServerEvent } from '@backspace/shared'; +import { MAX_MESSAGE_LENGTH, type MessageWithUser, type Attachment, type DmMessageWithUser, type Embed, type Activity, type ActivityType, type ActivityTimestamps, type ActivityAssets, type ServerEvent, type DmCallUndeliverableFailure, type DmCallUndeliverableReason } from '@backspace/shared'; +import type { CallRelayFailureReason } from '../utils/federationOutbox.js'; import { ACTIVITY_LIMITS } from '@backspace/shared/src/activities.js'; import { sanitizeUser } from '../utils/sanitize.js'; import { deleteAttachmentFiles } from '../utils/fileCleanup.js'; @@ -1692,7 +1693,8 @@ function handleDmCallEnd(event: Record, userId: string): void { /** * Send S2S dm_call_start to all remote instances with DM members. - * Fire-and-forget — if delivery fails, the call still works locally. + * Fire-and-forget per relay, but aggregates per-peer results to surface + * undeliverable calls via `dm_call_undeliverable` to the caller. */ async function sendFederatedCallStart( dmChannelId: string, @@ -1746,9 +1748,33 @@ async function sendFederatedCallStart( .run(); } - // Check LiveKit configuration - if (!config.livekit.apiKey || !config.livekit.apiSecret) { + // Classify members relative to this instance. + const remoteMembers = members.filter(m => { + if (!m.homeInstance) return false; + const normalized = m.homeInstance.startsWith('http') ? m.homeInstance : `https://${m.homeInstance}`; + return normalized !== ourOrigin; + }); + const localNonCallerMembers = members.filter(m => { + if (m.userId === callerId) return false; + const home = m.homeInstance + ? (m.homeInstance.startsWith('http') ? m.homeInstance : `https://${m.homeInstance}`) + : ourOrigin; + return home === ourOrigin; + }); + const hasConnectedLocalRingee = localNonCallerMembers.some(m => + connectionManager.isUserOnline(m.userId), + ); + + // ─── LiveKit pre-flight ──────────────────────────────────────────────────── + if ((!config.livekit.apiKey || !config.livekit.apiSecret) && remoteMembers.length > 0) { console.warn('[federation] Cannot start federated call: LiveKit not configured'); + emitUndeliverableAndMaybeDestroy({ + callerId, + dmChannelId, + federatedId, + terminal: !hasConnectedLocalRingee, + failures: [{ reason: 'livekit_unavailable' }], + }); return; } @@ -1793,55 +1819,133 @@ async function sendFederatedCallStart( }, }); - // Identify remote members for targeted relay - const remoteMembers = members.filter(m => { - if (!m.homeInstance) return false; - const normalized = m.homeInstance.startsWith('http') ? m.homeInstance : `https://${m.homeInstance}`; - return normalized !== ourOrigin; - }); - // Group remote members by home instance (targeted peers) - const targetedPeers = new Set(); + const targetedPeers = new Map(); for (const m of remoteMembers) { const origin = m.homeInstance!.startsWith('http') ? m.homeInstance! : `https://${m.homeInstance!}`; - targetedPeers.add(origin); + const bucket = targetedPeers.get(origin) ?? []; + bucket.push(m.userId); + targetedPeers.set(origin, bucket); } - // Query ALL active federation peers for broadcast - const allPeers = db.select({ origin: schema.federationPeers.origin }) + // Single query for all peers — derives both active-peer list and label map + const peerRows = db.select({ + origin: schema.federationPeers.origin, + instanceName: schema.federationPeers.instanceName, + status: schema.federationPeers.status, + }) .from(schema.federationPeers) - .where(eq(schema.federationPeers.status, 'active')) .all(); - // Build parallel relay promises - const relayPromises: Promise[] = []; + const allPeers = peerRows.filter(r => r.status === 'active'); - // Targeted relay: peers with known remote DM members - for (const peerOrigin of targetedPeers) { - relayPromises.push( - sendCallRelay(peerOrigin, [buildRelayEvent()]).then(result => { - if (!result.ok) { - console.error(`[federation] Failed to send dm_call_start to ${peerOrigin}: ${result.error}`); - } - }) - ); + const peerLabelByOrigin = new Map(); + for (const row of peerRows) { + if (row.instanceName) peerLabelByOrigin.set(row.origin, row.instanceName); } - // All-peers broadcast: every other active peer + // ─── Targeted relay: fan out in parallel, await results ──────────────────── + const targetedResults = await Promise.all( + Array.from(targetedPeers.keys()).map(async peerOrigin => { + const result = await sendCallRelay(peerOrigin, [buildRelayEvent()]); + if (result.ok) { + return { origin: peerOrigin, ok: true as const }; + } + const reason = mapCallReasonToEventReason(result.reason); + console.error(`[federation] dm_call_start to ${peerOrigin} failed (${result.reason}): ${result.error}`); + return { origin: peerOrigin, ok: false as const, reason, error: result.error }; + }), + ); + + // ─── All-peers broadcast: fire-and-forget; failures NOT surfaced ─────────── for (const peer of allPeers) { if (targetedPeers.has(peer.origin)) continue; if (peer.origin === ourOrigin) continue; - relayPromises.push( - sendCallRelay(peer.origin, [buildRelayEvent()]).then(result => { - if (!result.ok) { - console.debug(`[federation] All-peers dm_call_start to ${peer.origin}: ${result.error || 'failed'}`); - } - }) - ); + sendCallRelay(peer.origin, [buildRelayEvent()]).then(result => { + if (!result.ok) { + console.debug(`[federation] All-peers dm_call_start to ${peer.origin}: ${result.reason} ${result.error}`); + } + }).catch(err => console.warn('[federation] all-peers broadcast threw:', err)); } - // Fire all relays in parallel — each has its own 10s timeout - await Promise.all(relayPromises); + // ─── Aggregate failures → dm_call_undeliverable ─────────────────────────── + const failedTargeted = targetedResults.filter( + (r): r is Extract => !r.ok, + ); + if (failedTargeted.length === 0) return; + + const anyTargetedSuccess = targetedResults.some(r => r.ok); + const plausibleRecipientRemains = anyTargetedSuccess || hasConnectedLocalRingee; + + const failures: DmCallUndeliverableFailure[] = failedTargeted.map(r => { + const affectedUserIds = targetedPeers.get(r.origin) ?? []; + return { + reason: r.reason, + peerOrigin: r.origin, + peerLabel: peerLabelByOrigin.get(r.origin), + affectedUserIds, + }; + }); + + emitUndeliverableAndMaybeDestroy({ + callerId, + dmChannelId, + federatedId, + terminal: !plausibleRecipientRemains, + failures, + }); +} + +/** Map a sendCallRelay reason to the event-surface reason. */ +function mapCallReasonToEventReason(reason: CallRelayFailureReason): DmCallUndeliverableReason { + switch (reason) { + case 'peer_rejected': return 'peer_rejected'; + case 'peer_awaiting_approval': return 'peer_awaiting_approval'; + case 'peer_transient_failure': return 'peer_transient_failure'; + case 'post_failed': return 'peer_transient_failure'; // 4xx looks transient to users + } +} + +/** + * Emit dm_call_undeliverable to the caller. If terminal, also destroy the + * local ring room (which clears the ringing timer and voice WS binding) + * and broadcast dm_call_ended to non-caller DM members, mirroring the + * 60s auto-timeout's cleanup semantics (handler.ts:414-424) so any + * ringing client (e.g., a Connection WS from another instance) exits + * the ring state instead of hanging. + * + * Guard: if the caller already cancelled mid-race, the room is already + * gone — do NOT emit a phantom "could not reach" toast. + */ +function emitUndeliverableAndMaybeDestroy(args: { + callerId: string; + dmChannelId: string; + federatedId: string; + terminal: boolean; + failures: DmCallUndeliverableFailure[]; +}): void { + const { callerId, dmChannelId, federatedId, terminal, failures } = args; + + // Caller may have cancelled mid-race. If the room is gone, move on silently. + const room = connectionManager.getRoom(dmChannelId); + if (!room) return; + + if (terminal) { + connectionManager.clearVoiceWs(callerId); + connectionManager.destroyRoom(dmChannelId); + connectionManager.sendToDmMembers(dmChannelId, { + type: 'dm_call_ended', + dmChannelId, + }, callerId); + } + + connectionManager.sendToUser(callerId, { + type: 'dm_call_undeliverable', + dmChannelId, + federatedCallId: federatedId, + terminal, + failures, + }); } async function sendFederatedCallAccept(dmChannelId: string, acceptorUserId: string): Promise { diff --git a/packages/shared/src/types.ts b/packages/shared/src/types.ts index 24b452e9..0d8c123a 100644 --- a/packages/shared/src/types.ts +++ b/packages/shared/src/types.ts @@ -358,6 +358,19 @@ export interface Activity { // ─── WebSocket Event Types ────────────────────────────────────────────────── +export type DmCallUndeliverableReason = + | 'peer_rejected' + | 'peer_awaiting_approval' + | 'peer_transient_failure' + | 'livekit_unavailable'; + +export interface DmCallUndeliverableFailure { + reason: DmCallUndeliverableReason; + peerOrigin?: string; + peerLabel?: string; + affectedUserIds?: string[]; +} + // Client → Server Events export type ClientEvent = | { type: 'auth'; token: string } @@ -413,6 +426,7 @@ export type ServerEvent = | { type: 'dm_call_accepted'; dmChannelId: string | null; federatedCallId?: string } | { type: 'dm_call_rejected'; dmChannelId: string } | { type: 'dm_call_ended'; dmChannelId: string } + | { type: 'dm_call_undeliverable'; dmChannelId: string | null; federatedCallId: string; terminal: boolean; failures: DmCallUndeliverableFailure[] } | { type: 'voice_status_update'; userId: string; channelId: string; isMuted: boolean; isDeafened: boolean; isCameraOn: boolean; isScreenSharing: boolean } | { type: 'dm_channel_created'; dmChannel: DmChannel } | { type: 'dm_channel_closed'; dmChannelId: string } diff --git a/packages/web/src/hooks/useWebSocket.ts b/packages/web/src/hooks/useWebSocket.ts index 9ee59274..5cd33c53 100644 --- a/packages/web/src/hooks/useWebSocket.ts +++ b/packages/web/src/hooks/useWebSocket.ts @@ -114,6 +114,43 @@ function buildWsUrl(origin: string): string { return `${protocol}//${url.host}/ws`; } +// ─── Call relay helpers ─────────────────────────────────────────────────────── + +function buildCallUndeliverableToast( + failures: Array<{ reason: string; peerOrigin?: string; peerLabel?: string }>, + terminal: boolean, +): string { + const primary = failures[0]; + const labelFor = (f: { peerLabel?: string; peerOrigin?: string }) => + f.peerLabel ?? f.peerOrigin?.replace(/^https?:\/\//, '') ?? 'the remote instance'; + + if (!terminal) { + const labels = failures.map(labelFor).join(', '); + return `Some participants could not be reached: ${labels}.`; + } + + if (failures.length > 1) { + const labels = failures.map(labelFor).join(', '); + return `Could not reach ${failures.length} instances: ${labels}.`; + } + + if (!primary) return 'Call could not be placed.'; + + const label = labelFor(primary); + switch (primary.reason) { + case 'peer_rejected': + return `Cannot reach ${label} — this instance requires manual peering approval.`; + case 'peer_awaiting_approval': + return `Waiting for ${label} admin to approve your instance. Calls will work once approved.`; + case 'peer_transient_failure': + return `Could not reach ${label}. Try again in a moment.`; + case 'livekit_unavailable': + return 'Voice is not configured on this instance.'; + default: + return `Call to ${label} could not be placed.`; + } +} + // ─── Event handling ─────────────────────────────────────────────────────────── const HOME_ORIGIN = ''; @@ -954,6 +991,26 @@ function handleEvent(origin: string, event: ServerEvent): void { break; } + case 'dm_call_undeliverable': { + if (!isHome && !activePeerOrigins.has(origin)) break; + + const { setIncomingCall, setOutgoingCall, setActiveDmCall, disconnectFn, clearFederatedCallData } = useVoiceStore.getState(); + const { addToast } = useUIStore.getState(); + + if (event.terminal) { + // Tear down local outbound call state — mirrors dm_call_ended. + setIncomingCall(null); + setOutgoingCall(null); + setActiveDmCall(null); + clearFederatedCallData(); + if (disconnectFn) disconnectFn(); + } + + const msg = buildCallUndeliverableToast(event.failures, event.terminal); + addToast(msg, event.terminal ? 'warning' : 'info', 8_000); + break; + } + // ─── DM channel events (all origins) ──────────────────────────────────── case 'dm_channel_created': {