m4a_agent/backend/
sliding.rs

1//! Client side of simplified sliding sync (MSC4186). The engine speaks `/sync` v3; this turns its
2//! sync calls into sliding-sync calls and the answers back into v3 shape, so the engine does not
3//! change. A server that does not offer it (or refuses it) gets plain `/sync` v3.
4//!
5//! The v3 `next_batch` the engine keeps is the sliding `pos`. Rooms left or banned are not
6//! reported by sliding sync, so a leave seen elsewhere is the only way the engine learns of one.
7
8use std::sync::atomic::{AtomicU8, Ordering};
9use std::sync::Mutex;
10
11use mail4agent_messenger::{HttpMethod, HttpResponseDescriptor, OutgoingRequest};
12use serde_json::{json, Value};
13
14use super::live::Live;
15use super::matrix::SyncMode;
16use crate::error::Result;
17
18const PATH_SIMPLIFIED: &str = "/_matrix/client/unstable/org.matrix.simplified_msc3575/sync";
19const PATH_MSC4186: &str = "/_matrix/client/unstable/org.matrix.msc4186/sync";
20
21const UNDECIDED: u8 = 0;
22const ON: u8 = 1;
23const OFF: u8 = 2;
24
25pub struct Sliding {
26    state: AtomicU8,
27    forced_v3: AtomicU8,
28    path: Mutex<&'static str>,
29    pos: Mutex<Option<String>>,
30    to_device_since: Mutex<Option<String>>,
31}
32
33impl Default for Sliding {
34    fn default() -> Self {
35        Self::new()
36    }
37}
38
39impl Sliding {
40    pub fn new() -> Self {
41        Self { state: AtomicU8::new(UNDECIDED), forced_v3: AtomicU8::new(0), path: Mutex::new(PATH_SIMPLIFIED), pos: Mutex::new(None), to_device_since: Mutex::new(None) }
42    }
43
44    pub fn set_mode(&self, mode: SyncMode) {
45        self.forced_v3.store((mode == SyncMode::V3) as u8, Ordering::Release);
46    }
47
48    pub fn active(&self) -> bool {
49        self.state.load(Ordering::Acquire) == ON
50    }
51
52    /// Asks the server what it offers (`GET /versions`, `unstable_features`).
53    pub fn decide(&self, live: &Live) {
54        if self.forced_v3.load(Ordering::Acquire) == 1 {
55            self.state.store(OFF, Ordering::Release);
56            return;
57        }
58        let exec = live.exec();
59        let offered = exec.get_json("/_matrix/client/versions").ok().filter(|(s, _)| *s == 200).map(|(_, v)| v);
60        let feature = |k: &str| offered.as_ref().and_then(|v| v.get("unstable_features")).and_then(|f| f.get(k)).and_then(Value::as_bool).unwrap_or(false);
61        if feature("org.matrix.simplified_msc3575") {
62            *self.path.lock().unwrap_or_else(|e| e.into_inner()) = PATH_SIMPLIFIED;
63            self.state.store(ON, Ordering::Release);
64        } else if feature("org.matrix.msc4186") {
65            *self.path.lock().unwrap_or_else(|e| e.into_inner()) = PATH_MSC4186;
66            self.state.store(ON, Ordering::Release);
67        } else {
68            self.state.store(OFF, Ordering::Release);
69        }
70    }
71
72    fn body(&self) -> Value {
73        let since = self.to_device_since.lock().unwrap_or_else(|e| e.into_inner()).clone();
74        let mut to_device = json!({ "enabled": true });
75        if let Some(s) = since {
76            to_device["since"] = json!(s);
77        }
78        json!({
79            "conn_id": "m4a",
80            "lists": { "all": { "ranges": [[0, 99]], "timeline_limit": 20, "required_state": [["*", "*"]] } },
81            "extensions": { "to_device": to_device, "e2ee": { "enabled": true }, "account_data": { "enabled": true } }
82        })
83    }
84
85    /// Serves a v3 sync request by sliding sync. `Ok(None)`: not available, use v3.
86    pub fn sync(&self, live: &Live, request: &OutgoingRequest) -> Result<Option<HttpResponseDescriptor>> {
87        if self.state.load(Ordering::Acquire) != ON {
88            return Ok(None);
89        }
90        let timeout = request.query.iter().find(|(n, _)| n == "timeout").map(|(_, v)| v.clone()).unwrap_or_else(|| "0".into());
91        // A `since` the engine holds from another run means nothing here: the connection is new.
92        let since = request.query.iter().find(|(n, _)| n == "since").map(|(_, v)| v.clone());
93        let mut pos = self.pos.lock().unwrap_or_else(|e| e.into_inner()).clone().filter(|p| since.as_deref() == Some(p.as_str()));
94        for attempt in 0..2 {
95            let mut query = vec![("timeout".to_string(), timeout.clone())];
96            if let Some(p) = &pos {
97                query.push(("pos".to_string(), p.clone()));
98            }
99            let sliding_request = OutgoingRequest {
100                id: request.id.clone(),
101                method: HttpMethod::Post,
102                path: self.path.lock().unwrap_or_else(|e| e.into_inner()).to_string(),
103                query,
104                body: Some(self.body()),
105                kind: request.kind,
106            };
107            let r = live.execute(&sliding_request)?;
108            let errcode = serde_json::from_slice::<Value>(&r.body).ok().and_then(|v| v.get("errcode").and_then(Value::as_str).map(str::to_string));
109            match (r.status, errcode.as_deref()) {
110                (200, _) => {
111                    let v: Value = serde_json::from_slice(&r.body).unwrap_or(Value::Null);
112                    let v3 = to_v3(&v);
113                    *self.pos.lock().unwrap_or_else(|e| e.into_inner()) = v.get("pos").and_then(Value::as_str).map(str::to_string);
114                    if let Some(n) = v.pointer("/extensions/to_device/next_batch").and_then(Value::as_str) {
115                        *self.to_device_since.lock().unwrap_or_else(|e| e.into_inner()) = Some(n.to_string());
116                    }
117                    return Ok(Some(HttpResponseDescriptor { status: 200, body: serde_json::to_vec(&v3).unwrap_or_default() }));
118                }
119                (400, Some("M_UNKNOWN_POS")) if attempt == 0 => {
120                    // The server lost the connection: start again.
121                    *self.pos.lock().unwrap_or_else(|e| e.into_inner()) = None;
122                    *self.to_device_since.lock().unwrap_or_else(|e| e.into_inner()) = None;
123                    pos = None;
124                }
125                (404 | 405 | 501, _) | (_, Some("M_UNRECOGNIZED")) => {
126                    // Not served after all: v3 from now on.
127                    self.state.store(OFF, Ordering::Release);
128                    return Ok(None);
129                }
130                _ => return Ok(Some(r)),
131            }
132        }
133        Ok(None)
134    }
135}
136
137/// A sliding-sync answer in the shape of a v3 `/sync` answer.
138pub fn to_v3(v: &Value) -> Value {
139    let mut join = serde_json::Map::new();
140    let mut invite = serde_json::Map::new();
141    if let Some(rooms) = v.get("rooms").and_then(Value::as_object) {
142        for (id, room) in rooms {
143            if let Some(stripped) = room.get("invite_state").and_then(Value::as_array) {
144                invite.insert(id.clone(), json!({ "invite_state": { "events": stripped } }));
145                continue;
146            }
147            let timeline = room.get("timeline").cloned().unwrap_or_else(|| json!([]));
148            let mut entry = json!({
149                "timeline": { "events": timeline, "limited": room.get("limited").and_then(Value::as_bool).unwrap_or(false) },
150                "state": { "events": room.get("required_state").cloned().unwrap_or_else(|| json!([])) },
151            });
152            if let Some(p) = room.get("prev_batch") {
153                entry["timeline"]["prev_batch"] = p.clone();
154            }
155            let n = room.get("notification_count").and_then(Value::as_u64);
156            let h = room.get("highlight_count").and_then(Value::as_u64);
157            if n.is_some() || h.is_some() {
158                entry["unread_notifications"] = json!({ "notification_count": n.unwrap_or(0), "highlight_count": h.unwrap_or(0) });
159            }
160            join.insert(id.clone(), entry);
161        }
162    }
163    let ext = v.get("extensions").cloned().unwrap_or(Value::Null);
164    let mut out = json!({
165        "next_batch": v.get("pos").and_then(Value::as_str).unwrap_or(""),
166        "rooms": { "join": join, "invite": invite, "leave": {} },
167        "to_device": { "events": ext.pointer("/to_device/events").cloned().unwrap_or_else(|| json!([])) },
168        "account_data": { "events": ext.pointer("/account_data/global").cloned().unwrap_or_else(|| json!([])) },
169    });
170    if let Some(d) = ext.pointer("/e2ee/device_lists") {
171        out["device_lists"] = d.clone();
172    }
173    if let Some(c) = ext.pointer("/e2ee/device_one_time_keys_count") {
174        out["device_one_time_keys_count"] = c.clone();
175    }
176    if let Some(f) = ext.pointer("/e2ee/device_unused_fallback_key_types") {
177        out["device_unused_fallback_key_types"] = f.clone();
178    }
179    out
180}
181
182#[cfg(test)]
183mod tests {
184    use super::*;
185
186    #[test]
187    fn a_sliding_answer_becomes_a_v3_answer() {
188        let v = json!({
189            "pos": "p7",
190            "rooms": {
191                "!a:s": { "required_state": [{"type":"m.room.create"}], "timeline": [{"type":"m.room.message","event_id":"$1"}], "limited": true, "prev_batch": "pb", "notification_count": 2 },
192                "!i:s": { "invite_state": [{"type":"m.room.member","state_key":"@me:s"}] }
193            },
194            "extensions": {
195                "to_device": { "next_batch": "t1", "events": [{"type":"m.room.encrypted"}] },
196                "e2ee": { "device_lists": {"changed":["@x:s"]}, "device_one_time_keys_count": {"signed_curve25519": 5}, "device_unused_fallback_key_types": [] },
197                "account_data": { "global": [{"type":"m.direct"}] }
198            }
199        });
200        let o = to_v3(&v);
201        assert_eq!(o["next_batch"], "p7");
202        assert_eq!(o["rooms"]["join"]["!a:s"]["timeline"]["events"][0]["event_id"], "$1");
203        assert_eq!(o["rooms"]["join"]["!a:s"]["timeline"]["limited"], true);
204        assert_eq!(o["rooms"]["join"]["!a:s"]["state"]["events"][0]["type"], "m.room.create");
205        assert_eq!(o["rooms"]["join"]["!a:s"]["unread_notifications"]["notification_count"], 2);
206        assert!(o["rooms"]["join"].get("!i:s").is_none());
207        assert_eq!(o["rooms"]["invite"]["!i:s"]["invite_state"]["events"][0]["state_key"], "@me:s");
208        assert_eq!(o["to_device"]["events"][0]["type"], "m.room.encrypted");
209        assert_eq!(o["device_lists"]["changed"][0], "@x:s");
210        assert_eq!(o["device_one_time_keys_count"]["signed_curve25519"], 5);
211        assert_eq!(o["account_data"]["events"][0]["type"], "m.direct");
212        let e = to_v3(&json!({"pos":"p8"}));
213        assert_eq!(e["next_batch"], "p8");
214        assert!(e["rooms"]["join"].as_object().unwrap().is_empty());
215    }
216}
m4a_agent/backend/sliding.rs

m4a_agent/backend/
sliding.rs
