| 1 | use std::collections::{BTreeMap, BTreeSet}; |
| 2 | |
| 3 | use ruma::{OwnedEventId, OwnedUserId}; |
| 4 | use serde_json::Value; |
| 5 | |
| 6 | use crate::{ |
| 7 | Error, ErrorKind, EventContext, HistoryRepairPolicy, MentionCaptureState, MentionCoverage, |
| 8 | MentionEntry, MentionKey, MentionsBoundary, MentionsPage, SavedMentionContext, TimelineItem, |
| 9 | normalise::MediaRegistration, |
| 10 | }; |
| 11 | |
| 12 | #[derive(Clone)] |
| 13 | #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] |
| 14 | pub(super) struct SavedEvent { |
| 15 | pub item: TimelineItem, |
| 16 | pub media: Vec<MediaRegistration>, |
| 17 | pub raw: Option<Value>, |
| 18 | } |
| 19 | |
| 20 | #[derive(Clone)] |
| 21 | #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] |
| 22 | pub(super) struct Candidate { |
| 23 | pub room_id: ruma::OwnedRoomId, |
| 24 | pub raw: Value, |
| 25 | pub attempts: u32, |
| 26 | pub retry_at_ms: u64, |
| 27 | } |
| 28 | |
| 29 | #[derive(Clone)] |
| 30 | #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] |
| 31 | pub(super) struct Window { |
| 32 | pub before: Vec<String>, |
| 33 | pub target: Option<String>, |
| 34 | pub after: Vec<String>, |
| 35 | pub start: Option<String>, |
| 36 | pub end: Option<String>, |
| 37 | } |
| 38 | |
| 39 | #[derive(Clone, Debug, Default)] |
| 40 | #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] |
| 41 | pub(crate) struct RepairCoverage { |
| 42 | pub(crate) last_confirmed_ms: Option<u64>, |
| 43 | pub(crate) boundary_ms: Option<u64>, |
| 44 | pub(crate) reliable: bool, |
| 45 | pub(crate) unresolved: bool, |
| 46 | pub(crate) latest_anchor: Option<OwnedEventId>, |
| 47 | } |
| 48 | |
| 49 | impl RepairCoverage { |
| 50 | pub(crate) fn duration_ms(&self) -> Option<u64> { |
| 51 | if !self.reliable { |
| 52 | return None; |
| 53 | } |
| 54 | self.boundary_ms?.checked_sub(self.last_confirmed_ms?) |
| 55 | } |
| 56 | } |
| 57 | |
| 58 | #[derive(Clone, Default)] |
| 59 | #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] |
| 60 | pub(super) struct Data { |
| 61 | pub revision: u64, |
| 62 | pub entries: BTreeMap<String, MentionEntry>, |
| 63 | pub events: BTreeMap<String, SavedEvent>, |
| 64 | pub windows: BTreeMap<String, Window>, |
| 65 | pub capture_attempts: BTreeMap<String, u32>, |
| 66 | pub candidates: BTreeMap<String, Candidate>, |
| 67 | pub redacted: BTreeMap<String, u64>, |
| 68 | pub edits: BTreeMap<String, Value>, |
| 69 | pub coverage: MentionCoverage, |
| 70 | pub checkpoints: BTreeMap<String, String>, |
| 71 | pub observed_rooms: BTreeSet<ruma::OwnedRoomId>, |
| 72 | #[cfg_attr(feature = "serde", serde(default))] |
| 73 | pub repair_policy: HistoryRepairPolicy, |
| 74 | #[cfg_attr(feature = "serde", serde(default))] |
| 75 | pub repair_coverage: BTreeMap<ruma::OwnedRoomId, RepairCoverage>, |
| 76 | pub history_cursor: Option<String>, |
| 77 | pub history_exhausted: bool, |
| 78 | pub latest_cursor: Option<String>, |
| 79 | #[cfg_attr(feature = "serde", serde(default))] |
| 80 | pub latest_anchor: Option<String>, |
| 81 | #[cfg_attr(feature = "serde", serde(default))] |
| 82 | pub latest_head: Option<String>, |
| 83 | } |
| 84 | |
| 85 | pub(super) fn id(key: &MentionKey) -> String { |
| 86 | format!("{}\n{}", key.room_id, key.event_id) |
| 87 | } |
| 88 | |
| 89 | pub(super) fn now() -> u64 { |
| 90 | matrix_sdk::ruma::MilliSecondsSinceUnixEpoch::now() |
| 91 | .get() |
| 92 | .into() |
| 93 | } |
| 94 | |
| 95 | pub(super) fn invalid(message: &'static str) -> Error { |
| 96 | Error::new(ErrorKind::Invalid, message) |
| 97 | } |
| 98 | |
| 99 | impl Data { |
| 100 | pub fn touch_contexts(&mut self, identity: &str) { |
| 101 | for (key, window) in &self.windows { |
| 102 | if window.target.as_deref() == Some(identity) |
| 103 | || window |
| 104 | .before |
| 105 | .iter() |
| 106 | .chain(&window.after) |
| 107 | .any(|key| key == identity) |
| 108 | { |
| 109 | if let Some(entry) = self.entries.get_mut(key) { |
| 110 | entry.revision = self.revision + 1; |
| 111 | } |
| 112 | } |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | pub fn page( |
| 117 | &self, |
| 118 | account: &OwnedUserId, |
| 119 | cursor: Option<&str>, |
| 120 | limit: u16, |
| 121 | ) -> Result<MentionsPage, Error> { |
| 122 | let (boundary, last) = if let Some(cursor) = cursor { |
| 123 | let value: Value = |
| 124 | serde_json::from_str(cursor).map_err(|_| invalid("invalid mentions cursor"))?; |
| 125 | if value["account"].as_str() != Some(account.as_str()) { |
| 126 | return Err(invalid("mentions cursor belongs to a different account")); |
| 127 | } |
| 128 | let boundary = value["revision"] |
| 129 | .as_u64() |
| 130 | .ok_or_else(|| invalid("invalid mentions cursor"))?; |
| 131 | let timestamp = value["timestamp"] |
| 132 | .as_u64() |
| 133 | .ok_or_else(|| invalid("invalid mentions cursor"))?; |
| 134 | let key = value["key"] |
| 135 | .as_str() |
| 136 | .ok_or_else(|| invalid("invalid mentions cursor"))? |
| 137 | .to_owned(); |
| 138 | if boundary > self.revision { |
| 139 | return Err(invalid("mentions cursor is newer than this store")); |
| 140 | } |
| 141 | (boundary, Some((timestamp, key))) |
| 142 | } else { |
| 143 | (self.revision, None) |
| 144 | }; |
| 145 | let mut entries: Vec<_> = self |
| 146 | .entries |
| 147 | .iter() |
| 148 | .filter(|(key, entry)| { |
| 149 | entry.discovery_revision <= boundary |
| 150 | && last.as_ref().is_none_or(|last| { |
| 151 | (entry.timestamp_ms, key.as_str()) < (last.0, last.1.as_str()) |
| 152 | }) |
| 153 | }) |
| 154 | .map(|(_, entry)| entry.clone()) |
| 155 | .collect(); |
| 156 | entries.sort_by(|a, b| (b.timestamp_ms, id(&b.key)).cmp(&(a.timestamp_ms, id(&a.key)))); |
| 157 | let limit = usize::from(limit.clamp(1, 50)); |
| 158 | let end_reached = entries.len() <= limit; |
| 159 | entries.truncate(limit); |
| 160 | let next_cursor = if end_reached { |
| 161 | None |
| 162 | } else { |
| 163 | entries.last().map(|entry| serde_json::json!({ |
| 164 | "account": account.as_str(), "revision": boundary, "timestamp": entry.timestamp_ms, "key": id(&entry.key) |
| 165 | }).to_string()) |
| 166 | }; |
| 167 | let mut coverage = self.coverage.clone(); |
| 168 | coverage.unresolved_candidates = self.candidates.len(); |
| 169 | Ok(MentionsPage { |
| 170 | entries, |
| 171 | next_cursor, |
| 172 | end_reached, |
| 173 | coverage, |
| 174 | unread_count: self.entries.values().filter(|entry| !entry.read).count(), |
| 175 | revision: self.revision, |
| 176 | boundary: MentionsBoundary { |
| 177 | account: account.clone(), |
| 178 | revision: boundary, |
| 179 | }, |
| 180 | }) |
| 181 | } |
| 182 | |
| 183 | pub fn context( |
| 184 | &self, |
| 185 | key: &MentionKey, |
| 186 | ) -> Result<(SavedMentionContext, Vec<MediaRegistration>), Error> { |
| 187 | let entry = self |
| 188 | .entries |
| 189 | .get(&id(key)) |
| 190 | .ok_or_else(|| Error::new(ErrorKind::EventNotFound, "mention was not found"))?; |
| 191 | let mut media = Vec::new(); |
| 192 | let mut item = |key: &String| { |
| 193 | self.events.get(key).map(|event| { |
| 194 | media.extend(event.media.clone()); |
| 195 | event.item.clone() |
| 196 | }) |
| 197 | }; |
| 198 | let context = self.windows.get(&id(key)).map(|window| EventContext { |
| 199 | events_before: window.before.iter().filter_map(&mut item).collect(), |
| 200 | event: window.target.as_ref().and_then(&mut item), |
| 201 | events_after: window.after.iter().filter_map(&mut item).collect(), |
| 202 | start_token: window.start.clone(), |
| 203 | end_token: window.end.clone(), |
| 204 | }); |
| 205 | Ok(( |
| 206 | SavedMentionContext { |
| 207 | key: key.clone(), |
| 208 | state: entry.capture.clone(), |
| 209 | context, |
| 210 | revision: self.revision, |
| 211 | }, |
| 212 | media, |
| 213 | )) |
| 214 | } |
| 215 | |
| 216 | pub fn collect_events(&mut self) { |
| 217 | let mut used: BTreeSet<String> = self.entries.keys().cloned().collect(); |
| 218 | for window in self.windows.values() { |
| 219 | used.extend(window.before.iter().chain(&window.after).cloned()); |
| 220 | used.extend(window.target.iter().cloned()); |
| 221 | } |
| 222 | self.events.retain(|key, _| used.contains(key)); |
| 223 | self.capture_attempts |
| 224 | .retain(|key, _| self.entries.contains_key(key)); |
| 225 | } |
| 226 | |
| 227 | pub fn evict_one(&mut self) -> bool { |
| 228 | let oldest = |keys: Vec<String>, entries: &BTreeMap<String, MentionEntry>| { |
| 229 | keys.into_iter() |
| 230 | .min_by_key(|key| entries.get(key).map_or(0, |entry| entry.timestamp_ms)) |
| 231 | }; |
| 232 | if let Some(key) = oldest( |
| 233 | self.windows |
| 234 | .keys() |
| 235 | .filter(|key| self.entries.get(*key).is_some_and(|entry| entry.read)) |
| 236 | .cloned() |
| 237 | .collect(), |
| 238 | &self.entries, |
| 239 | ) |
| 240 | .or_else(|| oldest(self.windows.keys().cloned().collect(), &self.entries)) |
| 241 | { |
| 242 | self.windows.remove(&key); |
| 243 | if let Some(entry) = self.entries.get_mut(&key) { |
| 244 | entry.capture = MentionCaptureState::Unavailable; |
| 245 | entry.revision = self.revision; |
| 246 | } |
| 247 | } else if let Some(key) = self |
| 248 | .entries |
| 249 | .iter() |
| 250 | .min_by_key(|(_, entry)| (!entry.read, entry.timestamp_ms)) |
| 251 | .map(|(key, _)| key.clone()) |
| 252 | { |
| 253 | self.entries.remove(&key); |
| 254 | } else if let Some(key) = self.candidates.keys().next().cloned() { |
| 255 | self.candidates.remove(&key); |
| 256 | } else if let Some(key) = self.edits.keys().next().cloned() { |
| 257 | self.edits.remove(&key); |
| 258 | } else if let Some(key) = self.redacted.keys().next().cloned() { |
| 259 | self.redacted.remove(&key); |
| 260 | } else { |
| 261 | return false; |
| 262 | } |
| 263 | self.coverage.truncated = true; |
| 264 | self.collect_events(); |
| 265 | true |
| 266 | } |
| 267 | |
| 268 | pub fn bound(&mut self, entries: usize, events: usize, candidates: usize) { |
| 269 | while self.entries.len() > entries { |
| 270 | if let Some(key) = self |
| 271 | .entries |
| 272 | .iter() |
| 273 | .min_by_key(|(_, entry)| (!entry.read, entry.timestamp_ms)) |
| 274 | .map(|(key, _)| key.clone()) |
| 275 | { |
| 276 | self.entries.remove(&key); |
| 277 | self.windows.remove(&key); |
| 278 | self.coverage.truncated = true; |
| 279 | } |
| 280 | } |
| 281 | self.collect_events(); |
| 282 | while self.events.len() > events && self.evict_one() {} |
| 283 | while self.candidates.len() > candidates { |
| 284 | let key = self.candidates.keys().next().cloned().unwrap(); |
| 285 | self.candidates.remove(&key); |
| 286 | self.coverage.truncated = true; |
| 287 | } |
| 288 | for map in [&mut self.redacted] { |
| 289 | while map.len() > entries + candidates { |
| 290 | let key = map |
| 291 | .iter() |
| 292 | .min_by_key(|(_, timestamp)| *timestamp) |
| 293 | .map(|(key, _)| key.clone()) |
| 294 | .unwrap(); |
| 295 | map.remove(&key); |
| 296 | self.coverage.truncated = true; |
| 297 | } |
| 298 | } |
| 299 | while self.edits.len() > candidates { |
| 300 | let key = self.edits.keys().next().cloned().unwrap(); |
| 301 | self.edits.remove(&key); |
| 302 | self.coverage.truncated = true; |
| 303 | } |
| 304 | self.coverage.unresolved_candidates = self.candidates.len(); |
| 305 | } |
| 306 | } |