phoenix_channel_runtime/
presence.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use serde_json::{Map, Value};
4use thiserror::Error;
5
6use crate::Frame;
7
8/// Current Phoenix Presence entries keyed by application-defined presence key.
9#[derive(Clone, Debug, Default, PartialEq)]
10pub struct PresenceState(pub BTreeMap<String, Presence>);
11
12impl PresenceState {
13    /// Decodes a `presence_state` JSON object.
14    pub fn from_value(value: &Value) -> Result<Self, PresenceError> {
15        let state = value.as_object().ok_or(PresenceError::InvalidState)?;
16        state
17            .iter()
18            .map(|(key, presence)| Ok((key.clone(), Presence::from_value(key, presence)?)))
19            .collect::<Result<BTreeMap<_, _>, _>>()
20            .map(Self)
21    }
22
23    /// Returns the Presence entry for `key`.
24    pub fn get(&self, key: &str) -> Option<&Presence> {
25        self.0.get(key)
26    }
27
28    /// Iterates over presence keys and entries in key order.
29    pub fn iter(&self) -> impl Iterator<Item = (&str, &Presence)> {
30        self.0.iter().map(|(key, value)| (key.as_str(), value))
31    }
32
33    /// Returns whether the state contains no presence entries.
34    pub fn is_empty(&self) -> bool {
35        self.0.is_empty()
36    }
37}
38
39/// A Presence entry and its active connection metadata.
40#[derive(Clone, Debug, PartialEq)]
41pub struct Presence {
42    /// Active metas, each identified by its `phx_ref` field.
43    pub metas: Vec<Map<String, Value>>,
44    /// Additional fields attached to the presence entry outside `metas`.
45    pub fields: Map<String, Value>,
46}
47
48impl Presence {
49    fn from_value(key: &str, value: &Value) -> Result<Self, PresenceError> {
50        let mut fields = value
51            .as_object()
52            .cloned()
53            .ok_or_else(|| PresenceError::InvalidPresence(key.to_owned()))?;
54        let metas = fields
55            .remove("metas")
56            .and_then(|value| value.as_array().cloned())
57            .ok_or_else(|| PresenceError::InvalidMetas(key.to_owned()))?
58            .into_iter()
59            .map(|meta| {
60                let meta = meta
61                    .as_object()
62                    .cloned()
63                    .ok_or_else(|| PresenceError::InvalidMeta(key.to_owned()))?;
64                meta_ref(key, &meta)?;
65                Ok(meta)
66            })
67            .collect::<Result<Vec<_>, PresenceError>>()?;
68        Ok(Self { metas, fields })
69    }
70}
71
72/// Presence joins and leaves produced by a state or diff synchronization.
73#[derive(Clone, Debug, Default, PartialEq)]
74pub struct PresenceDiff {
75    /// Entries or metas that joined.
76    pub joins: PresenceState,
77    /// Entries or metas that left.
78    pub leaves: PresenceState,
79}
80
81impl PresenceDiff {
82    /// Decodes a `presence_diff` JSON object.
83    pub fn from_value(value: &Value) -> Result<Self, PresenceError> {
84        let diff = value.as_object().ok_or(PresenceError::InvalidDiff)?;
85        let joins = diff.get("joins").ok_or(PresenceError::InvalidDiff)?;
86        let leaves = diff.get("leaves").ok_or(PresenceError::InvalidDiff)?;
87        Ok(Self {
88            joins: PresenceState::from_value(joins)?,
89            leaves: PresenceState::from_value(leaves)?,
90        })
91    }
92
93    fn extend(&mut self, other: Self) {
94        extend_state(&mut self.joins, other.joins);
95        extend_state(&mut self.leaves, other.leaves);
96    }
97}
98
99fn extend_state(target: &mut PresenceState, source: PresenceState) {
100    for (key, presence) in source.0 {
101        if let Some(existing) = target.0.get_mut(&key) {
102            existing.metas.extend(presence.metas);
103        } else {
104            target.0.insert(key, presence);
105        }
106    }
107}
108
109/// Result of applying a frame to a [`PresenceTracker`].
110#[derive(Clone, Debug, PartialEq)]
111pub enum PresenceUpdate {
112    /// The frame was not a configured Presence event.
113    Ignored,
114    /// A diff was queued until a matching full state arrives.
115    Pending,
116    /// State was updated with the returned joins and leaves.
117    Synced(PresenceDiff),
118}
119
120/// Applies Phoenix Presence state and diff frames in join-generation order.
121#[derive(Clone, Debug)]
122pub struct PresenceTracker {
123    state: PresenceState,
124    join_ref: Option<String>,
125    pending_diffs: Vec<PresenceDiff>,
126    state_event: String,
127    diff_event: String,
128}
129
130impl PresenceTracker {
131    /// Creates a tracker for `presence_state` and `presence_diff` events.
132    pub fn new() -> Self {
133        Self {
134            state: PresenceState::default(),
135            join_ref: None,
136            pending_diffs: Vec::new(),
137            state_event: "presence_state".into(),
138            diff_event: "presence_diff".into(),
139        }
140    }
141
142    /// Creates a tracker with application-specific state and diff event names.
143    pub fn with_events(state_event: impl Into<String>, diff_event: impl Into<String>) -> Self {
144        Self {
145            state: PresenceState::default(),
146            join_ref: None,
147            pending_diffs: Vec::new(),
148            state_event: state_event.into(),
149            diff_event: diff_event.into(),
150        }
151    }
152
153    /// Returns the current synchronized state.
154    pub fn state(&self) -> &PresenceState {
155        &self.state
156    }
157
158    /// Clears state, join generation, and queued diffs.
159    pub fn reset(&mut self) {
160        self.state = PresenceState::default();
161        self.join_ref = None;
162        self.pending_diffs.clear();
163    }
164
165    /// Applies a Phoenix frame and returns its Presence effect.
166    pub fn apply(&mut self, frame: &Frame) -> Result<PresenceUpdate, PresenceError> {
167        if frame.event != self.state_event && frame.event != self.diff_event {
168            return Ok(PresenceUpdate::Ignored);
169        }
170        let payload = frame
171            .payload
172            .as_json()
173            .ok_or(PresenceError::BinaryPayload)?;
174        if frame.event == self.state_event {
175            let new_state = PresenceState::from_value(payload)?;
176            let mut changes = sync_state(&mut self.state, new_state)?;
177            for diff in self.pending_diffs.drain(..) {
178                changes.extend(sync_diff(&mut self.state, diff)?);
179            }
180            self.join_ref = frame.join_ref.clone();
181            Ok(PresenceUpdate::Synced(changes))
182        } else if frame.event == self.diff_event {
183            let diff = PresenceDiff::from_value(payload)?;
184            if self.join_ref.is_none() || self.join_ref != frame.join_ref {
185                self.pending_diffs.push(diff);
186                Ok(PresenceUpdate::Pending)
187            } else {
188                Ok(PresenceUpdate::Synced(sync_diff(&mut self.state, diff)?))
189            }
190        } else {
191            unreachable!("presence event names were checked before decoding")
192        }
193    }
194}
195
196impl Default for PresenceTracker {
197    fn default() -> Self {
198        Self::new()
199    }
200}
201
202/// Replaces `state` with a full server state and returns its joins and leaves.
203pub fn sync_state(
204    state: &mut PresenceState,
205    new_state: PresenceState,
206) -> Result<PresenceDiff, PresenceError> {
207    let mut joins = PresenceState::default();
208    let mut leaves = PresenceState::default();
209    for (key, current) in &state.0 {
210        if !new_state.0.contains_key(key) {
211            leaves.0.insert(key.clone(), current.clone());
212        }
213    }
214    for (key, new_presence) in &new_state.0 {
215        if let Some(current) = state.0.get(key) {
216            let current_refs = refs(key, &current.metas)?;
217            let new_refs = refs(key, &new_presence.metas)?;
218            let joined_metas = new_presence
219                .metas
220                .iter()
221                .filter(|meta| !current_refs.contains(meta_ref(key, meta).unwrap_or_default()))
222                .cloned()
223                .collect::<Vec<_>>();
224            let left_metas = current
225                .metas
226                .iter()
227                .filter(|meta| !new_refs.contains(meta_ref(key, meta).unwrap_or_default()))
228                .cloned()
229                .collect::<Vec<_>>();
230            if !joined_metas.is_empty() {
231                let mut joined = new_presence.clone();
232                joined.metas = joined_metas;
233                joins.0.insert(key.clone(), joined);
234            }
235            if !left_metas.is_empty() {
236                let mut left = current.clone();
237                left.metas = left_metas;
238                leaves.0.insert(key.clone(), left);
239            }
240        } else {
241            joins.0.insert(key.clone(), new_presence.clone());
242        }
243    }
244    let changes = PresenceDiff { joins, leaves };
245    sync_diff(state, changes.clone())?;
246    Ok(changes)
247}
248
249/// Applies incremental Presence joins and leaves to `state`.
250pub fn sync_diff(
251    state: &mut PresenceState,
252    diff: PresenceDiff,
253) -> Result<PresenceDiff, PresenceError> {
254    for (key, joined) in &diff.joins.0 {
255        let joined_refs = refs(key, &joined.metas)?;
256        let mut merged = joined.clone();
257        if let Some(current) = state.0.get(key) {
258            let mut existing = current
259                .metas
260                .iter()
261                .filter(|meta| !joined_refs.contains(meta_ref(key, meta).unwrap_or_default()))
262                .cloned()
263                .collect::<Vec<_>>();
264            existing.extend(merged.metas);
265            merged.metas = existing;
266        }
267        state.0.insert(key.clone(), merged);
268    }
269    for (key, left) in &diff.leaves.0 {
270        let left_refs = refs(key, &left.metas)?;
271        if let Some(current) = state.0.get_mut(key) {
272            current
273                .metas
274                .retain(|meta| !left_refs.contains(meta_ref(key, meta).unwrap_or_default()));
275            if current.metas.is_empty() {
276                state.0.remove(key);
277            }
278        }
279    }
280    Ok(diff)
281}
282
283fn refs(key: &str, metas: &[Map<String, Value>]) -> Result<BTreeSet<String>, PresenceError> {
284    metas
285        .iter()
286        .map(|meta| meta_ref(key, meta).map(str::to_owned))
287        .collect()
288}
289
290fn meta_ref<'a>(key: &str, meta: &'a Map<String, Value>) -> Result<&'a str, PresenceError> {
291    meta.get("phx_ref")
292        .and_then(Value::as_str)
293        .ok_or_else(|| PresenceError::MissingReference(key.to_owned()))
294}
295
296/// Invalid Phoenix Presence state, diff, or metadata.
297#[derive(Clone, Debug, Error, Eq, PartialEq)]
298pub enum PresenceError {
299    /// A full state was not a JSON object.
300    #[error("presence state must be an object")]
301    InvalidState,
302    /// A diff did not contain valid `joins` and `leaves` objects.
303    #[error("presence diff must contain joins and leaves objects")]
304    InvalidDiff,
305    /// The named presence entry was not an object.
306    #[error("presence entry for {0} must be an object")]
307    InvalidPresence(String),
308    /// The named presence entry had no `metas` array.
309    #[error("presence entry for {0} must contain a metas array")]
310    InvalidMetas(String),
311    /// A meta belonging to the named entry was not an object.
312    #[error("presence meta for {0} must be an object")]
313    InvalidMeta(String),
314    /// A meta belonging to the named entry had no string `phx_ref`.
315    #[error("presence meta for {0} must contain a string phx_ref")]
316    MissingReference(String),
317    /// Presence events cannot be decoded from binary payloads.
318    #[error("presence events cannot use binary payloads")]
319    BinaryPayload,
320}
321
322#[cfg(test)]
323mod tests {
324    use serde_json::json;
325
326    use super::*;
327
328    fn frame(join_ref: &str, event: &str, payload: Value) -> Frame {
329        Frame::new(Some(join_ref.into()), None, "room:lobby", event, payload)
330    }
331
332    #[test]
333    fn queues_diffs_until_state_and_tracks_meta_references() {
334        let mut tracker = PresenceTracker::new();
335        let pending = frame(
336            "1",
337            "presence_diff",
338            json!({
339                "joins": {"u1": {"metas": [{"phx_ref": "a", "online_at": 1}]}},
340                "leaves": {}
341            }),
342        );
343        assert_eq!(tracker.apply(&pending).unwrap(), PresenceUpdate::Pending);
344
345        let state = frame("1", "presence_state", json!({}));
346        let PresenceUpdate::Synced(changes) = tracker.apply(&state).unwrap() else {
347            panic!("expected a presence sync");
348        };
349        assert!(changes.joins.get("u1").is_some());
350        assert_eq!(tracker.state().get("u1").unwrap().metas.len(), 1);
351
352        let leave = frame(
353            "1",
354            "presence_diff",
355            json!({
356                "joins": {},
357                "leaves": {"u1": {"metas": [{"phx_ref": "a"}]}}
358            }),
359        );
360        tracker.apply(&leave).unwrap();
361        assert!(tracker.state().is_empty());
362    }
363
364    #[test]
365    fn replaces_state_and_reports_only_changed_metas() {
366        let mut current = PresenceState::from_value(&json!({
367            "u1": {"metas": [{"phx_ref": "a"}, {"phx_ref": "b"}]}
368        }))
369        .unwrap();
370        let next = PresenceState::from_value(&json!({
371            "u1": {"metas": [{"phx_ref": "b"}, {"phx_ref": "c"}]}
372        }))
373        .unwrap();
374        let diff = sync_state(&mut current, next).unwrap();
375        assert_eq!(diff.joins.get("u1").unwrap().metas.len(), 1);
376        assert_eq!(diff.leaves.get("u1").unwrap().metas.len(), 1);
377        assert_eq!(current.get("u1").unwrap().metas.len(), 2);
378    }
379
380    #[test]
381    fn reset_discards_state_and_pending_diffs() {
382        let mut tracker = PresenceTracker::new();
383        tracker
384            .apply(&frame(
385                "1",
386                "presence_state",
387                json!({"u1": {"metas": [{"phx_ref": "a"}]}}),
388            ))
389            .unwrap();
390        assert!(!tracker.state().is_empty());
391
392        tracker.reset();
393        assert!(tracker.state().is_empty());
394        assert_eq!(
395            tracker
396                .apply(&frame(
397                    "2",
398                    "presence_diff",
399                    json!({"joins": {}, "leaves": {}}),
400                ))
401                .unwrap(),
402            PresenceUpdate::Pending
403        );
404    }
405}
phoenix_channel_runtime/presence.rs

phoenix_channel_runtime/
presence.rs
