diff --git a/apps/chanora_flutter/lib/l10n/app_en.arb b/apps/chanora_flutter/lib/l10n/app_en.arb index d12ea5f..daaad93 100644 --- a/apps/chanora_flutter/lib/l10n/app_en.arb +++ b/apps/chanora_flutter/lib/l10n/app_en.arb @@ -227,6 +227,7 @@ "clientInfoOnline": "Online", "clientInfoIdle": "Idle", "clientInfoPing": "Ping", + "clientInfoPingDeviation": "Ping deviation", "clientInfoAddress": "Address", "clientInfoPacketLossClientToServer": "Packet loss C->S", "clientInfoPacketLossServerToClient": "Packet loss S->C", diff --git a/apps/chanora_flutter/lib/l10n/app_zh.arb b/apps/chanora_flutter/lib/l10n/app_zh.arb index a3af60c..d5bc8e7 100644 --- a/apps/chanora_flutter/lib/l10n/app_zh.arb +++ b/apps/chanora_flutter/lib/l10n/app_zh.arb @@ -176,6 +176,7 @@ "clientInfoOnline": "在线时长", "clientInfoIdle": "空闲", "clientInfoPing": "延迟", + "clientInfoPingDeviation": "延迟偏差", "clientInfoAddress": "地址", "clientInfoPacketLossClientToServer": "丢包 C->S", "clientInfoPacketLossServerToClient": "丢包 S->C", diff --git a/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart b/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart index a32831a..5b27ffa 100644 --- a/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart +++ b/apps/chanora_flutter/lib/l10n/generated/app_localizations.dart @@ -1069,6 +1069,12 @@ abstract class AppL10n { /// **'Ping'** String get clientInfoPing; + /// No description provided for @clientInfoPingDeviation. + /// + /// In en, this message translates to: + /// **'Ping deviation'** + String get clientInfoPingDeviation; + /// No description provided for @clientInfoAddress. /// /// In en, this message translates to: diff --git a/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart b/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart index cdc2b6b..cdf4d65 100644 --- a/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart +++ b/apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart @@ -543,6 +543,9 @@ class AppL10nEn extends AppL10n { @override String get clientInfoPing => 'Ping'; + @override + String get clientInfoPingDeviation => 'Ping deviation'; + @override String get clientInfoAddress => 'Address'; diff --git a/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart b/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart index 6586b08..d90a2dd 100644 --- a/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart +++ b/apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart @@ -531,6 +531,9 @@ class AppL10nZh extends AppL10n { @override String get clientInfoPing => '延迟'; + @override + String get clientInfoPingDeviation => '延迟偏差'; + @override String get clientInfoAddress => '地址'; diff --git a/apps/chanora_flutter/lib/main.dart b/apps/chanora_flutter/lib/main.dart index 553ac26..46a1254 100644 --- a/apps/chanora_flutter/lib/main.dart +++ b/apps/chanora_flutter/lib/main.dart @@ -809,17 +809,19 @@ class _BetaHomeState extends State<_BetaHome> with WidgetsBindingObserver { return updated; }); case rust.BridgeEvent_ClientJoined(:final clientId, :final channelId, :final name, :final inputMuted, :final outputMuted, :final isServerQuery, :final talkPower, :final talkPowerGranted): - _applyClientAdd(rust.BridgeClient( - id: clientId, - channel: channelId, - name: name, - inputMuted: inputMuted, - outputMuted: outputMuted, - isSpeaking: false, - isServerQuery: isServerQuery, - talkPower: talkPower, - talkPowerGranted: talkPowerGranted, - )); + if (!isServerQuery) { + _applyClientAdd(rust.BridgeClient( + id: clientId, + channel: channelId, + name: name, + inputMuted: inputMuted, + outputMuted: outputMuted, + isSpeaking: false, + isServerQuery: isServerQuery, + talkPower: talkPower, + talkPowerGranted: talkPowerGranted, + )); + } case rust.BridgeEvent_ClientLeft(:final clientId): _applyClientRemove(clientId); case rust.BridgeEvent_ClientUpdated(:final clientId, :final inputMuted, :final outputMuted, :final isServerQuery, :final talkPower, :final talkPowerGranted): @@ -871,14 +873,25 @@ class _BetaHomeState extends State<_BetaHome> with WidgetsBindingObserver { if (_serverReachable) _ensureStatsTimer(); } + DateTime? _lastSpeakingRefresh; + static const _speakingRefreshInterval = Duration(milliseconds: 750); + void _ensureStatsTimer() { if (_statsTimer != null) return; + _lastSpeakingRefresh = null; _statsTimer = Timer.periodic(const Duration(milliseconds: 250), (_) async { try { final s = await rust.audioStats(); if (!mounted) return; setState(() => _audioStats = s); } catch (_) {} + final now = DateTime.now(); + if (_inChannel && + (_lastSpeakingRefresh == null || + now.difference(_lastSpeakingRefresh!) >= _speakingRefreshInterval)) { + _lastSpeakingRefresh = now; + unawaited(_refreshSnapshot(recordActivity: false, reportErrors: false)); + } }); } diff --git a/apps/chanora_flutter/lib/src/rust/api.dart b/apps/chanora_flutter/lib/src/rust/api.dart index b37506d..2f96d0b 100644 --- a/apps/chanora_flutter/lib/src/rust/api.dart +++ b/apps/chanora_flutter/lib/src/rust/api.dart @@ -917,6 +917,9 @@ class BridgeClientProfile { /// Current ping in milliseconds. final PlatformInt64? pingMilliseconds; + /// Current ping deviation in milliseconds. + final PlatformInt64? pingDeviationMilliseconds; + /// Client address. Empty when permission-gated. final String clientAddress; @@ -963,6 +966,7 @@ class BridgeClientProfile { this.onlineSeconds, this.idleMilliseconds, this.pingMilliseconds, + this.pingDeviationMilliseconds, required this.clientAddress, required this.serverGroups, required this.channelGroup, @@ -992,6 +996,7 @@ class BridgeClientProfile { onlineSeconds.hashCode ^ idleMilliseconds.hashCode ^ pingMilliseconds.hashCode ^ + pingDeviationMilliseconds.hashCode ^ clientAddress.hashCode ^ serverGroups.hashCode ^ channelGroup.hashCode ^ @@ -1023,6 +1028,7 @@ class BridgeClientProfile { onlineSeconds == other.onlineSeconds && idleMilliseconds == other.idleMilliseconds && pingMilliseconds == other.pingMilliseconds && + pingDeviationMilliseconds == other.pingDeviationMilliseconds && clientAddress == other.clientAddress && serverGroups == other.serverGroups && channelGroup == other.channelGroup && diff --git a/apps/chanora_flutter/lib/src/rust/frb_generated.dart b/apps/chanora_flutter/lib/src/rust/frb_generated.dart index 51119c4..edb74e8 100644 --- a/apps/chanora_flutter/lib/src/rust/frb_generated.dart +++ b/apps/chanora_flutter/lib/src/rust/frb_generated.dart @@ -1844,8 +1844,8 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { BridgeClientProfile dco_decode_bridge_client_profile(dynamic raw) { // Codec=Dco (DartCObject based), see doc to use other codecs final arr = raw as List; - if (arr.length != 25) - throw Exception('unexpected arr length: expect 25 but see ${arr.length}'); + if (arr.length != 26) + throw Exception('unexpected arr length: expect 26 but see ${arr.length}'); return BridgeClientProfile( id: dco_decode_u_64(arr[0]), channel: dco_decode_u_64(arr[1]), @@ -1862,16 +1862,17 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { onlineSeconds: dco_decode_opt_box_autoadd_i_64(arr[12]), idleMilliseconds: dco_decode_opt_box_autoadd_i_64(arr[13]), pingMilliseconds: dco_decode_opt_box_autoadd_i_64(arr[14]), - clientAddress: dco_decode_String(arr[15]), - serverGroups: dco_decode_list_String(arr[16]), - channelGroup: dco_decode_String(arr[17]), - avatarPath: dco_decode_String(arr[18]), - bytesDownloadedMonth: dco_decode_opt_box_autoadd_u_64(arr[19]), - bytesUploadedMonth: dco_decode_opt_box_autoadd_u_64(arr[20]), - bytesDownloadedTotal: dco_decode_opt_box_autoadd_u_64(arr[21]), - bytesUploadedTotal: dco_decode_opt_box_autoadd_u_64(arr[22]), - packetLossClientToServerTotal: dco_decode_opt_box_autoadd_f_32(arr[23]), - packetLossServerToClientTotal: dco_decode_opt_box_autoadd_f_32(arr[24]), + pingDeviationMilliseconds: dco_decode_opt_box_autoadd_i_64(arr[15]), + clientAddress: dco_decode_String(arr[16]), + serverGroups: dco_decode_list_String(arr[17]), + channelGroup: dco_decode_String(arr[18]), + avatarPath: dco_decode_String(arr[19]), + bytesDownloadedMonth: dco_decode_opt_box_autoadd_u_64(arr[20]), + bytesUploadedMonth: dco_decode_opt_box_autoadd_u_64(arr[21]), + bytesDownloadedTotal: dco_decode_opt_box_autoadd_u_64(arr[22]), + bytesUploadedTotal: dco_decode_opt_box_autoadd_u_64(arr[23]), + packetLossClientToServerTotal: dco_decode_opt_box_autoadd_f_32(arr[24]), + packetLossServerToClientTotal: dco_decode_opt_box_autoadd_f_32(arr[25]), ); } @@ -2577,6 +2578,9 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { var var_onlineSeconds = sse_decode_opt_box_autoadd_i_64(deserializer); var var_idleMilliseconds = sse_decode_opt_box_autoadd_i_64(deserializer); var var_pingMilliseconds = sse_decode_opt_box_autoadd_i_64(deserializer); + var var_pingDeviationMilliseconds = sse_decode_opt_box_autoadd_i_64( + deserializer, + ); var var_clientAddress = sse_decode_String(deserializer); var var_serverGroups = sse_decode_list_String(deserializer); var var_channelGroup = sse_decode_String(deserializer); @@ -2611,6 +2615,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { onlineSeconds: var_onlineSeconds, idleMilliseconds: var_idleMilliseconds, pingMilliseconds: var_pingMilliseconds, + pingDeviationMilliseconds: var_pingDeviationMilliseconds, clientAddress: var_clientAddress, serverGroups: var_serverGroups, channelGroup: var_channelGroup, @@ -3436,6 +3441,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi { sse_encode_opt_box_autoadd_i_64(self.onlineSeconds, serializer); sse_encode_opt_box_autoadd_i_64(self.idleMilliseconds, serializer); sse_encode_opt_box_autoadd_i_64(self.pingMilliseconds, serializer); + sse_encode_opt_box_autoadd_i_64(self.pingDeviationMilliseconds, serializer); sse_encode_String(self.clientAddress, serializer); sse_encode_list_String(self.serverGroups, serializer); sse_encode_String(self.channelGroup, serializer); diff --git a/apps/chanora_flutter/lib/widgets/client_info_sheet.dart b/apps/chanora_flutter/lib/widgets/client_info_sheet.dart index 061edba..94ae579 100644 --- a/apps/chanora_flutter/lib/widgets/client_info_sheet.dart +++ b/apps/chanora_flutter/lib/widgets/client_info_sheet.dart @@ -214,6 +214,11 @@ class _ClientInfoContent extends StatelessWidget { l10n.clientInfoPing, _formatMilliseconds(profile.pingMilliseconds, l10n), ), + if (profile.pingDeviationMilliseconds != null) + _InfoRowData( + l10n.clientInfoPingDeviation, + _formatMilliseconds(profile.pingDeviationMilliseconds, l10n), + ), _InfoRowData( l10n.clientInfoAddress, _emptyAsHidden(profile.clientAddress, l10n), diff --git a/apps/chanora_flutter/test/widgets/client_info_sheet_test.dart b/apps/chanora_flutter/test/widgets/client_info_sheet_test.dart index cc0e42a..2fe70a4 100644 --- a/apps/chanora_flutter/test/widgets/client_info_sheet_test.dart +++ b/apps/chanora_flutter/test/widgets/client_info_sheet_test.dart @@ -20,6 +20,7 @@ void main() { onlineSeconds: 3661, idleMilliseconds: 42000, pingMilliseconds: 38, + pingDeviationMilliseconds: 7, clientAddress: '203.0.113.24', serverGroups: const ['Admin', 'Talk Power'], channelGroup: 'Guest', @@ -77,9 +78,56 @@ void main() { expect(find.text('Connection'), findsOneWidget); expect(find.text('1h 1m 1s'), findsOneWidget); expect(find.text('42.00 s'), findsOneWidget); + expect(find.text('7 ms'), findsOneWidget); expect(find.text('203.0.113.24'), findsOneWidget); }); + testWidgets('hides ping deviation row when the value is absent', (tester) async { + await tester.pumpWidget( + MaterialApp( + localizationsDelegates: AppL10n.localizationsDelegates, + supportedLocales: AppL10n.supportedLocales, + home: Scaffold( + body: SizedBox( + height: 500, + child: ClientInfoSheet( + clientName: 'Bob', + loadProfile: () async => rust.BridgeClientProfile( + id: BigInt.from(101), + channel: BigInt.from(7), + name: 'Bob', + uniqueId: 'client-unique-id', + databaseId: BigInt.from(55), + countryCode: 'US', + description: 'Operator', + version: '3.6.2', + platform: 'Windows', + onlineSeconds: 3661, + idleMilliseconds: 42000, + pingMilliseconds: 38, + clientAddress: '203.0.113.24', + serverGroups: const ['Admin'], + channelGroup: 'Guest', + avatarPath: '/avatar_aabbcc', + bytesDownloadedMonth: BigInt.from(2048), + bytesUploadedMonth: BigInt.from(4096), + bytesDownloadedTotal: BigInt.from(1048576), + bytesUploadedTotal: BigInt.from(2097152), + packetLossClientToServerTotal: 0.0123, + packetLossServerToClientTotal: 0.0456, + ), + ), + ), + ), + ), + ); + + await tester.pumpAndSettle(); + + expect(find.text('Ping deviation'), findsNothing); + expect(find.text('7 ms'), findsNothing); + }); + testWidgets('loading and error states use localized copy', (tester) async { await tester.pumpWidget( MaterialApp( diff --git a/crates/chanora_bridge/src/api.rs b/crates/chanora_bridge/src/api.rs index 6f74f1b..a57ec69 100644 --- a/crates/chanora_bridge/src/api.rs +++ b/crates/chanora_bridge/src/api.rs @@ -473,6 +473,8 @@ pub struct BridgeClientProfile { pub idle_milliseconds: Option, /// Current ping in milliseconds. pub ping_milliseconds: Option, + /// Current ping deviation in milliseconds. + pub ping_deviation_milliseconds: Option, /// Client address. Empty when permission-gated. pub client_address: String, /// Resolved server group names. @@ -534,6 +536,7 @@ impl From for BridgeClientProfile { online_seconds: profile.online_seconds, idle_milliseconds: profile.idle_milliseconds, ping_milliseconds: profile.ping_milliseconds, + ping_deviation_milliseconds: profile.ping_deviation_milliseconds, client_address: profile.client_address, server_groups: profile.server_groups, channel_group: profile.channel_group, diff --git a/crates/chanora_bridge/src/frb_generated.rs b/crates/chanora_bridge/src/frb_generated.rs index 0ccbd89..c7aa75c 100644 --- a/crates/chanora_bridge/src/frb_generated.rs +++ b/crates/chanora_bridge/src/frb_generated.rs @@ -2052,6 +2052,7 @@ impl SseDecode for crate::api::BridgeClientProfile { let mut var_onlineSeconds = >::sse_decode(deserializer); let mut var_idleMilliseconds = >::sse_decode(deserializer); let mut var_pingMilliseconds = >::sse_decode(deserializer); + let mut var_pingDeviationMilliseconds = >::sse_decode(deserializer); let mut var_clientAddress = ::sse_decode(deserializer); let mut var_serverGroups = >::sse_decode(deserializer); let mut var_channelGroup = ::sse_decode(deserializer); @@ -2078,6 +2079,7 @@ impl SseDecode for crate::api::BridgeClientProfile { online_seconds: var_onlineSeconds, idle_milliseconds: var_idleMilliseconds, ping_milliseconds: var_pingMilliseconds, + ping_deviation_milliseconds: var_pingDeviationMilliseconds, client_address: var_clientAddress, server_groups: var_serverGroups, channel_group: var_channelGroup, @@ -3077,6 +3079,9 @@ impl flutter_rust_bridge::IntoDart for crate::api::BridgeClientProfile { self.online_seconds.into_into_dart().into_dart(), self.idle_milliseconds.into_into_dart().into_dart(), self.ping_milliseconds.into_into_dart().into_dart(), + self.ping_deviation_milliseconds + .into_into_dart() + .into_dart(), self.client_address.into_into_dart().into_dart(), self.server_groups.into_into_dart().into_dart(), self.channel_group.into_into_dart().into_dart(), @@ -3835,6 +3840,7 @@ impl SseEncode for crate::api::BridgeClientProfile { >::sse_encode(self.online_seconds, serializer); >::sse_encode(self.idle_milliseconds, serializer); >::sse_encode(self.ping_milliseconds, serializer); + >::sse_encode(self.ping_deviation_milliseconds, serializer); ::sse_encode(self.client_address, serializer); >::sse_encode(self.server_groups, serializer); ::sse_encode(self.channel_group, serializer); diff --git a/crates/chanora_protocol/src/adapter.rs b/crates/chanora_protocol/src/adapter.rs index 2e3a46c..c99c000 100644 --- a/crates/chanora_protocol/src/adapter.rs +++ b/crates/chanora_protocol/src/adapter.rs @@ -31,8 +31,8 @@ use tsclientlib::data::{self, Channel, Client}; use tsclientlib::messages::s2c::{InClientDbInfoPart, InMessage}; use tsclientlib::prelude::*; use tsclientlib::{ - ChannelId as TsChannelId, ClientId as TsClientId, Connection, DisconnectOptions, Identity, - MessageHandle, OutCommandExt, StreamItem, Version, + ChannelId as TsChannelId, ClientId as TsClientId, Connection, ConnectionStats, + DisconnectOptions, Identity, MessageHandle, OutCommandExt, StreamItem, Version, }; use tsproto_packets::packets::{Direction, Flags, InAudioBuf, OutCommand, OutPacket, PacketType}; use tsproto_types::ClientType; @@ -45,6 +45,16 @@ use crate::ProtocolError; const SPEAKING_ACTIVITY_WINDOW: Duration = Duration::from_millis(750); const INBOUND_VOICE_SEND_TIMEOUT: Duration = Duration::from_millis(40); +const PROFILE_REFRESH_RESULT_TIMEOUT: Duration = Duration::from_secs(3); + +type PendingMoves = HashMap< + MessageHandle, + ( + u64, + Option>>, + std::time::Instant, + ), +>; #[derive(Debug, PartialEq, Eq)] enum SendTimeoutError { @@ -652,14 +662,7 @@ async fn connection_task( // resolve the oneshot back to the caller. Entries also carry a // deadline so a server that never replies doesn't leak the // reply channel — at most 3 s of pending state per move. - let mut pending_moves: HashMap< - MessageHandle, - ( - u64, - Option>>, - std::time::Instant, - ), - > = HashMap::new(); + let mut pending_moves: PendingMoves = HashMap::new(); let mut voice_activity: HashMap = HashMap::new(); // Main loop: pump events, service requests, forward voice. @@ -680,157 +683,16 @@ async fn connection_task( Ok(Some(Ok(item))) => { match item { StreamItem::Audio(buf) => { - let from = packet_sender_id(&buf); - if let Some(from) = from { - voice_activity.insert(from, Instant::now()); - let inbound = InboundVoice { - from_client: from, - packet: buf, - }; - match send_with_timeout( - &voice_in_tx, - inbound, - INBOUND_VOICE_SEND_TIMEOUT, - ) - .await - { - Ok(()) => {} - Err(SendTimeoutError::Timeout(_)) => { - warn!( - target: "chanora_protocol", - from_client = from, - timeout_ms = INBOUND_VOICE_SEND_TIMEOUT.as_millis() as u64, - "inbound voice queue stayed full; dropping packet" - ); - } - Err(SendTimeoutError::Closed(_)) => { - warn!( - target: "chanora_protocol", - from_client = from, - "inbound voice consumer closed; dropping packet" - ); - } - } - } + handle_audio_stream_item(&voice_in_tx, &mut voice_activity, buf).await; } - StreamItem::BookEvents(events) => { - for ev in events { - if let tsclientlib::events::Event::PropertyChanged { - id: ts_bookkeeping::events::PropertyId::ClientChannel(client_id), - .. - } = &ev - { - let own_client = con.get_state().ok().map(|state| state.own_client); - if let Some(state) = con.get_state().ok() { - if let Some(client) = state.clients.get(client_id) { - let _ = delta_tx.try_send(ProtocolDelta::ClientMoved { - client_id: client_id.0 as u64, - new_channel_id: client.channel.0, - }); - } - } - if own_client == Some(*client_id) { - let current_channel = con.get_state().ok().and_then(|state| { - state.clients.get(client_id).map(|client| client.channel.0) - }); - if let Some(current_channel) = current_channel { - let matched: Vec = pending_moves - .iter() - .filter_map(|(handle, (target_channel, _, _))| { - if *target_channel == current_channel { - Some(*handle) - } else { - None - } - }) - .collect(); - for handle in matched { - if let Some((_, reply, _)) = - pending_moves.remove(&handle) - { - info!( - target: "chanora_protocol", - channel_id = current_channel, - "client_move resolved by authoritative self channel change" - ); - if let Some(reply) = reply { - let _ = reply.send(Ok(())); - } - } - } - } - } - } - - if let Some(activity) = format_server_activity(&con, &ev) { - let _ = activity_tx.try_send(ServerActivity { message: activity }); - } - - forward_delta(&con, &ev, &delta_tx); - - if let tsclientlib::events::Event::Message { - target, - invoker, - message, - } = ev - { - let mapped = match target { - tsclientlib::MessageTarget::Server => MessageTarget::Server, - tsclientlib::MessageTarget::Channel => MessageTarget::Channel, - tsclientlib::MessageTarget::Client(id) => { - MessageTarget::Client(id.0 as u64) - } - tsclientlib::MessageTarget::Poke(id) => { - MessageTarget::Poke(id.0 as u64) - } - }; - let _ = chat_tx.try_send(ChatMessage { - sender_id: ClientId(invoker.id.0 as u64), - sender_name: sanitize(&invoker.name), - message: sanitize(&message), - target: mapped, - }); - } - } - } - StreamItem::MessageResult(handle, result) => { - if let Some((_target_channel, reply, _deadline)) = - pending_moves.remove(&handle) - { - let mapped = match result { - Ok(()) => Ok(()), - Err(cmd_err) => { - // tsclientlib's CommandError carries a - // typed `TsError` (the canonical TS3 - // error code) plus an optional missing - // permission. We convert to our typed - // ProtocolError::ServerRejected so the - // upper layers can render a localised - // explanation by code instead of a - // generic backend string. - let code = cmd_err.error as u32; - let message = cmd_err.error.to_string(); - info!( - target: "chanora_protocol", - code, - message = %message, - "server rejected client_move" - ); - Err(ProtocolError::ServerRejected { code, message }) - } - }; - if let Some(reply) = reply { - let _ = reply.send(mapped); - } else if let Err(err) = mapped { - info!( - target: "chanora_protocol", - error = %err, - "client_move completed in background with error" - ); - } - } - } - _ => { /* book / message / other events: ignore */ } + other => handle_non_audio_stream_item( + &con, + other, + &chat_tx, + &activity_tx, + &delta_tx, + &mut pending_moves, + ), } } Ok(Some(Err(e))) => { @@ -926,7 +788,17 @@ async fn connection_task( let _ = reply.send(r); } Ok(Request::FetchClientProfile { client_id, reply }) => { - let r = fetch_client_profile(&mut con, client_id).await; + let r = fetch_client_profile( + &mut con, + client_id, + &voice_in_tx, + &chat_tx, + &activity_tx, + &delta_tx, + &mut pending_moves, + &mut voice_activity, + ) + .await; let _ = reply.send(r); } Ok(Request::Disconnect(reply)) => { @@ -947,6 +819,154 @@ async fn connection_task( } } +async fn handle_audio_stream_item( + voice_in_tx: &mpsc::Sender, + voice_activity: &mut HashMap, + buf: InAudioBuf, +) { + let from = packet_sender_id(&buf); + if let Some(from) = from { + voice_activity.insert(from, Instant::now()); + let inbound = InboundVoice { + from_client: from, + packet: buf, + }; + match send_with_timeout(voice_in_tx, inbound, INBOUND_VOICE_SEND_TIMEOUT).await { + Ok(()) => {} + Err(SendTimeoutError::Timeout(_)) => { + warn!( + target: "chanora_protocol", + from_client = from, + timeout_ms = INBOUND_VOICE_SEND_TIMEOUT.as_millis() as u64, + "inbound voice queue stayed full; dropping packet" + ); + } + Err(SendTimeoutError::Closed(_)) => { + warn!( + target: "chanora_protocol", + from_client = from, + "inbound voice consumer closed; dropping packet" + ); + } + } + } +} + +fn handle_non_audio_stream_item( + con: &Connection, + item: StreamItem, + chat_tx: &mpsc::Sender, + activity_tx: &mpsc::Sender, + delta_tx: &mpsc::Sender, + pending_moves: &mut PendingMoves, +) { + match item { + StreamItem::BookEvents(events) => { + for ev in events { + if let tsclientlib::events::Event::PropertyChanged { + id: ts_bookkeeping::events::PropertyId::ClientChannel(client_id), + .. + } = &ev + { + let own_client = con.get_state().ok().map(|state| state.own_client); + if let Some(state) = con.get_state().ok() { + if let Some(client) = state.clients.get(client_id) { + let _ = delta_tx.try_send(ProtocolDelta::ClientMoved { + client_id: client_id.0 as u64, + new_channel_id: client.channel.0, + }); + } + } + if own_client == Some(*client_id) { + let current_channel = con.get_state().ok().and_then(|state| { + state.clients.get(client_id).map(|client| client.channel.0) + }); + if let Some(current_channel) = current_channel { + let matched: Vec = pending_moves + .iter() + .filter_map(|(handle, (target_channel, _, _))| { + if *target_channel == current_channel { + Some(*handle) + } else { + None + } + }) + .collect(); + for handle in matched { + if let Some((_, reply, _)) = pending_moves.remove(&handle) { + info!( + target: "chanora_protocol", + channel_id = current_channel, + "client_move resolved by authoritative self channel change" + ); + if let Some(reply) = reply { + let _ = reply.send(Ok(())); + } + } + } + } + } + } + + if let Some(activity) = format_server_activity(con, &ev) { + let _ = activity_tx.try_send(ServerActivity { message: activity }); + } + + forward_delta(con, &ev, delta_tx); + + if let tsclientlib::events::Event::Message { + target, + invoker, + message, + } = ev + { + let mapped = match target { + tsclientlib::MessageTarget::Server => MessageTarget::Server, + tsclientlib::MessageTarget::Channel => MessageTarget::Channel, + tsclientlib::MessageTarget::Client(id) => MessageTarget::Client(id.0 as u64), + tsclientlib::MessageTarget::Poke(id) => MessageTarget::Poke(id.0 as u64), + }; + let _ = chat_tx.try_send(ChatMessage { + sender_id: ClientId(invoker.id.0 as u64), + sender_name: sanitize(&invoker.name), + message: sanitize(&message), + target: mapped, + }); + } + } + } + StreamItem::MessageResult(handle, result) => { + if let Some((_target_channel, reply, _deadline)) = pending_moves.remove(&handle) { + let mapped = match result { + Ok(()) => Ok(()), + Err(cmd_err) => { + let code = cmd_err.error as u32; + let message = cmd_err.error.to_string(); + info!( + target: "chanora_protocol", + code, + message = %message, + "server rejected client_move" + ); + Err(ProtocolError::ServerRejected { code, message }) + } + }; + if let Some(reply) = reply { + let _ = reply.send(mapped); + } else if let Err(err) = mapped { + info!( + target: "chanora_protocol", + error = %err, + "client_move completed in background with error" + ); + } + } + } + StreamItem::Audio(_) => unreachable!("audio handled separately"), + _ => {} + } +} + async fn resolve_server_socket(address: &str) -> Result { let resolver = ChanoraResolver::new().map_err(|err| ProtocolError::DnsFailed { host: address.to_string(), @@ -1095,37 +1115,132 @@ fn send_text_to_mode( async fn fetch_client_profile( con: &mut Connection, client_id: u64, + voice_in_tx: &mpsc::Sender, + chat_tx: &mpsc::Sender, + activity_tx: &mpsc::Sender, + delta_tx: &mpsc::Sender, + pending_moves: &mut PendingMoves, + voice_activity: &mut HashMap, ) -> Result { let target_id = TsClientId(client_id as u16); - let (database_id, uid_b64) = { + + let (database_id, uid_b64, has_optional, has_connection, is_own, needs_server_groups, needs_channel_groups) = { let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; - let client = find_client_by_id(state.clients.values(), client_id)?; + let client = state + .clients + .get(&target_id) + .ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?; ( client.database_id, client.uid.as_ref().map(|uid| uid_to_b64(uid.as_ref())), + client.optional_data.is_some(), + client.connection_data.is_some(), + state.own_client == target_id, + state.server_groups.is_empty(), + state.channel_groups.is_empty(), ) }; - let _ = request_messages(con, build_command("servergrouplist", &[], &[])).await; - let _ = request_messages(con, build_command("channelgrouplist", &[], &[])).await; - let _ = request_messages( - con, - build_command( - "clientgetvariables", - &[("clid", client_id.to_string())], - &[], - ), - ) - .await; - let _ = request_messages( - con, - build_command("getconnectioninfo", &[("clid", client_id.to_string())], &[]), - ) - .await; + let refresh_plan = client_profile_refresh_plan( + is_own, + has_optional, + has_connection, + needs_server_groups, + needs_channel_groups, + ); + + if refresh_plan.needs_server_groups { + let _ = request_messages( + con, + build_command("servergrouplist", &[], &[]), + voice_in_tx, + chat_tx, + activity_tx, + delta_tx, + pending_moves, + voice_activity, + ) + .await; + } + if refresh_plan.needs_channel_groups { + let _ = request_messages( + con, + build_command("channelgrouplist", &[], &[]), + voice_in_tx, + chat_tx, + activity_tx, + delta_tx, + pending_moves, + voice_activity, + ) + .await; + } + if refresh_plan.needs_client_variables { + if let Err(e) = request_messages( + con, + build_command( + "clientgetvariables", + &[("clid", client_id.to_string())], + &[], + ), + voice_in_tx, + chat_tx, + activity_tx, + delta_tx, + pending_moves, + voice_activity, + ) + .await + { + warn!( + target: "chanora_protocol", + client_id, + error = %e, + "clientgetvariables failed" + ); + } + } + if refresh_plan.needs_connection_info { + if let Err(e) = request_messages( + con, + build_command("getconnectioninfo", &[("clid", client_id.to_string())], &[]), + voice_in_tx, + chat_tx, + activity_tx, + delta_tx, + pending_moves, + voice_activity, + ) + .await + { + warn!( + target: "chanora_protocol", + client_id, + error = %e, + "getconnectioninfo failed (ping/packet_loss will be unavailable)" + ); + } + } + + let db_info = if refresh_plan.needs_client_db_info { + request_client_db_info( + con, + database_id, + voice_in_tx, + chat_tx, + activity_tx, + delta_tx, + pending_moves, + voice_activity, + ) + .await + .ok() + } else { + None + }; - let db_info = request_client_db_info(con, database_id).await.ok(); let state = con .get_state() .map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?; @@ -1133,8 +1248,14 @@ async fn fetch_client_profile( .clients .get(&target_id) .ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?; + let optional = client.optional_data.as_ref(); let connection = client.connection_data.as_ref(); + let net_stats: Option<&ConnectionStats> = if is_own { + con.get_network_stats().ok() + } else { + None + }; let server_group_names: HashMap<_, _> = state .server_groups .iter() @@ -1182,22 +1303,19 @@ async fn fetch_client_profile( .or_else(|| db_info.as_ref().map(|info| info.created.unix_timestamp())), last_connected_unix_seconds: optional .map(|info| info.last_connected.unix_timestamp()) - .or_else(|| { - db_info - .as_ref() - .map(|info| info.last_connected.unix_timestamp()) - }), + .or_else(|| db_info.as_ref().map(|info| info.last_connected.unix_timestamp())), connections_total: optional .map(|info| u64::from(info.connections_total)) - .or_else(|| { - db_info - .as_ref() - .map(|info| u64::from(info.connections_total)) - }), + .or_else(|| db_info.as_ref().map(|info| u64::from(info.connections_total))), online_seconds: connection .and_then(|info| info.connected_time.map(|duration| duration.whole_seconds())), idle_milliseconds: connection.map(|info| duration_millis(info.idle_time)), - ping_milliseconds: connection.and_then(|info| info.ping.map(duration_millis)), + ping_milliseconds: net_stats + .and_then(|s| std_duration_millis(s.rtt)) + .or_else(|| connection.and_then(|info| info.ping.map(duration_millis))), + ping_deviation_milliseconds: net_stats + .and_then(|s| std_duration_millis(s.rtt_dev)) + .or_else(|| connection.and_then(|info| info.ping_deviation.map(duration_millis))), client_address: connection .and_then(|info| info.client_address.map(|address| address.to_string())) .unwrap_or_default(), @@ -1219,10 +1337,16 @@ async fn fetch_client_profile( bytes_uploaded_total: optional .map(|info| info.bytes_uploaded_total) .or_else(|| db_info.as_ref().map(|info| info.bytes_uploaded_total)), - packet_loss_client_to_server_total: connection - .map(|info| info.client_to_server_packetloss_total), - packet_loss_server_to_client_total: connection - .and_then(|info| info.server_to_client_packetloss_total), + packet_loss_client_to_server_total: net_stats + .map(|s| s.get_packetloss()) + .or_else(|| { + connection.map(|info| info.client_to_server_packetloss_total) + }), + packet_loss_server_to_client_total: net_stats + .map(|s| s.get_packetloss_s2c_total()) + .or_else(|| { + connection.and_then(|info| info.server_to_client_packetloss_total) + }), }) } @@ -1237,21 +1361,63 @@ fn build_command(name: &str, args: &[(&str, String)], flags: &[&str]) -> OutComm command } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ClientProfileRefreshPlan { + needs_server_groups: bool, + needs_channel_groups: bool, + needs_client_variables: bool, + needs_connection_info: bool, + needs_client_db_info: bool, +} + +fn client_profile_refresh_plan( + is_own: bool, + has_optional: bool, + has_connection: bool, + needs_server_groups: bool, + needs_channel_groups: bool, +) -> ClientProfileRefreshPlan { + ClientProfileRefreshPlan { + needs_server_groups, + needs_channel_groups, + needs_client_variables: !has_optional || !is_own, + needs_connection_info: !has_connection || !is_own, + needs_client_db_info: !has_optional || !is_own, + } +} + async fn request_messages( con: &mut Connection, command: OutCommand, + voice_in_tx: &mpsc::Sender, + chat_tx: &mpsc::Sender, + activity_tx: &mpsc::Sender, + delta_tx: &mpsc::Sender, + pending_moves: &mut PendingMoves, + voice_activity: &mut HashMap, ) -> Result, ProtocolError> { let handle = command .send_with_result(con) .map_err(|e| ProtocolError::Backend(format!("send command: {e}")))?; let mut messages = Vec::new(); + let deadline = Instant::now() + PROFILE_REFRESH_RESULT_TIMEOUT; loop { - let item = con - .events() - .next() - .await - .ok_or_else(|| ProtocolError::Lost("event stream ended".to_string()))? - .map_err(|e| ProtocolError::Lost(e.to_string()))?; + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return Err(ProtocolError::Backend( + "profile refresh command timed out".to_string(), + )); + } + let item = { + let mut stream = con.events(); + tokio::time::timeout(remaining, stream.next()) + .await + .map_err(|_| { + ProtocolError::Backend("profile refresh command timed out".to_string()) + })? + .ok_or_else(|| ProtocolError::Lost("event stream ended".to_string()))? + .map_err(|e| ProtocolError::Lost(e.to_string()))? + }; match item { StreamItem::MessageEvent(message) => messages.push(message), StreamItem::MessageResult(reply, status) if reply == handle => { @@ -1261,7 +1427,17 @@ async fn request_messages( })?; return Ok(messages); } - _ => {} + StreamItem::Audio(buf) => { + handle_audio_stream_item(voice_in_tx, voice_activity, buf).await; + } + other => handle_non_audio_stream_item( + con, + other, + chat_tx, + activity_tx, + delta_tx, + pending_moves, + ), } } } @@ -1269,10 +1445,22 @@ async fn request_messages( async fn request_client_db_info( con: &mut Connection, dbid: tsclientlib::ClientDbId, + voice_in_tx: &mpsc::Sender, + chat_tx: &mpsc::Sender, + activity_tx: &mpsc::Sender, + delta_tx: &mpsc::Sender, + pending_moves: &mut PendingMoves, + voice_activity: &mut HashMap, ) -> Result { let messages = request_messages( con, build_command("clientdbinfo", &[("cldbid", dbid.0.to_string())], &[]), + voice_in_tx, + chat_tx, + activity_tx, + delta_tx, + pending_moves, + voice_activity, ) .await?; for message in messages { @@ -1310,6 +1498,10 @@ fn duration_millis(duration: time::Duration) -> i64 { .clamp(i64::MIN as i128, i64::MAX as i128) as i64 } +fn std_duration_millis(duration: std::time::Duration) -> Option { + duration.as_millis().try_into().ok() +} + fn uid_to_b64(uid: &tsclientlib::Uid) -> String { BASE64_STANDARD.encode(&uid.0) } @@ -1709,8 +1901,9 @@ const _: () = { #[cfg(test)] mod tests { use super::{ - is_server_query_client_type, send_with_timeout, server_socket_from_config, - sort_channels_tree_by, ConnectConfig, SendTimeoutError, + client_profile_refresh_plan, is_server_query_client_type, send_with_timeout, + server_socket_from_config, sort_channels_tree_by, std_duration_millis, + ConnectConfig, SendTimeoutError, }; use std::time::Duration; use tokio::sync::mpsc; @@ -1772,6 +1965,41 @@ mod tests { assert_eq!(server_socket_from_config(&cfg), Some(addr)); } + #[test] + fn std_duration_millis_returns_none_when_duration_exceeds_i64() { + let overflow = Duration::from_millis(i64::MAX as u64) + Duration::from_millis(1); + + assert_eq!(std_duration_millis(overflow), None); + } + + #[test] + fn non_self_profile_refresh_requests_variables_connection_and_db_info() { + let plan = client_profile_refresh_plan(false, true, true, false, false); + + assert!(plan.needs_client_variables); + assert!(plan.needs_connection_info); + assert!(plan.needs_client_db_info); + } + + #[test] + fn own_profile_refresh_skips_complete_optional_and_connection_requests() { + let plan = client_profile_refresh_plan(true, true, true, false, false); + + assert!(!plan.needs_server_groups); + assert!(!plan.needs_channel_groups); + assert!(!plan.needs_client_variables); + assert!(!plan.needs_connection_info); + assert!(!plan.needs_client_db_info); + } + + #[test] + fn own_profile_refresh_requests_missing_group_lists() { + let plan = client_profile_refresh_plan(true, true, true, true, true); + + assert!(plan.needs_server_groups); + assert!(plan.needs_channel_groups); + } + #[test] fn channel_sort_linked_list_under_one_parent() { // Server emits four root-level channels in arbitrary HashMap diff --git a/crates/chanora_protocol/src/dto.rs b/crates/chanora_protocol/src/dto.rs index 168ead8..ca1c73b 100644 --- a/crates/chanora_protocol/src/dto.rs +++ b/crates/chanora_protocol/src/dto.rs @@ -120,6 +120,8 @@ pub struct ClientProfile { pub idle_milliseconds: Option, /// Current ping in milliseconds, when visible. pub ping_milliseconds: Option, + /// Current ping deviation in milliseconds, when visible. + pub ping_deviation_milliseconds: Option, /// Client address. Empty when permission-gated. pub client_address: String, /// Resolved server group names for the online client.