811 lines
28 KiB
Rust
811 lines
28 KiB
Rust
//! # `chanora_state`
|
|
//!
|
|
//! Authoritative client-side mirror of server state: server identity,
|
|
//! channel tree, client list, our own connection state. Owns the
|
|
//! reducers that fold protocol events into a snapshot, and the
|
|
//! deltas that the bridge publishes to Flutter view-models
|
|
//! (per SAD 7.2 and SDD 5).
|
|
//!
|
|
//! ## Reducer contract
|
|
//!
|
|
//! Reducers mutate the caller-owned state passed by `&mut` and return a
|
|
//! `Reduction` containing only the emitted `Delta` values. The caller
|
|
//! (typically `chanora_core`) owns state storage, publishes deltas to the
|
|
//! bridge, and executes any side effects.
|
|
//!
|
|
//! This design satisfies SRS-056 (deterministic deltas), SRS-057
|
|
//! (per-connection ordering), and SRS-058 (reducer functions).
|
|
|
|
#![forbid(unsafe_code)]
|
|
#![warn(missing_docs)]
|
|
|
|
pub mod channel_join;
|
|
|
|
use std::collections::HashMap;
|
|
|
|
use chanora_protocol::{ChannelId, ChannelInfo, ChatMessage, ClientId, ClientInfo, ServerSnapshot};
|
|
use thiserror::Error;
|
|
|
|
/// Errors raised while reducing protocol events into state or
|
|
/// producing deltas for the bridge.
|
|
#[derive(Debug, Error)]
|
|
pub enum StateError {
|
|
/// Reducer received an event referencing an unknown entity.
|
|
#[error("unknown entity: {0}")]
|
|
Unknown(&'static str),
|
|
/// A reducer invariant was violated (e.g. two clients claiming the same id).
|
|
#[error("state invariant: {0}")]
|
|
Invariant(&'static str),
|
|
}
|
|
|
|
/// Coarse connection lifecycle states surfaced to the bridge.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum ConnectionState {
|
|
/// No connection attempt yet, or fully disconnected.
|
|
Idle,
|
|
/// Handshake in progress.
|
|
Connecting,
|
|
/// Initial state snapshot has been received; ready for use.
|
|
Ready,
|
|
/// Connection lost; reconnecting.
|
|
Reconnecting,
|
|
/// Connection lost; will not auto-reconnect at this stage.
|
|
Lost,
|
|
}
|
|
|
|
/// Authoritative mirror of a connected server's published state.
|
|
///
|
|
/// Constructed from a [`ServerSnapshot`] and updated through delta
|
|
/// events. Owns the channel tree, client list, and server metadata.
|
|
/// Satisfies SRS-054.
|
|
#[derive(Debug, Clone)]
|
|
pub struct ServerState {
|
|
/// Retained connection lifecycle state.
|
|
pub connection_state: ConnectionState,
|
|
/// Server name from the most recent snapshot.
|
|
pub server_name: String,
|
|
/// Welcome message / banner (server-provided markup).
|
|
pub welcome_message: String,
|
|
/// Server platform string.
|
|
pub platform: String,
|
|
/// Server version string.
|
|
pub version: String,
|
|
channels: HashMap<u64, ChannelInfo>,
|
|
clients: HashMap<u64, ClientInfo>,
|
|
channel_order: Vec<u64>,
|
|
client_order: Vec<u64>,
|
|
/// Our own client id, as the server reported it.
|
|
pub own_client_id: u64,
|
|
}
|
|
|
|
impl ServerState {
|
|
/// Build state from the initial full snapshot. Satisfies SRS-055.
|
|
pub fn from_snapshot(snapshot: ServerSnapshot) -> Self {
|
|
let snapshot = normalize_snapshot(snapshot);
|
|
let channels = snapshot
|
|
.channels
|
|
.iter()
|
|
.map(|c| (c.id.0, c.clone()))
|
|
.collect();
|
|
let clients = snapshot
|
|
.clients
|
|
.iter()
|
|
.map(|c| (c.id.0, c.clone()))
|
|
.collect();
|
|
let channel_order: Vec<u64> = snapshot.channels.iter().map(|c| c.id.0).collect();
|
|
let mut client_order: Vec<u64> = snapshot.clients.iter().map(|c| c.id.0).collect();
|
|
client_order.sort_unstable();
|
|
client_order.dedup();
|
|
Self {
|
|
connection_state: ConnectionState::Ready,
|
|
server_name: snapshot.server_name,
|
|
welcome_message: snapshot.welcome_message,
|
|
platform: snapshot.platform,
|
|
version: snapshot.version,
|
|
channels,
|
|
clients,
|
|
channel_order,
|
|
client_order,
|
|
own_client_id: snapshot.own_client_id,
|
|
}
|
|
}
|
|
|
|
/// Replace state with a fresh snapshot (post-reconnect). Satisfies SRS-059.
|
|
pub fn replace_from_snapshot(&mut self, snapshot: ServerSnapshot) {
|
|
*self = Self::from_snapshot(snapshot);
|
|
}
|
|
|
|
/// Look up a channel by id.
|
|
pub fn channel(&self, id: ChannelId) -> Option<&ChannelInfo> {
|
|
self.channels.get(&id.0)
|
|
}
|
|
|
|
/// Look up a client by id.
|
|
pub fn client(&self, id: ClientId) -> Option<&ClientInfo> {
|
|
self.clients.get(&id.0)
|
|
}
|
|
|
|
/// All channels in protocol snapshot order. Live inserts are appended
|
|
/// until the next full snapshot re-normalizes order.
|
|
pub fn channels(&self) -> impl Iterator<Item = &ChannelInfo> {
|
|
self.channel_order
|
|
.iter()
|
|
.filter_map(|id| self.channels.get(id))
|
|
}
|
|
|
|
/// All clients in stable reducer order.
|
|
///
|
|
/// Full snapshots normalize clients by id to avoid inheriting upstream
|
|
/// HashMap iteration order. Live inserts are appended in arrival order
|
|
/// until the next full snapshot re-normalizes the list.
|
|
pub fn clients(&self) -> impl Iterator<Item = &ClientInfo> {
|
|
self.client_order
|
|
.iter()
|
|
.filter_map(|id| self.clients.get(id))
|
|
}
|
|
|
|
/// Number of channels.
|
|
pub fn channel_count(&self) -> usize {
|
|
self.channels.len()
|
|
}
|
|
|
|
/// Number of clients.
|
|
pub fn client_count(&self) -> usize {
|
|
self.clients.len()
|
|
}
|
|
|
|
/// The channel our own client is currently in.
|
|
pub fn own_channel(&self) -> Option<&ChannelInfo> {
|
|
self.client(ClientId(self.own_client_id))
|
|
.and_then(|c| self.channel(c.channel))
|
|
}
|
|
|
|
/// Clients in a given channel.
|
|
pub fn clients_in_channel(&self, channel_id: ChannelId) -> impl Iterator<Item = &ClientInfo> {
|
|
self.client_order
|
|
.iter()
|
|
.filter_map(|id| self.clients.get(id))
|
|
.filter(move |c| c.channel == channel_id)
|
|
}
|
|
}
|
|
|
|
fn normalize_snapshot(snapshot: ServerSnapshot) -> ServerSnapshot {
|
|
let mut channels = HashMap::new();
|
|
let mut channel_order = Vec::new();
|
|
for channel in snapshot.channels {
|
|
if !channels.contains_key(&channel.id.0) {
|
|
channel_order.push(channel.id.0);
|
|
}
|
|
channels.insert(channel.id.0, channel);
|
|
}
|
|
let mut clients = HashMap::new();
|
|
for client in snapshot.clients {
|
|
clients.insert(client.id.0, client);
|
|
}
|
|
let mut client_order: Vec<u64> = clients.keys().copied().collect();
|
|
client_order.sort_unstable();
|
|
|
|
ServerSnapshot {
|
|
server_name: snapshot.server_name,
|
|
welcome_message: snapshot.welcome_message,
|
|
platform: snapshot.platform,
|
|
version: snapshot.version,
|
|
channels: channel_order
|
|
.into_iter()
|
|
.filter_map(|id| channels.remove(&id))
|
|
.collect(),
|
|
clients: client_order
|
|
.into_iter()
|
|
.filter_map(|id| clients.remove(&id))
|
|
.collect(),
|
|
own_client_id: snapshot.own_client_id,
|
|
}
|
|
}
|
|
|
|
/// A change to the server state that the bridge should publish to
|
|
/// Flutter. Deltas are cheap to construct and carry only the
|
|
/// information that changed.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub enum Delta {
|
|
/// A full snapshot was applied; the entire UI should refresh.
|
|
SnapshotApplied(ServerSnapshot),
|
|
/// A channel was added or updated.
|
|
ChannelUpserted(ChannelInfo),
|
|
/// A channel was removed.
|
|
ChannelRemoved(ChannelId),
|
|
/// A client joined or was updated (includes moves).
|
|
ClientUpserted(ClientInfo),
|
|
/// A client left.
|
|
ClientRemoved(ClientId),
|
|
/// A chat message arrived.
|
|
ChatMessage(ChatMessage),
|
|
/// A client's speaking state changed.
|
|
VoiceActivityChanged {
|
|
/// Client whose voice activity changed.
|
|
client_id: ClientId,
|
|
/// Whether voice activity is currently observed.
|
|
is_speaking: bool,
|
|
},
|
|
/// The connection state changed.
|
|
ConnectionStateChanged(ConnectionState),
|
|
}
|
|
|
|
/// Events that flow from the protocol layer into the state reducer.
|
|
/// Each variant carries all data the reducer needs; no external
|
|
/// lookups are required.
|
|
#[derive(Debug, Clone)]
|
|
pub enum StateEvent {
|
|
/// Initial or post-reconnect snapshot arrived.
|
|
Snapshot(ServerSnapshot),
|
|
/// A new channel appeared or an existing channel was updated.
|
|
ChannelChanged(ChannelInfo),
|
|
/// A channel was deleted.
|
|
ChannelDeleted(ChannelId),
|
|
/// A client joined or was updated (including channel moves and
|
|
/// mute/speaking state changes).
|
|
ClientChanged(ClientInfo),
|
|
/// A client left the server.
|
|
ClientLeft(ClientId),
|
|
/// A text message was received.
|
|
ChatReceived(ChatMessage),
|
|
/// A client's speaking state changed.
|
|
VoiceActivityChanged {
|
|
/// Client whose voice activity changed.
|
|
client_id: ClientId,
|
|
/// Whether voice activity is currently observed.
|
|
is_speaking: bool,
|
|
},
|
|
/// Connection lifecycle transition.
|
|
ConnectionChanged(ConnectionState),
|
|
/// Reconnect started; stale state must be discarded.
|
|
ReconnectStarted,
|
|
}
|
|
|
|
/// Result of applying a single event through the reducer.
|
|
#[derive(Debug, Clone)]
|
|
pub struct Reduction {
|
|
/// Deltas describing what changed, in application order.
|
|
pub deltas: Vec<Delta>,
|
|
}
|
|
|
|
/// Apply a [`StateEvent`] to optional [`ServerState`], returning
|
|
/// the resulting [`Reduction`].
|
|
///
|
|
/// The state is taken by `&mut` so the caller retains ownership.
|
|
/// When `state` is `None` (no active connection), most events are
|
|
/// ignored except `ConnectionChanged` and `Snapshot` (which creates
|
|
/// the state).
|
|
///
|
|
/// Deterministic: the same `(state, event)` pair always produces the
|
|
/// same `Reduction`. Satisfies SRS-056.
|
|
pub fn reduce(state: &mut Option<ServerState>, event: StateEvent) -> Reduction {
|
|
match event {
|
|
StateEvent::ConnectionChanged(cs) => {
|
|
if let Some(s) = state {
|
|
s.connection_state = cs;
|
|
}
|
|
Reduction {
|
|
deltas: vec![Delta::ConnectionStateChanged(cs)],
|
|
}
|
|
}
|
|
|
|
StateEvent::ReconnectStarted => {
|
|
*state = None;
|
|
Reduction {
|
|
deltas: vec![Delta::ConnectionStateChanged(ConnectionState::Reconnecting)],
|
|
}
|
|
}
|
|
|
|
StateEvent::Snapshot(snap) => {
|
|
let normalized = normalize_snapshot(snap);
|
|
*state = Some(ServerState::from_snapshot(normalized.clone()));
|
|
Reduction {
|
|
deltas: vec![
|
|
Delta::ConnectionStateChanged(ConnectionState::Ready),
|
|
Delta::SnapshotApplied(normalized),
|
|
],
|
|
}
|
|
}
|
|
|
|
StateEvent::ChannelChanged(info) => match state {
|
|
Some(s) if s.connection_state == ConnectionState::Ready => {
|
|
if !s.channels.contains_key(&info.id.0) {
|
|
s.channel_order.push(info.id.0);
|
|
}
|
|
s.channels.insert(info.id.0, info.clone());
|
|
Reduction {
|
|
deltas: vec![Delta::ChannelUpserted(info)],
|
|
}
|
|
}
|
|
_ => Reduction { deltas: vec![] },
|
|
},
|
|
|
|
StateEvent::ChannelDeleted(id) => match state {
|
|
Some(s) if s.connection_state == ConnectionState::Ready => {
|
|
let removed_clients: Vec<u64> = s
|
|
.client_order
|
|
.iter()
|
|
.copied()
|
|
.filter(|client_id| {
|
|
s.clients
|
|
.get(client_id)
|
|
.is_some_and(|client| client.channel == id)
|
|
})
|
|
.collect();
|
|
for client_id in &removed_clients {
|
|
s.clients.remove(client_id);
|
|
}
|
|
s.client_order
|
|
.retain(|existing| !removed_clients.contains(existing));
|
|
let removed_channel = s.channels.remove(&id.0).is_some();
|
|
s.channel_order.retain(|existing| *existing != id.0);
|
|
let mut deltas: Vec<Delta> = removed_clients
|
|
.into_iter()
|
|
.map(|client_id| Delta::ClientRemoved(ClientId(client_id)))
|
|
.collect();
|
|
if removed_channel {
|
|
deltas.push(Delta::ChannelRemoved(id));
|
|
}
|
|
Reduction { deltas }
|
|
}
|
|
_ => Reduction { deltas: vec![] },
|
|
},
|
|
|
|
StateEvent::ClientChanged(info) => match state {
|
|
Some(s) if s.connection_state == ConnectionState::Ready => {
|
|
if !s.clients.contains_key(&info.id.0) {
|
|
s.client_order.push(info.id.0);
|
|
}
|
|
s.clients.insert(info.id.0, info.clone());
|
|
Reduction {
|
|
deltas: vec![Delta::ClientUpserted(info)],
|
|
}
|
|
}
|
|
_ => Reduction { deltas: vec![] },
|
|
},
|
|
|
|
StateEvent::ClientLeft(id) => match state {
|
|
Some(s) if s.connection_state == ConnectionState::Ready => {
|
|
s.clients.remove(&id.0);
|
|
s.client_order.retain(|existing| *existing != id.0);
|
|
Reduction {
|
|
deltas: vec![Delta::ClientRemoved(id)],
|
|
}
|
|
}
|
|
_ => Reduction { deltas: vec![] },
|
|
},
|
|
|
|
StateEvent::ChatReceived(msg) => match state {
|
|
Some(s) if s.connection_state == ConnectionState::Ready => Reduction {
|
|
deltas: vec![Delta::ChatMessage(msg)],
|
|
},
|
|
_ => Reduction { deltas: vec![] },
|
|
},
|
|
|
|
StateEvent::VoiceActivityChanged {
|
|
client_id,
|
|
is_speaking,
|
|
} => match state {
|
|
Some(s) if s.connection_state == ConnectionState::Ready => {
|
|
if let Some(client) = s.clients.get_mut(&client_id.0) {
|
|
client.is_speaking = is_speaking;
|
|
Reduction {
|
|
deltas: vec![Delta::VoiceActivityChanged {
|
|
client_id,
|
|
is_speaking,
|
|
}],
|
|
}
|
|
} else {
|
|
Reduction { deltas: vec![] }
|
|
}
|
|
}
|
|
_ => Reduction { deltas: vec![] },
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Replace server state with a fresh snapshot after a reconnect.
|
|
/// Discards all previous state unconditionally (SRS-059).
|
|
pub fn reduce_reconnect_snapshot(
|
|
state: &mut Option<ServerState>,
|
|
snap: ServerSnapshot,
|
|
) -> Reduction {
|
|
reduce(state, StateEvent::Snapshot(snap))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use chanora_protocol::MessageTarget;
|
|
|
|
fn sample_channel(id: u64) -> ChannelInfo {
|
|
ChannelInfo {
|
|
id: ChannelId(id),
|
|
parent: ChannelId(0),
|
|
name: format!("channel-{id}"),
|
|
order: 0,
|
|
has_password: false,
|
|
needed_talk_power: None,
|
|
}
|
|
}
|
|
|
|
fn sample_client(id: u64, channel: u64) -> ClientInfo {
|
|
ClientInfo {
|
|
id: ClientId(id),
|
|
channel: ChannelId(channel),
|
|
name: format!("client-{id}"),
|
|
input_muted: false,
|
|
output_muted: false,
|
|
is_speaking: false,
|
|
is_server_query: false,
|
|
talk_power: 0,
|
|
talk_power_granted: false,
|
|
}
|
|
}
|
|
|
|
fn sample_snapshot() -> ServerSnapshot {
|
|
ServerSnapshot {
|
|
server_name: "Test Server".into(),
|
|
welcome_message: String::new(),
|
|
platform: "Linux".into(),
|
|
version: "3.13".into(),
|
|
channels: vec![sample_channel(1), sample_channel(2)],
|
|
clients: vec![sample_client(10, 1)],
|
|
own_client_id: 10,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn snapshot_creates_state() {
|
|
let mut state = None;
|
|
let snapshot = sample_snapshot();
|
|
let reduction = reduce(&mut state, StateEvent::Snapshot(snapshot.clone()));
|
|
assert!(state.is_some());
|
|
let s = state.as_ref().unwrap();
|
|
assert_eq!(s.connection_state, ConnectionState::Ready);
|
|
assert_eq!(s.channel_count(), 2);
|
|
assert_eq!(s.client_count(), 1);
|
|
assert_eq!(s.own_client_id, 10);
|
|
assert_eq!(
|
|
reduction.deltas,
|
|
vec![
|
|
Delta::ConnectionStateChanged(ConnectionState::Ready),
|
|
Delta::SnapshotApplied(snapshot),
|
|
]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn channel_upsert_adds_and_updates() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let ch = ChannelInfo {
|
|
id: ChannelId(3),
|
|
parent: ChannelId(1),
|
|
name: "new-channel".into(),
|
|
order: 1,
|
|
has_password: true,
|
|
needed_talk_power: Some(50),
|
|
};
|
|
let reduction = reduce(&mut state, StateEvent::ChannelChanged(ch.clone()));
|
|
let s = state.as_ref().unwrap();
|
|
assert_eq!(s.channel_count(), 3);
|
|
assert!(s.channel(ChannelId(3)).is_some());
|
|
assert!(matches!(&reduction.deltas[..], [Delta::ChannelUpserted(_)]));
|
|
let updated = ChannelInfo {
|
|
name: "renamed".into(),
|
|
..ch
|
|
};
|
|
reduce(&mut state, StateEvent::ChannelChanged(updated));
|
|
assert_eq!(
|
|
state.as_ref().unwrap().channel(ChannelId(3)).unwrap().name,
|
|
"renamed"
|
|
);
|
|
assert_eq!(state.as_ref().unwrap().channel_count(), 3);
|
|
}
|
|
|
|
#[test]
|
|
fn channel_delete_removes() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let reduction = reduce(&mut state, StateEvent::ChannelDeleted(ChannelId(2)));
|
|
let s = state.as_ref().unwrap();
|
|
assert_eq!(s.channel_count(), 1);
|
|
assert!(s.channel(ChannelId(2)).is_none());
|
|
assert!(matches!(&reduction.deltas[..], [Delta::ChannelRemoved(_)]));
|
|
}
|
|
|
|
#[test]
|
|
fn channel_delete_removes_clients_in_deleted_channel() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
reduce(&mut state, StateEvent::ClientChanged(sample_client(20, 2)));
|
|
reduce(&mut state, StateEvent::ClientChanged(sample_client(30, 2)));
|
|
|
|
let reduction = reduce(&mut state, StateEvent::ChannelDeleted(ChannelId(2)));
|
|
let s = state.as_ref().unwrap();
|
|
|
|
assert!(s.channel(ChannelId(2)).is_none());
|
|
assert!(s.client(ClientId(20)).is_none());
|
|
assert!(s.client(ClientId(30)).is_none());
|
|
assert_eq!(s.client_count(), 1);
|
|
assert_eq!(s.clients_in_channel(ChannelId(2)).count(), 0);
|
|
assert_eq!(
|
|
reduction.deltas,
|
|
vec![
|
|
Delta::ClientRemoved(ClientId(20)),
|
|
Delta::ClientRemoved(ClientId(30)),
|
|
Delta::ChannelRemoved(ChannelId(2)),
|
|
]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn client_upsert_adds_and_moves() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let new_client = sample_client(20, 2);
|
|
let reduction = reduce(&mut state, StateEvent::ClientChanged(new_client));
|
|
let s = state.as_ref().unwrap();
|
|
assert_eq!(s.client_count(), 2);
|
|
assert!(matches!(&reduction.deltas[..], [Delta::ClientUpserted(_)]));
|
|
let moved = ClientInfo {
|
|
channel: ChannelId(2),
|
|
..sample_client(10, 2)
|
|
};
|
|
reduce(&mut state, StateEvent::ClientChanged(moved));
|
|
assert_eq!(
|
|
state
|
|
.as_ref()
|
|
.unwrap()
|
|
.client(ClientId(10))
|
|
.unwrap()
|
|
.channel,
|
|
ChannelId(2)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn client_leave_removes() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let reduction = reduce(&mut state, StateEvent::ClientLeft(ClientId(10)));
|
|
assert_eq!(state.as_ref().unwrap().client_count(), 0);
|
|
assert!(matches!(&reduction.deltas[..], [Delta::ClientRemoved(_)]));
|
|
}
|
|
|
|
#[test]
|
|
fn reconnect_discards_stale_state() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
assert_eq!(state.as_ref().unwrap().channel_count(), 2);
|
|
let reduction = reduce(&mut state, StateEvent::ReconnectStarted);
|
|
assert!(state.is_none());
|
|
assert!(matches!(
|
|
&reduction.deltas[..],
|
|
[Delta::ConnectionStateChanged(ConnectionState::Reconnecting)]
|
|
));
|
|
let snap2 = ServerSnapshot {
|
|
server_name: "New Server".into(),
|
|
channels: vec![sample_channel(100)],
|
|
clients: vec![sample_client(200, 100)],
|
|
..sample_snapshot()
|
|
};
|
|
reduce_reconnect_snapshot(&mut state, snap2);
|
|
let s = state.as_ref().unwrap();
|
|
assert_eq!(s.server_name, "New Server");
|
|
assert_eq!(s.channel_count(), 1);
|
|
assert_eq!(s.client_count(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn deltas_ignored_when_disconnected() {
|
|
let mut state: Option<ServerState> = None;
|
|
let r1 = reduce(&mut state, StateEvent::ChannelChanged(sample_channel(1)));
|
|
assert!(r1.deltas.is_empty());
|
|
let r2 = reduce(&mut state, StateEvent::ClientChanged(sample_client(1, 1)));
|
|
assert!(r2.deltas.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn chat_message_requires_connected_state() {
|
|
let mut state: Option<ServerState> = None;
|
|
let msg = ChatMessage {
|
|
sender_id: ClientId(10),
|
|
sender_name: "Alice".into(),
|
|
message: "hello".into(),
|
|
target: MessageTarget::Channel,
|
|
};
|
|
let disconnected = reduce(&mut state, StateEvent::ChatReceived(msg.clone()));
|
|
assert!(disconnected.deltas.is_empty());
|
|
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
reduce(
|
|
&mut state,
|
|
StateEvent::ConnectionChanged(ConnectionState::Lost),
|
|
);
|
|
let lost = reduce(&mut state, StateEvent::ChatReceived(msg.clone()));
|
|
assert!(lost.deltas.is_empty());
|
|
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let reduction = reduce(&mut state, StateEvent::ChatReceived(msg));
|
|
assert!(matches!(&reduction.deltas[..], [Delta::ChatMessage(_)]));
|
|
}
|
|
|
|
#[test]
|
|
fn connection_state_change_produces_delta() {
|
|
let mut state: Option<ServerState> = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let reduction = reduce(
|
|
&mut state,
|
|
StateEvent::ConnectionChanged(ConnectionState::Connecting),
|
|
);
|
|
assert_eq!(
|
|
state.as_ref().unwrap().connection_state,
|
|
ConnectionState::Connecting
|
|
);
|
|
assert!(matches!(
|
|
&reduction.deltas[..],
|
|
[Delta::ConnectionStateChanged(ConnectionState::Connecting)]
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn snapshot_order_is_deterministic_and_live_inserts_append() {
|
|
let mut state = None;
|
|
let snapshot = ServerSnapshot {
|
|
channels: vec![sample_channel(20), sample_channel(10)],
|
|
clients: vec![sample_client(30, 20), sample_client(10, 10)],
|
|
..sample_snapshot()
|
|
};
|
|
reduce(&mut state, StateEvent::Snapshot(snapshot));
|
|
reduce(&mut state, StateEvent::ChannelChanged(sample_channel(15)));
|
|
reduce(&mut state, StateEvent::ClientChanged(sample_client(20, 2)));
|
|
let s = state.as_ref().unwrap();
|
|
let channel_ids: Vec<u64> = s.channels().map(|c| c.id.0).collect();
|
|
let client_ids: Vec<u64> = s.clients().map(|c| c.id.0).collect();
|
|
assert_eq!(channel_ids, vec![20, 10, 15]);
|
|
assert_eq!(client_ids, vec![10, 30, 20]);
|
|
}
|
|
|
|
#[test]
|
|
fn malformed_snapshot_duplicate_ids_are_deduplicated_in_order_vectors() {
|
|
let mut snapshot = sample_snapshot();
|
|
snapshot.channels.push(ChannelInfo {
|
|
name: "duplicate".into(),
|
|
..sample_channel(1)
|
|
});
|
|
snapshot.clients.push(ClientInfo {
|
|
name: "duplicate".into(),
|
|
..sample_client(10, 2)
|
|
});
|
|
|
|
let mut state = None;
|
|
let reduction = reduce(&mut state, StateEvent::Snapshot(snapshot));
|
|
let s = state.as_ref().unwrap();
|
|
|
|
assert_eq!(s.channel_count(), 2);
|
|
assert_eq!(s.channels().count(), 2);
|
|
assert_eq!(s.client_count(), 1);
|
|
assert_eq!(s.clients().count(), 1);
|
|
assert_eq!(s.channel(ChannelId(1)).unwrap().name, "duplicate");
|
|
assert_eq!(s.client(ClientId(10)).unwrap().channel, ChannelId(2));
|
|
let Delta::SnapshotApplied(emitted) = &reduction.deltas[1] else {
|
|
panic!("expected snapshot delta");
|
|
};
|
|
assert_eq!(emitted.channels.len(), 2);
|
|
assert_eq!(emitted.clients.len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn voice_activity_updates_existing_client_and_emits_delta() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let reduction = reduce(
|
|
&mut state,
|
|
StateEvent::VoiceActivityChanged {
|
|
client_id: ClientId(10),
|
|
is_speaking: true,
|
|
},
|
|
);
|
|
assert!(
|
|
state
|
|
.as_ref()
|
|
.unwrap()
|
|
.client(ClientId(10))
|
|
.unwrap()
|
|
.is_speaking
|
|
);
|
|
assert_eq!(
|
|
reduction.deltas,
|
|
vec![Delta::VoiceActivityChanged {
|
|
client_id: ClientId(10),
|
|
is_speaking: true,
|
|
}]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn voice_activity_for_unknown_client_is_ignored() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let reduction = reduce(
|
|
&mut state,
|
|
StateEvent::VoiceActivityChanged {
|
|
client_id: ClientId(999),
|
|
is_speaking: true,
|
|
},
|
|
);
|
|
assert!(reduction.deltas.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn own_channel_returns_correct_channel() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let own = state.as_ref().unwrap().own_channel();
|
|
assert!(own.is_some());
|
|
assert_eq!(own.unwrap().id, ChannelId(1));
|
|
}
|
|
|
|
#[test]
|
|
fn clients_in_channel_filters_correctly() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
reduce(&mut state, StateEvent::ClientChanged(sample_client(20, 1)));
|
|
reduce(&mut state, StateEvent::ClientChanged(sample_client(30, 2)));
|
|
let s = state.as_ref().unwrap();
|
|
assert_eq!(s.clients_in_channel(ChannelId(1)).count(), 2);
|
|
assert_eq!(s.clients_in_channel(ChannelId(2)).count(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn malformed_delete_is_tolerated() {
|
|
let mut state = None;
|
|
reduce(&mut state, StateEvent::Snapshot(sample_snapshot()));
|
|
let reduction = reduce(&mut state, StateEvent::ChannelDeleted(ChannelId(999)));
|
|
assert!(reduction.deltas.is_empty());
|
|
assert_eq!(state.as_ref().unwrap().channel_count(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn same_event_sequence_produces_same_state() {
|
|
let mut a: Option<ServerState> = None;
|
|
let mut b: Option<ServerState> = None;
|
|
let events: Vec<StateEvent> = vec![
|
|
StateEvent::Snapshot(sample_snapshot()),
|
|
StateEvent::ClientChanged(sample_client(20, 2)),
|
|
StateEvent::ChannelChanged(ChannelInfo {
|
|
id: ChannelId(3),
|
|
parent: ChannelId(1),
|
|
name: "x".into(),
|
|
order: 0,
|
|
has_password: false,
|
|
needed_talk_power: None,
|
|
}),
|
|
StateEvent::ClientLeft(ClientId(20)),
|
|
];
|
|
let mut deltas_a = Vec::new();
|
|
let mut deltas_b = Vec::new();
|
|
for e in &events {
|
|
deltas_a.extend(reduce(&mut a, e.clone()).deltas);
|
|
}
|
|
for e in &events {
|
|
deltas_b.extend(reduce(&mut b, e.clone()).deltas);
|
|
}
|
|
assert_eq!(
|
|
a.as_ref().unwrap().channel_count(),
|
|
b.as_ref().unwrap().channel_count()
|
|
);
|
|
assert_eq!(
|
|
a.as_ref().unwrap().client_count(),
|
|
b.as_ref().unwrap().client_count()
|
|
);
|
|
assert_eq!(
|
|
a.as_ref().unwrap().own_client_id,
|
|
b.as_ref().unwrap().own_client_id
|
|
);
|
|
assert_eq!(deltas_a, deltas_b);
|
|
}
|
|
}
|