matrix_sdk_base/
response_processors.rs

1// Copyright 2024 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{
16    collections::{BTreeMap, HashMap, HashSet},
17    mem,
18};
19
20use ruma::{
21    events::{
22        direct::OwnedDirectUserIdentifier, AnyGlobalAccountDataEvent, GlobalAccountDataEventType,
23    },
24    serde::Raw,
25    RoomId,
26};
27use tracing::{debug, instrument, trace, warn};
28
29use crate::{store::Store, RoomInfo, StateChanges};
30
31/// Applies a function to an existing `RoomInfo` if present in changes, or one
32/// loaded from the database.
33fn map_info<F: FnOnce(&mut RoomInfo)>(
34    room_id: &RoomId,
35    changes: &mut StateChanges,
36    store: &Store,
37    f: F,
38) {
39    if let Some(info) = changes.room_infos.get_mut(room_id) {
40        f(info);
41    } else if let Some(room) = store.room(room_id) {
42        let mut info = room.clone_info();
43        f(&mut info);
44        changes.add_room(info);
45    } else {
46        debug!(room = %room_id, "couldn't find room in state changes or store");
47    }
48}
49
50#[must_use]
51pub(crate) struct AccountDataProcessor {
52    parsed_events: Vec<AnyGlobalAccountDataEvent>,
53    raw_by_type: BTreeMap<GlobalAccountDataEventType, Raw<AnyGlobalAccountDataEvent>>,
54}
55
56impl AccountDataProcessor {
57    /// Creates a new processor for global account data.
58    pub fn process(events: &[Raw<AnyGlobalAccountDataEvent>]) -> Self {
59        let mut raw_by_type = BTreeMap::new();
60        let mut parsed_events = Vec::new();
61
62        for raw_event in events {
63            let event = match raw_event.deserialize() {
64                Ok(e) => e,
65                Err(e) => {
66                    let event_type: Option<String> = raw_event.get_field("type").ok().flatten();
67                    warn!(event_type, "Failed to deserialize a global account data event: {e}");
68                    continue;
69                }
70            };
71
72            raw_by_type.insert(event.event_type(), raw_event.clone());
73            parsed_events.push(event);
74        }
75
76        Self { raw_by_type, parsed_events }
77    }
78
79    /// Returns the push rules found by this processor.
80    pub fn push_rules(&self) -> Option<&Raw<AnyGlobalAccountDataEvent>> {
81        self.raw_by_type.get(&GlobalAccountDataEventType::PushRules)
82    }
83
84    /// Processes the direct rooms in a sync response:
85    ///
86    /// Given a [`StateChanges`] instance, processes any direct room info
87    /// from the global account data and adds it to the room infos to
88    /// save.
89    #[instrument(skip_all)]
90    pub(crate) fn process_direct_rooms(
91        &self,
92        events: &[AnyGlobalAccountDataEvent],
93        store: &Store,
94        changes: &mut StateChanges,
95    ) {
96        for event in events {
97            let AnyGlobalAccountDataEvent::Direct(direct_event) = event else { continue };
98
99            let mut new_dms = HashMap::<&RoomId, HashSet<OwnedDirectUserIdentifier>>::new();
100            for (user_identifier, rooms) in direct_event.content.iter() {
101                for room_id in rooms {
102                    new_dms.entry(room_id).or_default().insert(user_identifier.clone());
103                }
104            }
105
106            let rooms = store.rooms();
107            let mut old_dms = rooms
108                .iter()
109                .filter_map(|r| {
110                    let direct_targets = r.direct_targets();
111                    (!direct_targets.is_empty()).then(|| (r.room_id(), direct_targets))
112                })
113                .collect::<HashMap<_, _>>();
114
115            // Update the direct targets of rooms if they changed.
116            for (room_id, new_direct_targets) in new_dms {
117                if let Some(old_direct_targets) = old_dms.remove(&room_id) {
118                    if old_direct_targets == new_direct_targets {
119                        continue;
120                    }
121                }
122                trace!(?room_id, targets = ?new_direct_targets, "Marking room as direct room");
123                map_info(room_id, changes, store, |info| {
124                    info.base_info.dm_targets = new_direct_targets;
125                });
126            }
127
128            // Remove the targets of old direct chats.
129            for room_id in old_dms.keys() {
130                trace!(?room_id, "Unmarking room as direct room");
131                map_info(room_id, changes, store, |info| {
132                    info.base_info.dm_targets.clear();
133                });
134            }
135        }
136    }
137
138    /// Applies the processed data to the state changes.
139    pub async fn apply(mut self, changes: &mut StateChanges, store: &Store) {
140        // Fill in the content of `changes.account_data`.
141        mem::swap(&mut changes.account_data, &mut self.raw_by_type);
142
143        // Process direct rooms.
144        let has_new_direct_room_data = self
145            .parsed_events
146            .iter()
147            .any(|event| event.event_type() == GlobalAccountDataEventType::Direct);
148
149        if has_new_direct_room_data {
150            self.process_direct_rooms(&self.parsed_events, store, changes);
151        } else if let Ok(Some(direct_account_data)) =
152            store.get_account_data_event(GlobalAccountDataEventType::Direct).await
153        {
154            debug!("Found direct room data in the Store, applying it");
155            if let Ok(direct_account_data) = direct_account_data.deserialize() {
156                self.process_direct_rooms(&[direct_account_data], store, changes);
157            } else {
158                warn!("Failed to deserialize direct room account data");
159            }
160        }
161    }
162}
matrix_sdk_base/response_processors.rs

matrix_sdk_base/
response_processors.rs
