diff --git a/docs/systems/dm-system.md b/docs/systems/dm-system.md index 245bf385..710d015b 100644 --- a/docs/systems/dm-system.md +++ b/docs/systems/dm-system.md @@ -283,7 +283,7 @@ If a `member_add` federation event arrives for a soft-deleted channel (non-null **Request:** `{ content?: string, attachments?: string[], replyToId?: string }` -**Cross-instance access:** Federated users (those with `homeInstance` set) can send messages on any DM channel where they are a member, regardless of which instance serves the request. The `requireLocalUser` gate that previously blocked federated users from DM write endpoints has been removed. DM calls work across federated instances. The caller's instance hosts the LiveKit room; remote clients connect directly. Call signaling is relayed to all active federation peers via synchronous HTTP POST (not the outbox worker). See `docs/systems/voice.md` for the full federated call architecture. +**Cross-instance access:** Federated users (those with `homeInstance` set) can send messages on any DM channel where they are a member, regardless of which instance serves the request. The `requireLocalUser` gate that previously blocked federated users from DM write endpoints has been removed. DM calls work across federated instances. The caller's instance hosts the LiveKit room; remote clients connect directly. Call signaling is relayed to all active federation peers via synchronous HTTP POST (not the outbox worker). Relay failures at any call state transition emit `dm_call_undeliverable { phase, terminal, failures }` to the originator — see `docs/systems/voice.md` for the full call state machine and failure surface. **Validation:** - Caller must be a member (`isDmMember`) diff --git a/docs/systems/federation.md b/docs/systems/federation.md index b845672d..8b7b20f5 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -1134,7 +1134,11 @@ All events carry standard relay fields: `eventType`, `messageId`, `encryptionVer - 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-start failure surfacing.** `sendFederatedCallStart` aggregates targeted-peer results and emits `dm_call_undeliverable { phase: 'start' }` to the caller for failed targeted peers. See `docs/systems/voice.md` and `docs/systems/websocket.md` for the event contract. + +**Accept / reject / end relay discipline.** Every `handleDmCall{Accept,Reject,End}` Path-2 branch awaits `sendCallRelay` and emits `dm_call_undeliverable { phase, terminal, failures }` to the originator on failure. Accept is pessimistic-rollback (terminal: true — local `FederatedCallEntry` cleared, optimistic `dm_call_accepted` walked back via `sendToFederatedCallUsers`); reject and end are optimistic (terminal: false — state already cleared, informational toast only). Path-1 fan-outs via `sendFederatedCallAccept` / `sendFederatedCallEnd` / `fanOutCallEvent` return `CallFanoutFailure[]` and the host-side caller receives `dm_call_undeliverable { terminal: false }` listing peers that were not reached. + +**Ring-timeout fan-out.** `ConnectionManager.createDmRoom`'s 60 s ringing auto-clean now invokes a registered hook (`setRingTimeoutFanoutHook`, registered from `ws/events.ts:registerCallRelayHooks`) that fans `dm_call_end` out to remote peers, so stranded Path-A/B ringees on other instances exit their ring state instead of lingering until their own 60 s cleanup fires. ### Call Flows @@ -1269,6 +1273,4 @@ If the `federation_mutation_log` table exists but is empty, populates it with `c 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. +- **Accept/reject/end relay failures now surfaced.** All three federated call-state transitions emit `dm_call_undeliverable { phase, terminal, failures }` to the originator on relay failure — accept rolls back optimistic state (terminal: true), reject/end keep the optimistic clear and emit an informational toast (terminal: false). See `docs/systems/voice.md` "Call relay failure surface" for the full contract. The host-side ring timeout also fans `dm_call_end` out to peers so stranded Path-A/B ringees exit the ring. One remaining edge documented in voice.md: non-host end-relay failure leaves the host's local `activeDmCall` marker until manual cleanup. diff --git a/docs/systems/voice.md b/docs/systems/voice.md index 155171b0..353a6c76 100644 --- a/docs/systems/voice.md +++ b/docs/systems/voice.md @@ -60,9 +60,24 @@ All `dm_call_*` signaling events (`start`, `accept`, `reject`, `end`) are relaye **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. +**Call relay failure surface.** Every `dm_call_{start,accept,reject,end}` relay is failure-aware. On failure the originating server emits a `dm_call_undeliverable` event with a `phase` discriminator identifying which action failed. Client copy is phase-specific; state rollback depends on the phase. -Accept and end relay failures are NOT surfaced today; see the federation doc's "Known issues" section for the deferred accept-failure dead-end. +| `phase` | `terminal` | Emitted when | Client action | +|---------|------------|--------------|---------------| +| `start` | true | No plausible recipient after targeted-peer fan-out; ring room destroyed. | Clear `outgoingCall`, disconnect LK, warning toast. | +| `start` | false | Some targeted peers failed but reachable recipients remain; ring continues. | Keep state; info toast. | +| `accept` | true | Acceptor's B→host relay failed; optimistic state is rolled back on B. | Clear `activeDmCall` + `incomingCall`, disconnect LK, warning toast. | +| `accept` | false | Host → peer fan-out of accept failed; local host call continues. | No state change; info toast. | +| `reject` | false | Rejector's relay to host failed OR host's fan-out after a local reject failed; state already cleared. | No state change; info toast. | +| `end` | false | Ender's relay to host failed OR host's fan-out after a local end failed; state already cleared. | No state change; info toast. | + +**Accept-rollback semantics.** `handleDmCallAccept` Path 2 transitions the `FederatedCallEntry` to active and broadcasts `dm_call_accepted` optimistically so the acceptor's UI flips immediately. If the B→host relay fails, the server clears the entry, fans `dm_call_undeliverable { phase: 'accept', terminal: true }` out to all ringed users on B (via `sendToFederatedCallUsers`), and the client tears its call state back down. + +**Reject / end are optimistic.** Local state is cleared before the relay is awaited because the user's intent is to terminate. If the relay fails, the originator receives an informational `dm_call_undeliverable { terminal: false }` so they know remote peers may briefly display stale state; no local rollback. + +**Ring-timeout fan-out.** When the host's 60 s ringing timeout fires without an accept, `dm_call_end` is fanned out to all remote peers so stranded Path-A/B ringees on other instances exit their ring state instead of lingering. Registered via `connectionManager.setRingTimeoutFanoutHook` from the WS events module. + +**Remaining edge.** When a non-host participant (Bob on B) ends an active call and the relay to the host (Alice on A) fails, Alice's `activeDmCall` marker lingers until she manually ends — LK `ParticipantDisconnected` tears down her voice UI but does not clear the DM-call marker. This is a host-side cleanup concern, tracked separately. ### Dual-Path Processing diff --git a/docs/systems/websocket.md b/docs/systems/websocket.md index 9f160208..0d40582f 100644 --- a/docs/systems/websocket.md +++ b/docs/systems/websocket.md @@ -170,7 +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 | +| `dm_call_undeliverable` | Sent to the originator when a call relay (start / accept / reject / end) to one or more peers fails. Includes `phase: 'start' \| 'accept' \| 'reject' \| 'end'` identifying the action; `failures[]` enumerates failed peers with a `reason` (`peer_rejected` / `peer_awaiting_approval` / `peer_transient_failure` / `livekit_unavailable`). `terminal: true` means local call state should be (or has been) torn down; `terminal: false` is informational. See `docs/systems/voice.md` for the full phase × terminal matrix. | originator (caller / acceptor / rejector / ender) | ### Social | type | fields | scope | diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 5ea35097..511c3c41 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -26,6 +26,7 @@ import { adminRoutes } from './routes/admin.js'; import { gifRoutes } from './routes/gif.js'; import { federationRoutes } from './routes/federation.js'; import { startFederationWorkers, stopFederationWorkers } from './utils/federationWorker.js'; +import { registerCallRelayHooks } from './ws/events.js'; import { registerWebSocket } from './ws/handler.js'; import path from 'path'; @@ -124,6 +125,9 @@ async function main(): Promise { // Log ffmpeg availability at startup (so admins see the warning immediately) checkFfmpeg(); + // Register WS-layer call relay hooks (ring-timeout fan-out). + registerCallRelayHooks(); + // Start federation background workers (outbox delivery, file download, health check) startFederationWorkers(); diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts index 9d41f8d6..a321e38a 100644 --- a/packages/server/src/routes/federation.ts +++ b/packages/server/src/routes/federation.ts @@ -12,6 +12,8 @@ import { getDb, getRawDb, schema } from '../db/index.js'; import { config } from '../config.js'; import { connectionManager } from '../ws/handler.js'; import type { FederatedCallEntry, DmRoomMeta } from '../ws/handler.js'; +import { mapCallReasonToEventReason, type CallFanoutFailure } from '../utils/federationOutbox.js'; +import type { DmCallUndeliverableFailure } from '@backspace/shared'; import { sanitizeUser } from '../utils/sanitize.js'; import { deleteAttachmentFiles, deleteUploadFile } from '../utils/fileCleanup.js'; import { tombstoneUser, collectDeletionBroadcastTargets, collectProfileBroadcastTargetIds } from '../utils/userDeletion.js'; @@ -4464,10 +4466,14 @@ function processDmCallAcceptEvent( // Fan out to ALL other remote instances (exclude the one that sent the accept) const normalizedSource = sourceInstance.startsWith('http') ? sourceInstance : `https://${sourceInstance}`; - fanOutCallEvent(dmChannelId!, event.federatedId, 'dm_call_accept', { + const hostCallerId = (room.metadata as DmRoomMeta).callerId; + const localDmId = dmChannelId!; + void fanOutCallEvent(localDmId, event.federatedId, 'dm_call_accept', { call: { acceptor: event.call.acceptor }, - }, normalizedSource, db).catch(err => - console.error('[federation] Fan-out dm_call_accept failed:', err) + }, normalizedSource, db).then(failures => { + emitHostFanoutUndeliverable(hostCallerId, localDmId, event.federatedId!, 'accept', failures); + }).catch(err => + console.error('[federation] Fan-out dm_call_accept threw:', err), ); } else { // We're a REMOTE instance receiving fan-out — transition local state @@ -4517,19 +4523,23 @@ function processDmCallRejectEvent( const room = dmChannelId ? connectionManager.getRoom(dmChannelId) : undefined; if (room && room.roomType === 'dm') { const meta = room.metadata as DmRoomMeta; + const hostCallerId = meta.callerId; + const localDmId = dmChannelId!; connectionManager.clearVoiceWs(meta.callerId); - connectionManager.destroyRoom(dmChannelId!); + connectionManager.destroyRoom(localDmId); - connectionManager.sendToDmMembers(dmChannelId!, { + connectionManager.sendToDmMembers(localDmId, { type: 'dm_call_rejected', - dmChannelId: dmChannelId!, + dmChannelId: localDmId, }); const normalizedSource = sourceInstance.startsWith('http') ? sourceInstance : `https://${sourceInstance}`; - fanOutCallEvent(dmChannelId!, event.federatedId, 'dm_call_end', { + void fanOutCallEvent(localDmId, event.federatedId, 'dm_call_end', { call: { endedBy: event.call.rejector }, - }, normalizedSource, db).catch(err => - console.error('[federation] Fan-out dm_call_end (reject) failed:', err) + }, normalizedSource, db).then(failures => { + emitHostFanoutUndeliverable(hostCallerId, localDmId, event.federatedId!, 'reject', failures); + }).catch(err => + console.error('[federation] Fan-out dm_call_end (reject) threw:', err), ); } else { const fedCall = connectionManager.getFederatedCall(event.federatedId); @@ -4572,23 +4582,27 @@ function processDmCallEndEvent( const room = dmChannelId ? connectionManager.getRoom(dmChannelId) : undefined; if (room && room.roomType === 'dm') { const meta = room.metadata as DmRoomMeta; + const hostCallerId = meta.callerId; + const localDmId = dmChannelId!; connectionManager.clearVoiceWs(meta.callerId); for (const pid of room.participants) { connectionManager.clearVoiceUserStatus(pid); connectionManager.clearVoiceWs(pid); } - connectionManager.destroyRoom(dmChannelId!); + connectionManager.destroyRoom(localDmId); - connectionManager.sendToDmMembers(dmChannelId!, { + connectionManager.sendToDmMembers(localDmId, { type: 'dm_call_ended', - dmChannelId: dmChannelId!, + dmChannelId: localDmId, }); const normalizedSource = sourceInstance.startsWith('http') ? sourceInstance : `https://${sourceInstance}`; - fanOutCallEvent(dmChannelId!, event.federatedId, 'dm_call_end', { + void fanOutCallEvent(localDmId, event.federatedId, 'dm_call_end', { call: { endedBy: event.call.endedBy }, - }, normalizedSource, db).catch(err => - console.error('[federation] Fan-out dm_call_end failed:', err) + }, normalizedSource, db).then(failures => { + emitHostFanoutUndeliverable(hostCallerId, localDmId, event.federatedId!, 'end', failures); + }).catch(err => + console.error('[federation] Fan-out dm_call_end threw:', err), ); } else { const fedCall = connectionManager.getFederatedCall(event.federatedId); @@ -5144,7 +5158,7 @@ async function fanOutCallEvent( extraFields: Partial, excludeOrigin: string | undefined, db: ReturnType, -): Promise { +): Promise { const members = db.select({ homeInstance: schema.users.homeInstance }) .from(schema.dmMembers) .innerJoin(schema.users, eq(schema.dmMembers.userId, schema.users.id)) @@ -5161,7 +5175,7 @@ async function fanOutCallEvent( } } } - if (targets.size === 0) return; + if (targets.size === 0) return []; const relayEvent: FederationRelayEvent = { eventType, @@ -5172,11 +5186,51 @@ async function fanOutCallEvent( ...extraFields, } as FederationRelayEvent; - await Promise.all( - Array.from(targets).map(origin => - sendCallRelay(origin, [relayEvent]).catch(err => - console.error(`[federation] Fan-out ${eventType} to ${origin} failed:`, err) - ) - ) + const labelByOrigin = new Map(); + for (const r of db.select({ origin: schema.federationPeers.origin, instanceName: schema.federationPeers.instanceName }) + .from(schema.federationPeers) + .all()) { + labelByOrigin.set(r.origin, r.instanceName ?? null); + } + + const results = await Promise.all( + Array.from(targets).map(async origin => ({ origin, result: await sendCallRelay(origin, [relayEvent]) })), ); + + const failures: CallFanoutFailure[] = []; + for (const { origin, result } of results) { + if (!result.ok) { + console.error(`[federation] Fan-out ${eventType} to ${origin} failed (${result.reason}): ${result.error}`); + failures.push({ + origin, + peerLabel: labelByOrigin.get(origin) ?? undefined, + reason: mapCallReasonToEventReason(result.reason), + }); + } + } + return failures; +} + +/** Emit a non-terminal dm_call_undeliverable for a host-side fan-out failure. */ +function emitHostFanoutUndeliverable( + userId: string, + dmChannelId: string, + federatedId: string, + phase: 'accept' | 'reject' | 'end', + fanoutFailures: CallFanoutFailure[], +): void { + if (fanoutFailures.length === 0) return; + const failures: DmCallUndeliverableFailure[] = fanoutFailures.map(f => ({ + reason: f.reason, + peerOrigin: f.origin, + peerLabel: f.peerLabel, + })); + connectionManager.sendToUser(userId, { + type: 'dm_call_undeliverable', + dmChannelId, + federatedCallId: federatedId, + terminal: false, + phase, + failures, + }); } diff --git a/packages/server/src/utils/federationOutbox.ts b/packages/server/src/utils/federationOutbox.ts index d73ee0df..57aad54d 100644 --- a/packages/server/src/utils/federationOutbox.ts +++ b/packages/server/src/utils/federationOutbox.ts @@ -3,7 +3,7 @@ import * as schema from '../db/schema.js'; import { eq, and, inArray } from 'drizzle-orm'; import { generateSnowflake } from './snowflake.js'; import crypto from 'node:crypto'; -import type { FederationRelayEvent, FederationRelayParticipant, FederationRelayAttachment, DmMessageWithUser, FederationRelayRequest } from '@backspace/shared'; +import type { FederationRelayEvent, FederationRelayParticipant, FederationRelayAttachment, DmMessageWithUser, FederationRelayRequest, DmCallUndeliverableReason } from '@backspace/shared'; import { getOurOrigin, buildFederationHeaders, generateHmacSecret } from './federationAuth.js'; import { extractDomain } from '../routes/federation.js'; import { racePeering, ensurePeered } from './federationPeering.js'; @@ -527,6 +527,23 @@ export type CallRelayResult = | { ok: true } | { ok: false; reason: CallRelayFailureReason; error: string }; +/** Per-peer failure record returned by Path-1 call fan-out helpers. */ +export interface CallFanoutFailure { + origin: string; + peerLabel?: string; + reason: DmCallUndeliverableReason; +} + +/** Map a sendCallRelay reason to the dm_call_undeliverable event-surface reason. */ +export 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 + } +} + /** * Send call signaling events directly to a remote peer (bypasses outbox). * Latency-sensitive: if no active peer exists, race an ensurePeered handshake diff --git a/packages/server/src/ws/events.dmCallRelay.test.ts b/packages/server/src/ws/events.dmCallRelay.test.ts new file mode 100644 index 00000000..26547070 --- /dev/null +++ b/packages/server/src/ws/events.dmCallRelay.test.ts @@ -0,0 +1,300 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import Database from 'better-sqlite3'; +import { drizzle } from 'drizzle-orm/better-sqlite3'; +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import * as schema from '../db/schema.js'; +import { setWorkerId } from '../utils/snowflake.js'; + +setWorkerId(1); + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +type TestDb = ReturnType>; + +let testDb: TestDb; + +vi.mock('../db/index.js', () => ({ + getDb: () => testDb, + schema, +})); + +vi.mock('../utils/federationAuth.js', () => ({ + getOurOrigin: () => 'https://local.example', + buildFederationHeaders: () => ({}), + generateHmacSecret: () => 'test-secret', +})); + +// Mock sendCallRelay so tests can control relay-result per case. +const sendCallRelayMock = vi.fn(); +vi.mock('../utils/federationOutbox.js', async () => { + const actual = await vi.importActual('../utils/federationOutbox.js'); + return { + ...actual, + sendCallRelay: (...args: unknown[]) => sendCallRelayMock(...args), + }; +}); + +function applyMigrations(db: Database.Database): void { + const migrationsDir = path.resolve(__dirname, '../../drizzle'); + const files = fs.readdirSync(migrationsDir).filter(f => f.endsWith('.sql')).sort(); + for (const f of files) { + const sql = fs.readFileSync(path.join(migrationsDir, f), 'utf8'); + const statements = sql.split(/-->\s*statement-breakpoint/); + for (const stmt of statements) { + const clean = stmt.trim(); + if (clean) db.exec(clean); + } + } +} + +function seedPeerLabel(origin: string, instanceName: string): void { + testDb.insert(schema.federationPeers).values({ + id: `peer-${origin}`, + origin, + hmacSecret: 'secret', + status: 'active', + instanceName, + lastSyncedAt: 0, + createdAt: Date.now(), + }).run(); +} + +function seedLocalUser(id: string, homeUserId: string | null): void { + testDb.insert(schema.users).values({ + id, + username: id, + passwordHash: 'test', + homeUserId, + homeInstance: null, + createdAt: Date.now(), + }).run(); +} + +async function importSUT() { + // Late import so the module picks up the mocked dependencies. + return await import('./events.js'); +} + +async function importManager() { + const mod = await import('./handler.js'); + return mod.connectionManager; +} + +type FedCallEntry = import('./handler.js').FederatedCallEntry; + +function makeFedCall(partial: Partial = {}): FedCallEntry { + return { + dmChannelId: 'dm-1', + federatedId: `fed-${Math.random().toString(36).slice(2, 10)}`, + callerId: 'caller-user', + callerHomeUserId: 'caller@pi', + federatedCallHost: 'https://pi.example', + livekitUrl: 'wss://pi.example/lk', + tokens: new Map([['acceptor@vm', 'token']]), + ringedUserIds: ['acceptor-user'], + state: 'ringing', + startedAt: Date.now(), + ...partial, + }; +} + +beforeEach(() => { + const sqlite = new Database(':memory:'); + testDb = drizzle(sqlite, { schema }); + applyMigrations(sqlite); + seedPeerLabel('https://pi.example', 'Pi-Instance'); + seedLocalUser('acceptor-user', 'acceptor@vm'); + seedLocalUser('caller-user', 'caller@pi'); + sendCallRelayMock.mockReset(); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('handleDmCallEnd Path-2 relay failure', () => { + it('emits dm_call_undeliverable { phase:"end", terminal:false } when the relay fails', async () => { + const { handleDmCallEndForTest } = await importSUT(); + const connectionManager = await importManager(); + + const fedCall = makeFedCall({ state: 'active' }); + connectionManager.createFederatedCall(fedCall); + const sendToUserSpy = vi.spyOn(connectionManager, 'sendToUser'); + sendCallRelayMock.mockResolvedValue({ ok: false, reason: 'peer_transient_failure', error: 'timeout' }); + + await handleDmCallEndForTest( + { federatedCallId: fedCall.federatedId }, + fedCall.ringedUserIds[0]!, + ); + + const undelivCalls = sendToUserSpy.mock.calls.filter(([, ev]) => + (ev as { type: string }).type === 'dm_call_undeliverable', + ); + expect(undelivCalls).toHaveLength(1); + const ev = undelivCalls[0]![1] as { phase: string; terminal: boolean }; + expect(ev.phase).toBe('end'); + expect(ev.terminal).toBe(false); + }); + + it('does not emit undeliverable on relay success', async () => { + const { handleDmCallEndForTest } = await importSUT(); + const connectionManager = await importManager(); + + const fedCall = makeFedCall({ state: 'active' }); + connectionManager.createFederatedCall(fedCall); + const sendToUserSpy = vi.spyOn(connectionManager, 'sendToUser'); + sendCallRelayMock.mockResolvedValue({ ok: true }); + + await handleDmCallEndForTest( + { federatedCallId: fedCall.federatedId }, + fedCall.ringedUserIds[0]!, + ); + + const undelivCalls = sendToUserSpy.mock.calls.filter(([, ev]) => + (ev as { type: string }).type === 'dm_call_undeliverable', + ); + expect(undelivCalls).toHaveLength(0); + }); +}); + +describe('handleDmCallReject Path-2 relay failure', () => { + it('emits dm_call_undeliverable { phase:"reject", terminal:false } to the rejector when the relay fails', async () => { + const { handleDmCallRejectForTest } = await importSUT(); + const connectionManager = await importManager(); + + const fedCall = makeFedCall(); + connectionManager.createFederatedCall(fedCall); + const sendToUserSpy = vi.spyOn(connectionManager, 'sendToUser'); + sendCallRelayMock.mockResolvedValue({ ok: false, reason: 'peer_rejected', error: 'rejected' }); + + await handleDmCallRejectForTest( + { federatedCallId: fedCall.federatedId }, + fedCall.ringedUserIds[0]!, + ); + + const undelivCalls = sendToUserSpy.mock.calls.filter(([, ev]) => + (ev as { type: string }).type === 'dm_call_undeliverable', + ); + expect(undelivCalls).toHaveLength(1); + const ev = undelivCalls[0]![1] as { phase: string; terminal: boolean }; + expect(ev.phase).toBe('reject'); + expect(ev.terminal).toBe(false); + // Local state was cleared before the relay even fired. + expect(connectionManager.getFederatedCall(fedCall.federatedId)).toBeUndefined(); + }); + + it('does not emit undeliverable on relay success', async () => { + const { handleDmCallRejectForTest } = await importSUT(); + const connectionManager = await importManager(); + + const fedCall = makeFedCall(); + connectionManager.createFederatedCall(fedCall); + const sendToUserSpy = vi.spyOn(connectionManager, 'sendToUser'); + sendCallRelayMock.mockResolvedValue({ ok: true }); + + await handleDmCallRejectForTest( + { federatedCallId: fedCall.federatedId }, + fedCall.ringedUserIds[0]!, + ); + + const undelivCalls = sendToUserSpy.mock.calls.filter(([, ev]) => + (ev as { type: string }).type === 'dm_call_undeliverable', + ); + expect(undelivCalls).toHaveLength(0); + }); +}); + +describe('ring-timeout fan-out hook', () => { + it('invokes the fan-out hook when the ring timer fires', async () => { + const connectionManager = await importManager(); + + const hook = vi.fn(async () => {}); + connectionManager.setRingTimeoutFanoutHook(hook); + + vi.useFakeTimers(); + try { + connectionManager.createDmRoom('dm-ringout-test', 'caller-ringout'); + vi.advanceTimersByTime(60_000 + 10); + } finally { + vi.useRealTimers(); + } + + expect(hook).toHaveBeenCalledWith('dm-ringout-test', 'caller-ringout'); + }); + + it('registerCallRelayHooks wires up the hook', async () => { + const connectionManager = await importManager(); + const { registerCallRelayHooks } = await importSUT(); + + const setSpy = vi.spyOn(connectionManager, 'setRingTimeoutFanoutHook'); + registerCallRelayHooks(); + expect(setSpy).toHaveBeenCalledTimes(1); + }); +}); + +describe('handleDmCallAccept Path-2 relay failure', () => { + it('emits dm_call_undeliverable { phase:"accept", terminal:true } to the acceptor ONLY and clears the fedCall when the relay fails', async () => { + const { handleDmCallAcceptForTest } = await importSUT(); + const connectionManager = await importManager(); + + const fedCall = makeFedCall({ + // Two ringed users — a group DM where only one accepts. Only the acceptor + // should receive the terminal undeliverable; the other ringee must stay + // in ring state so their own timeout/reject path governs teardown. + ringedUserIds: ['acceptor-user', 'other-ringee'], + }); + connectionManager.createFederatedCall(fedCall); + const sendToUserSpy = vi.spyOn(connectionManager, 'sendToUser'); + const sendToCallUsersSpy = vi.spyOn(connectionManager, 'sendToFederatedCallUsers'); + sendCallRelayMock.mockResolvedValue({ ok: false, reason: 'peer_transient_failure', error: 'timeout' }); + + await handleDmCallAcceptForTest( + { federatedCallId: fedCall.federatedId }, + fedCall.ringedUserIds[0]!, + {} as never, + ); + + expect(connectionManager.getFederatedCall(fedCall.federatedId)).toBeUndefined(); + + // Terminal undeliverable went to the acceptor only. + const undelivCalls = sendToUserSpy.mock.calls.filter(([, ev]) => + (ev as { type: string }).type === 'dm_call_undeliverable', + ); + expect(undelivCalls).toHaveLength(1); + expect(undelivCalls[0]![0]).toBe('acceptor-user'); + const undeliverable = undelivCalls[0]![1] as { phase: string; terminal: boolean; failures: unknown[] }; + expect(undeliverable.phase).toBe('accept'); + expect(undeliverable.terminal).toBe(true); + expect(undeliverable.failures).toHaveLength(1); + + // No undeliverable broadcast to all ringed users. + const callUsersCalls = sendToCallUsersSpy.mock.calls.filter(([, ev]) => + (ev as { type: string }).type === 'dm_call_undeliverable', + ); + expect(callUsersCalls).toHaveLength(0); + }); + + it('does not emit undeliverable on relay success', async () => { + const { handleDmCallAcceptForTest } = await importSUT(); + const connectionManager = await importManager(); + + const fedCall = makeFedCall(); + connectionManager.createFederatedCall(fedCall); + const sendToUserSpy = vi.spyOn(connectionManager, 'sendToUser'); + sendCallRelayMock.mockResolvedValue({ ok: true }); + + await handleDmCallAcceptForTest( + { federatedCallId: fedCall.federatedId }, + fedCall.ringedUserIds[0]!, + {} as never, + ); + + const undelivCalls = sendToUserSpy.mock.calls.filter(([, ev]) => + (ev as { type: string }).type === 'dm_call_undeliverable', + ); + expect(undelivCalls).toHaveLength(0); + expect(connectionManager.getFederatedCall(fedCall.federatedId)?.state).toBe('active'); + connectionManager.clearFederatedCall(fedCall.federatedId); + }); +}); diff --git a/packages/server/src/ws/events.ts b/packages/server/src/ws/events.ts index 393ae9bf..61955143 100644 --- a/packages/server/src/ws/events.ts +++ b/packages/server/src/ws/events.ts @@ -7,7 +7,8 @@ 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, type DmCallUndeliverableFailure, type DmCallUndeliverableReason } from '@backspace/shared'; -import type { CallRelayFailureReason } from '../utils/federationOutbox.js'; +import type { CallRelayResult, CallFanoutFailure } from '../utils/federationOutbox.js'; +import { mapCallReasonToEventReason } 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'; @@ -196,13 +197,19 @@ export function handleClientEvent( handleDmCallStart(event, userId, username, ws); break; case 'dm_call_accept': - handleDmCallAccept(event, userId, ws); + handleDmCallAccept(event, userId, ws).catch(err => + console.error('[ws] handleDmCallAccept error:', err), + ); break; case 'dm_call_reject': - handleDmCallReject(event, userId); + handleDmCallReject(event, userId).catch(err => + console.error('[ws] handleDmCallReject error:', err), + ); break; case 'dm_call_end': - handleDmCallEnd(event, userId); + handleDmCallEnd(event, userId).catch(err => + console.error('[ws] handleDmCallEnd error:', err), + ); break; case 'voice_status': handleVoiceStatus(event, userId); @@ -1442,7 +1449,7 @@ function handleDmCallStart(event: Record, userId: string, usern .catch(err => console.error('[federation] sendFederatedCallStart error:', err)); } -function handleDmCallAccept(event: Record, userId: string, ws: WebSocket): void { +async function handleDmCallAccept(event: Record, userId: string, ws: WebSocket): Promise { let dmChannelId = (event.dmChannelId as string) || null; const federatedCallId = (event.federatedCallId as string) || null; if (!dmChannelId && !federatedCallId) { @@ -1506,8 +1513,14 @@ function handleDmCallAccept(event: Record, userId: string, ws: userId, action: 'join', }); - sendFederatedCallAccept(dmChannelId, userId) - .catch(err => console.error('[federation] sendFederatedCallAccept error:', err)); + const acceptFanoutFailures = await sendFederatedCallAccept(dmChannelId, userId); + emitFanoutUndeliverable( + userId, + dmChannelId, + fedIdRow?.federatedId ?? null, + 'accept', + acceptFanoutFailures, + ); return; } } @@ -1520,6 +1533,8 @@ function handleDmCallAccept(event: Record, userId: string, ws: : undefined; if (fedCall) { + // Optimistic transition so the acceptor's client flips to active immediately. + // Rolled back below if the relay to host fails. connectionManager.activateFederatedCall(fedCall.federatedId); connectionManager.sendToFederatedCallUsers(fedCall.federatedId, { type: 'dm_call_accepted', @@ -1534,7 +1549,7 @@ function handleDmCallAccept(event: Record, userId: string, ws: .get(); const homeUserId = user?.homeUserId || userId; - sendCallRelay(fedCall.federatedCallHost, [{ + const result = await sendCallRelay(fedCall.federatedCallHost, [{ eventType: 'dm_call_accept', messageId: generateSnowflake(), encryptionVersion: 0, @@ -1543,14 +1558,32 @@ function handleDmCallAccept(event: Record, userId: string, ws: call: { acceptor: { homeUserId, homeInstance: getOurOrigin() }, }, - }]).catch(err => console.error('[federation] Failed to send dm_call_accept:', err)); + }]); + + if (!result.ok) { + console.error(`[federation] dm_call_accept relay to ${fedCall.federatedCallHost} failed (${result.reason}): ${result.error}`); + const failure = buildFailureFromResult(result, fedCall.federatedCallHost, db); + // Clear first so a concurrent end-handler sees a cleared entry (idempotent). + connectionManager.clearFederatedCall(fedCall.federatedId); + // Terminal targets ONLY the acceptor — other ringed users (group DM) didn't + // accept and should stay in their ring state; their own dm_call_end / timeout + // paths govern their teardown. + connectionManager.sendToUser(userId, { + type: 'dm_call_undeliverable', + dmChannelId: fedCall.dmChannelId, + federatedCallId: fedCall.federatedId, + terminal: true, + phase: 'accept', + failures: [failure], + }); + } return; } connectionManager.sendToUser(userId, { type: 'error', message: 'No active call in this DM channel' }); } -function handleDmCallReject(event: Record, userId: string): void { +async function handleDmCallReject(event: Record, userId: string): Promise { let dmChannelId = (event.dmChannelId as string) || null; const federatedCallId = (event.federatedCallId as string) || null; @@ -1576,8 +1609,16 @@ function handleDmCallReject(event: Record, userId: string): voi connectionManager.clearVoiceWs(meta.callerId); connectionManager.destroyRoom(dmChannelId); connectionManager.sendToDmMembers(dmChannelId, { type: 'dm_call_rejected', dmChannelId }); - sendFederatedCallEnd(dmChannelId, userId) - .catch(err => console.error('[federation] sendFederatedCallEnd error:', err)); + const fedIdRejectRow = getDb().select({ federatedId: schema.dmChannels.federatedId }) + .from(schema.dmChannels).where(eq(schema.dmChannels.id, dmChannelId)).get(); + const rejectFanoutFailures = await sendFederatedCallEnd(dmChannelId, userId); + emitFanoutUndeliverable( + userId, + dmChannelId, + fedIdRejectRow?.federatedId ?? null, + 'reject', + rejectFanoutFailures, + ); return; } } @@ -1590,13 +1631,16 @@ function handleDmCallReject(event: Record, userId: string): voi : undefined; if (fedCall) { - // Exclude the rejecting user — they already handled their own state + // Optimistic local clear — user intent is to reject. connectionManager.sendToFederatedCallUsers(fedCall.federatedId, { type: 'dm_call_rejected', dmChannelId: fedCall.dmChannelId, federatedCallId: fedCall.federatedId, } as ServerEvent, userId); - connectionManager.clearFederatedCall(fedCall.federatedId); + const host = fedCall.federatedCallHost; + const fedId = fedCall.federatedId; + const dmId = fedCall.dmChannelId; + connectionManager.clearFederatedCall(fedId); const db = getDb(); const user = db.select({ homeUserId: schema.users.homeUserId }) @@ -1605,20 +1649,33 @@ function handleDmCallReject(event: Record, userId: string): voi .get(); const homeUserId = user?.homeUserId || userId; - sendCallRelay(fedCall.federatedCallHost, [{ + const result = await sendCallRelay(host, [{ eventType: 'dm_call_reject', messageId: generateSnowflake(), encryptionVersion: 0, timestamp: Date.now(), - federatedId: fedCall.federatedId, + federatedId: fedId, call: { rejector: { homeUserId, homeInstance: getOurOrigin() }, }, - }]).catch(err => console.error('[federation] Failed to send dm_call_reject:', err)); + }]); + + if (!result.ok) { + console.error(`[federation] dm_call_reject relay to ${host} failed (${result.reason}): ${result.error}`); + const failure = buildFailureFromResult(result, host, db); + connectionManager.sendToUser(userId, { + type: 'dm_call_undeliverable', + dmChannelId: dmId, + federatedCallId: fedId, + terminal: false, + phase: 'reject', + failures: [failure], + }); + } } } -function handleDmCallEnd(event: Record, userId: string): void { +async function handleDmCallEnd(event: Record, userId: string): Promise { let dmChannelId = (event.dmChannelId as string) || null; const federatedCallId = (event.federatedCallId as string) || null; @@ -1646,10 +1703,18 @@ function handleDmCallEnd(event: Record, userId: string): void { connectionManager.clearVoiceUserStatus(participantId); connectionManager.clearVoiceWs(participantId); } + const fedIdEndRow = getDb().select({ federatedId: schema.dmChannels.federatedId }) + .from(schema.dmChannels).where(eq(schema.dmChannels.id, dmChannelId)).get(); connectionManager.destroyRoom(dmChannelId); connectionManager.sendToDmMembers(dmChannelId, { type: 'dm_call_ended', dmChannelId }); - sendFederatedCallEnd(dmChannelId, userId) - .catch(err => console.error('[federation] sendFederatedCallEnd error:', err)); + const endFanoutFailures = await sendFederatedCallEnd(dmChannelId, userId); + emitFanoutUndeliverable( + userId, + dmChannelId, + fedIdEndRow?.federatedId ?? null, + 'end', + endFanoutFailures, + ); return; } } @@ -1662,14 +1727,16 @@ function handleDmCallEnd(event: Record, userId: string): void { : undefined; if (fedCall) { - // Exclude the user who ended the call — they already disconnected in their click handler. - // Sending dm_call_ended back to them causes redundant disconnectFn() and double sounds. + // Exclude the user who ended the call — they already disconnected client-side. connectionManager.sendToFederatedCallUsers(fedCall.federatedId, { type: 'dm_call_ended', dmChannelId: fedCall.dmChannelId, federatedCallId: fedCall.federatedId, } as ServerEvent, userId); - connectionManager.clearFederatedCall(fedCall.federatedId); + const host = fedCall.federatedCallHost; + const fedId = fedCall.federatedId; + const dmId = fedCall.dmChannelId; + connectionManager.clearFederatedCall(fedId); const db = getDb(); const user = db.select({ homeUserId: schema.users.homeUserId }) @@ -1678,16 +1745,29 @@ function handleDmCallEnd(event: Record, userId: string): void { .get(); const homeUserId = user?.homeUserId || userId; - sendCallRelay(fedCall.federatedCallHost, [{ + const result = await sendCallRelay(host, [{ eventType: 'dm_call_end', messageId: generateSnowflake(), encryptionVersion: 0, timestamp: Date.now(), - federatedId: fedCall.federatedId, + federatedId: fedId, call: { endedBy: { homeUserId, homeInstance: getOurOrigin() }, }, - }]).catch(err => console.error('[federation] Failed to send dm_call_end:', err)); + }]); + + if (!result.ok) { + console.error(`[federation] dm_call_end relay to ${host} failed (${result.reason}): ${result.error}`); + const failure = buildFailureFromResult(result, host, db); + connectionManager.sendToUser(userId, { + type: 'dm_call_undeliverable', + dmChannelId: dmId, + federatedCallId: fedId, + terminal: false, + phase: 'end', + failures: [failure], + }); + } } } @@ -1896,16 +1976,24 @@ async function sendFederatedCallStart( }); } -/** 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 - } +/** Build a DmCallUndeliverableFailure from a failed CallRelayResult, enriching with peer label. */ +function buildFailureFromResult( + result: Extract, + peerOrigin: string, + db: ReturnType, +): DmCallUndeliverableFailure { + const label = db.select({ instanceName: schema.federationPeers.instanceName }) + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, peerOrigin)) + .get()?.instanceName; + return { + reason: mapCallReasonToEventReason(result.reason), + peerOrigin, + peerLabel: label ?? undefined, + }; } + /** * Emit dm_call_undeliverable to the caller. If terminal, also destroy the * local ring room (which clears the ringing timer and voice WS binding) @@ -1944,17 +2032,49 @@ function emitUndeliverableAndMaybeDestroy(args: { dmChannelId, federatedCallId: federatedId, terminal, + phase: 'start', failures, }); } -async function sendFederatedCallAccept(dmChannelId: string, acceptorUserId: string): Promise { +/** + * Emit a non-terminal dm_call_undeliverable to the local user whose Path-1 action + * (accept / reject / end) had one or more fan-out failures reach peers. + * No-op when there were no failures or no federatedId to reference. + */ +function emitFanoutUndeliverable( + userId: string, + dmChannelId: string | null, + federatedId: string | null | undefined, + phase: 'accept' | 'reject' | 'end', + fanoutFailures: CallFanoutFailure[], +): void { + if (fanoutFailures.length === 0 || !federatedId) return; + const failures: DmCallUndeliverableFailure[] = fanoutFailures.map(f => ({ + reason: f.reason, + peerOrigin: f.origin, + peerLabel: f.peerLabel, + })); + connectionManager.sendToUser(userId, { + type: 'dm_call_undeliverable', + dmChannelId, + federatedCallId: federatedId, + terminal: false, + phase, + failures, + }); +} + +async function sendFederatedCallAccept( + dmChannelId: string, + acceptorUserId: string, +): Promise { const db = getDb(); const channel = db.select({ federatedId: schema.dmChannels.federatedId }) .from(schema.dmChannels) .where(eq(schema.dmChannels.id, dmChannelId)) .get(); - if (!channel?.federatedId) return; + if (!channel?.federatedId) return []; const members = db.select({ homeInstance: schema.users.homeInstance }) .from(schema.dmMembers) @@ -1970,7 +2090,7 @@ async function sendFederatedCallAccept(dmChannelId: string, acceptorUserId: stri if (normalized !== ourOrigin) targets.add(normalized); } } - if (targets.size === 0) return; + if (targets.size === 0) return []; const user = db.select({ homeUserId: schema.users.homeUserId }) .from(schema.users) @@ -1989,22 +2109,41 @@ async function sendFederatedCallAccept(dmChannelId: string, acceptorUserId: stri }, }; - await Promise.all( - Array.from(targets).map(origin => - sendCallRelay(origin, [event]).catch(err => - console.error(`[federation] Failed to send dm_call_accept to ${origin}:`, err) - ) - ) + const labelByOrigin = new Map(); + for (const r of db.select({ origin: schema.federationPeers.origin, instanceName: schema.federationPeers.instanceName }) + .from(schema.federationPeers) + .all()) { + labelByOrigin.set(r.origin, r.instanceName ?? null); + } + + const results = await Promise.all( + Array.from(targets).map(async origin => ({ origin, result: await sendCallRelay(origin, [event]) })), ); + + const failures: CallFanoutFailure[] = []; + for (const { origin, result } of results) { + if (!result.ok) { + console.error(`[federation] dm_call_accept fanout to ${origin} failed (${result.reason}): ${result.error}`); + failures.push({ + origin, + peerLabel: labelByOrigin.get(origin) ?? undefined, + reason: mapCallReasonToEventReason(result.reason), + }); + } + } + return failures; } -async function sendFederatedCallEnd(dmChannelId: string, endedByUserId: string): Promise { +async function sendFederatedCallEnd( + dmChannelId: string, + endedByUserId: string, +): Promise { const db = getDb(); const channel = db.select({ federatedId: schema.dmChannels.federatedId }) .from(schema.dmChannels) .where(eq(schema.dmChannels.id, dmChannelId)) .get(); - if (!channel?.federatedId) return; + if (!channel?.federatedId) return []; const members = db.select({ homeInstance: schema.users.homeInstance }) .from(schema.dmMembers) @@ -2020,9 +2159,8 @@ async function sendFederatedCallEnd(dmChannelId: string, endedByUserId: string): if (normalized !== ourOrigin) targets.add(normalized); } } - if (targets.size === 0) return; + if (targets.size === 0) return []; - // Resolve actual homeUserId from DB const endUser = db.select({ homeUserId: schema.users.homeUserId }) .from(schema.users) .where(eq(schema.users.id, endedByUserId)) @@ -2040,13 +2178,29 @@ async function sendFederatedCallEnd(dmChannelId: string, endedByUserId: string): }, }; - await Promise.all( - Array.from(targets).map(origin => - sendCallRelay(origin, [event]).catch(err => - console.error(`[federation] Failed to send dm_call_end to ${origin}:`, err) - ) - ) + const labelByOrigin = new Map(); + for (const r of db.select({ origin: schema.federationPeers.origin, instanceName: schema.federationPeers.instanceName }) + .from(schema.federationPeers) + .all()) { + labelByOrigin.set(r.origin, r.instanceName ?? null); + } + + const results = await Promise.all( + Array.from(targets).map(async origin => ({ origin, result: await sendCallRelay(origin, [event]) })), ); + + const failures: CallFanoutFailure[] = []; + for (const { origin, result } of results) { + if (!result.ok) { + console.error(`[federation] dm_call_end fanout to ${origin} failed (${result.reason}): ${result.error}`); + failures.push({ + origin, + peerLabel: labelByOrigin.get(origin) ?? undefined, + reason: mapCallReasonToEventReason(result.reason), + }); + } + } + return failures; } // ─── Voice Moderation Handlers ────────────────────────────────────────────── @@ -2308,3 +2462,24 @@ function handleVoiceDisconnect(event: Record, userId: string): channelId, }); } + +/** + * Register the ring-timeout fan-out so a host-side 60s auto-clean notifies remote peers. + * Called from server startup; split from module-load to keep test isolation clean + * (tests that exercise ring timeouts can register their own stub via + * `connectionManager.setRingTimeoutFanoutHook`). + */ +export function registerCallRelayHooks(): void { + connectionManager.setRingTimeoutFanoutHook(async (dmChannelId, callerId) => { + const failures = await sendFederatedCallEnd(dmChannelId, callerId); + if (failures.length > 0) { + console.warn('[federation] Ring-timeout fan-out had failures:', failures); + } + }); +} + +// ─── Test-only exports ────────────────────────────────────────────────────── +/** Direct export for unit tests — do not use in production code paths. */ +export const handleDmCallAcceptForTest = handleDmCallAccept; +export const handleDmCallRejectForTest = handleDmCallReject; +export const handleDmCallEndForTest = handleDmCallEnd; diff --git a/packages/server/src/ws/handler.ts b/packages/server/src/ws/handler.ts index 713c20c8..72e36f1f 100644 --- a/packages/server/src/ws/handler.ts +++ b/packages/server/src/ws/handler.ts @@ -99,6 +99,9 @@ class ConnectionManager { private pendingOfflineTimeouts: Map = new Map(); // roomId → Timeout for ringing DM rooms (60s auto-cleanup) private ringingTimeouts: Map = new Map(); + // Callback registered by events.ts to fan dm_call_end out to peers on ring timeout. + // Null during startup — ring timeouts that fire before registration simply no-op (there are no peers to notify before boot completes). + private ringTimeoutFanoutHook: ((dmChannelId: string, callerId: string) => Promise) | null = null; /** Federated calls where this instance is NOT the host. Keyed by federatedId. */ private federatedCalls: Map = new Map(); private federatedCallTimeouts: Map = new Map(); @@ -401,6 +404,11 @@ class ConnectionManager { return true; } + /** Register a fan-out callback invoked when a ringing DM room hits its 60s timeout. */ + setRingTimeoutFanoutHook(fn: (dmChannelId: string, callerId: string) => Promise): void { + this.ringTimeoutFanoutHook = fn; + } + /** Create a DM room in ringing state with 60s auto-cleanup. */ createDmRoom(dmChannelId: string, callerId: string): boolean { const created = this.createRoom(dmChannelId, 'dm', { @@ -415,11 +423,19 @@ class ConnectionManager { this.ringingTimeouts.delete(dmChannelId); const room = this.voiceRooms.get(dmChannelId); if (room && room.roomType === 'dm' && (room.metadata as DmRoomMeta).state === 'ringing') { + const ringedCallerId = (room.metadata as DmRoomMeta).callerId; this.destroyRoom(dmChannelId); this.sendToDmMembers(dmChannelId, { type: 'dm_call_ended', dmChannelId, }); + // Fan dm_call_end out to remote peers so stranded Path-A/B ringees exit the ring. + // Without this, an accept-relay failure → Alice's 60s auto-clean leaves Bob's FederatedCallEntry lingering with no terminal event. + if (this.ringTimeoutFanoutHook) { + this.ringTimeoutFanoutHook(dmChannelId, ringedCallerId).catch(err => + console.error('[ws] ring-timeout fan-out error:', err), + ); + } } }, 60_000); this.ringingTimeouts.set(dmChannelId, timeout); diff --git a/packages/shared/src/types.ts b/packages/shared/src/types.ts index 6d5a3903..e2907658 100644 --- a/packages/shared/src/types.ts +++ b/packages/shared/src/types.ts @@ -364,6 +364,8 @@ export type DmCallUndeliverableReason = | 'peer_transient_failure' | 'livekit_unavailable'; +export type DmCallPhase = 'start' | 'accept' | 'reject' | 'end'; + export interface DmCallUndeliverableFailure { reason: DmCallUndeliverableReason; peerOrigin?: string; @@ -426,7 +428,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: 'dm_call_undeliverable'; dmChannelId: string | null; federatedCallId: string; terminal: boolean; phase: DmCallPhase; 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/__tests__/useWebSocket.callUndeliverable.test.ts b/packages/web/src/hooks/__tests__/useWebSocket.callUndeliverable.test.ts new file mode 100644 index 00000000..2905ef14 --- /dev/null +++ b/packages/web/src/hooks/__tests__/useWebSocket.callUndeliverable.test.ts @@ -0,0 +1,37 @@ +import { describe, it, expect } from 'vitest'; +import { buildCallUndeliverableToast } from '../../utils/callUndeliverableToast.js'; + +describe('buildCallUndeliverableToast', () => { + const fail = (overrides: Partial<{ reason: string; peerOrigin?: string; peerLabel?: string }> = {}) => ({ + reason: 'peer_transient_failure', + peerLabel: 'nova', + ...overrides, + }); + + it('start + terminal single failure: existing copy', () => { + expect(buildCallUndeliverableToast([fail()], true, 'start')).toMatch(/Could not reach nova/); + }); + + it('start + non-terminal: "some participants" copy', () => { + expect(buildCallUndeliverableToast([fail()], false, 'start')).toMatch(/Some participants could not be reached/); + }); + + it('accept + terminal: tear-down copy', () => { + expect(buildCallUndeliverableToast([fail()], true, 'accept')) + .toMatch(/Couldn't confirm your accept with nova/); + }); + + it('reject + non-terminal: info copy', () => { + expect(buildCallUndeliverableToast([fail()], false, 'reject')) + .toMatch(/Couldn't notify nova that you declined/); + }); + + it('end + non-terminal: info copy', () => { + expect(buildCallUndeliverableToast([fail()], false, 'end')) + .toMatch(/Couldn't notify nova that you hung up/); + }); + + it('legacy two-arg signature still works', () => { + expect(buildCallUndeliverableToast([fail()], true)).toMatch(/Could not reach nova/); + }); +}); diff --git a/packages/web/src/hooks/useWebSocket.ts b/packages/web/src/hooks/useWebSocket.ts index 80fc875c..3a5b537c 100644 --- a/packages/web/src/hooks/useWebSocket.ts +++ b/packages/web/src/hooks/useWebSocket.ts @@ -116,40 +116,9 @@ function buildWsUrl(origin: string): string { // ─── 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'; +import { buildCallUndeliverableToast } from '../utils/callUndeliverableToast'; - 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.`; - } -} +export { buildCallUndeliverableToast }; // ─── Event handling ─────────────────────────────────────────────────────────── @@ -1030,7 +999,7 @@ function handleEvent(origin: string, event: ServerEvent): void { if (disconnectFn) disconnectFn(); } - const msg = buildCallUndeliverableToast(event.failures, event.terminal); + const msg = buildCallUndeliverableToast(event.failures, event.terminal, event.phase); addToast(msg, event.terminal ? 'warning' : 'info', 8_000); break; } diff --git a/packages/web/src/utils/callUndeliverableToast.ts b/packages/web/src/utils/callUndeliverableToast.ts new file mode 100644 index 00000000..2d14f830 --- /dev/null +++ b/packages/web/src/utils/callUndeliverableToast.ts @@ -0,0 +1,64 @@ +/** + * Builds a user-facing toast message from a `dm_call_undeliverable` event. + * + * Copy is phase-aware: + * - `start`: call-start delivery; terminal means the ring was destroyed, non-terminal + * means the call continues for other reachable recipients. + * - `accept`: the acceptor's B→host relay failed; terminal means their optimistic + * active-call state was rolled back. + * - `reject`: the rejector's relay to the host failed; state was already cleared + * locally, so non-terminal info toast only. + * - `end`: the ender's relay to the host failed; state was already cleared locally. + * + * Extracted from `useWebSocket.ts` so it can be unit-tested without pulling in + * the full WS handler graph (livekit / audio deps). + */ +export function buildCallUndeliverableToast( + failures: Array<{ reason: string; peerOrigin?: string; peerLabel?: string }>, + terminal: boolean, + phase: 'start' | 'accept' | 'reject' | 'end' = 'start', +): string { + const primary = failures[0]; + const labelFor = (f: { peerLabel?: string; peerOrigin?: string }) => + f.peerLabel ?? f.peerOrigin?.replace(/^https?:\/\//, '') ?? 'the remote instance'; + + if (phase === 'accept' && terminal) { + const label = primary ? labelFor(primary) : 'the host instance'; + return `Couldn't confirm your accept with ${label} — the call was dropped.`; + } + + if (phase === 'reject') { + const labels = failures.map(labelFor).join(', ') || 'the host instance'; + return `Couldn't notify ${labels} that you declined. Caller may still see you as ringing briefly.`; + } + + if (phase === 'end') { + const labels = failures.map(labelFor).join(', ') || 'the host instance'; + return `Couldn't notify ${labels} that you hung up. Remote participants may see the call for up to 60 seconds.`; + } + + // phase === 'start' (default + legacy) + 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.`; + } +}