lb_rs/service/
activity.rs

1use crate::LocalLb;
2use crate::model::errors::LbResult;
3use crate::model::tree_like::TreeLike;
4use crate::service::events::Actor;
5use serde::{Deserialize, Serialize};
6use std::cmp::{self, Ordering};
7use std::collections::HashMap;
8use uuid::Uuid;
9
10impl LocalLb {
11    #[instrument(level = "debug", skip(self), err(Debug))]
12    pub async fn suggested_docs(&self, settings: RankingWeights) -> LbResult<Vec<Uuid>> {
13        let db = self.ro_tx().await;
14        let db = db.db();
15
16        let mut scores = db.doc_events.get().iter().get_activity_metrics();
17        self.normalize(&mut scores);
18
19        scores.sort_unstable_by_key(|b| cmp::Reverse(b.score(settings)));
20
21        scores.truncate(10);
22
23        let mut result = Vec::new();
24        let mut tree = (&db.base_metadata).to_staged(&db.local_metadata).to_lazy();
25        for score in scores {
26            if tree.maybe_find(&score.id).is_none() {
27                continue;
28            }
29
30            if tree.calculate_deleted(&score.id)? {
31                continue;
32            }
33            if tree.in_pending_share(&score.id)? {
34                continue;
35            }
36            result.push(score.id);
37        }
38
39        Ok(result)
40    }
41
42    #[instrument(level = "debug", skip(self), err(Debug))]
43    pub async fn clear_suggested(&self) -> LbResult<()> {
44        let mut tx = self.begin_tx().await;
45        let db = tx.db();
46        db.doc_events.clear()?;
47        self.events.meta_changed(Actor::User(None));
48        Ok(())
49    }
50
51    #[instrument(level = "debug", skip(self), err(Debug))]
52    pub async fn clear_suggested_id(&self, id: Uuid) -> LbResult<()> {
53        let mut tx = self.begin_tx().await;
54        let db = tx.db();
55
56        let mut entries = db.doc_events.get().to_vec();
57        db.doc_events.clear()?;
58        entries.retain(|e| e.id() != id);
59        for entry in entries {
60            db.doc_events.push(entry)?;
61        }
62
63        Ok(())
64    }
65
66    pub(crate) async fn add_doc_event(&self, event: DocEvent) -> LbResult<()> {
67        let mut tx = self.begin_tx().await;
68        let db = tx.db();
69
70        let max_stored_events = 1000;
71        let events = &db.doc_events;
72
73        if events.get().len() > max_stored_events {
74            db.doc_events.remove(0)?;
75        }
76        db.doc_events.push(event)?;
77        Ok(())
78    }
79
80    pub(crate) fn normalize(&self, docs: &mut [DocActivityMetrics]) {
81        let read_count_range = StatisticValueRange {
82            max: docs.iter().map(|f| f.read_count).max().unwrap_or_default(),
83            min: docs.iter().map(|f| f.read_count).min().unwrap_or_default(),
84        };
85
86        let write_count_range = StatisticValueRange {
87            max: docs.iter().map(|f| f.write_count).max().unwrap_or_default(),
88            min: docs.iter().map(|f| f.write_count).min().unwrap_or_default(),
89        };
90
91        let last_read_range = StatisticValueRange {
92            max: docs
93                .iter()
94                .map(|f| f.last_read_timestamp)
95                .max()
96                .unwrap_or_default(),
97            min: docs
98                .iter()
99                .map(|f| f.last_read_timestamp)
100                .min()
101                .unwrap_or_default(),
102        };
103        let last_write_range = StatisticValueRange {
104            max: docs
105                .iter()
106                .map(|f| f.last_write_timestamp)
107                .max()
108                .unwrap_or_default(),
109            min: docs
110                .iter()
111                .map(|f| f.last_write_timestamp)
112                .min()
113                .unwrap_or_default(),
114        };
115
116        docs.iter_mut().for_each(|f| {
117            f.read_count.normalize(read_count_range);
118            f.write_count.normalize(write_count_range);
119            f.last_read_timestamp.normalize(last_read_range);
120            f.last_write_timestamp.normalize(last_write_range);
121        });
122    }
123
124    /// hint to background processing pipelines whether or not a user is around
125    pub fn app_foregrounded(&self) {
126        let bg_lb = self.clone();
127        tokio::spawn(async move {
128            *bg_lb.user_last_seen.write().await = web_time::Instant::now();
129            bg_lb.user_wake.notify_one();
130        });
131    }
132}
133
134#[derive(Debug, Clone, Serialize, Deserialize, Ord, PartialEq, PartialOrd, Eq, Hash, Copy)]
135pub enum DocEvent {
136    Read(Uuid, i64),
137    Write(Uuid, i64),
138}
139impl DocEvent {
140    pub fn timestamp(&self) -> i64 {
141        match *self {
142            DocEvent::Read(_, timestamp) => timestamp,
143            DocEvent::Write(_, timestamp) => timestamp,
144        }
145    }
146    pub fn id(&self) -> Uuid {
147        match *self {
148            DocEvent::Read(id, _) => id,
149            DocEvent::Write(id, _) => id,
150        }
151    }
152}
153
154#[derive(Debug, Copy, Clone, Serialize, Deserialize)]
155pub struct RankingWeights {
156    /// the freshness of a doc as determined by the last activity
157    pub temporality: i64,
158    /// the amount of write and read on a doc
159    pub io: i64,
160}
161
162impl Default for RankingWeights {
163    fn default() -> Self {
164        Self { temporality: 60, io: 40 }
165    }
166}
167#[derive(Default, Copy, Clone, PartialEq)]
168pub struct StatisticValue {
169    pub raw: i64,
170    pub normalized: Option<f64>,
171}
172
173impl Ord for StatisticValue {
174    fn cmp(&self, other: &Self) -> Ordering {
175        (self.raw).cmp(&other.raw)
176    }
177}
178
179impl PartialOrd for StatisticValue {
180    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
181        Some(self.cmp(other))
182    }
183}
184
185impl Eq for StatisticValue {}
186
187#[derive(Clone, Copy)]
188pub struct StatisticValueRange {
189    pub max: StatisticValue,
190    pub min: StatisticValue,
191}
192impl StatisticValue {
193    pub fn normalize(&mut self, range: StatisticValueRange) {
194        let mut range_distance = range.max.raw - range.min.raw;
195        if range_distance == 0 {
196            range_distance = 1
197        };
198        let normalized = (self.raw - range.min.raw) as f64 / range_distance as f64;
199        self.normalized = Some(normalized);
200    }
201}
202/// DocActivityMetrics stores key document activity features, which are used to recommend relevant documents to the user.
203/// Here's a walkthrough of the recommendation procedure: collect 1k most recent document events (write/read), use that activity to construct a DocActivtyMetrics struct for each document. Min-max normalizes the activity features, then rank the documents.
204#[derive(Default, Copy, Clone, PartialEq)]
205pub struct DocActivityMetrics {
206    pub id: Uuid,
207    /// the latest epoch timestamp that the user read a document
208    pub last_read_timestamp: StatisticValue,
209    /// the latest epoch timestamp that the user wrote a document
210    pub last_write_timestamp: StatisticValue,
211    /// the total number of times that a user reads a document
212    pub read_count: StatisticValue,
213    /// the total number of times that a user wrote a document
214    pub write_count: StatisticValue,
215}
216
217impl DocActivityMetrics {
218    pub fn score(&self, weights: RankingWeights) -> i64 {
219        let timestamp_weight = weights.temporality;
220        let io_count_weight = weights.io;
221
222        let temporality_score = (self.last_read_timestamp.normalized.unwrap_or_default()
223            + self.last_write_timestamp.normalized.unwrap_or_default())
224            * timestamp_weight as f64;
225
226        let io_score = (self.read_count.normalized.unwrap_or_default()
227            + self.write_count.normalized.unwrap_or_default())
228            * io_count_weight as f64;
229
230        (io_score + temporality_score).ceil() as i64
231    }
232}
233pub trait Stats {
234    fn get_activity_metrics(self) -> Vec<DocActivityMetrics>;
235}
236impl<'a, T> Stats for T
237where
238    T: Iterator<Item = &'a DocEvent>,
239{
240    fn get_activity_metrics(self) -> Vec<DocActivityMetrics> {
241        let mut result = Vec::new();
242
243        let mut set = HashMap::new();
244        for event in self {
245            match set.get_mut(&event.id()) {
246                None => {
247                    set.insert(event.id(), vec![event]);
248                }
249                Some(events) => {
250                    events.push(event);
251                }
252            }
253        }
254
255        for (id, events) in set {
256            let read_events = events.iter().filter(|e| matches!(e, DocEvent::Read(_, _)));
257
258            let last_read = read_events
259                .clone()
260                .max_by(|x, y| x.timestamp().cmp(&y.timestamp()));
261
262            let last_read = match last_read {
263                None => 0,
264                Some(x) => x.timestamp(),
265            };
266
267            let write_events = events.iter().filter(|e| matches!(e, DocEvent::Write(_, _)));
268
269            let last_write = write_events
270                .clone()
271                .max_by(|x, y| x.timestamp().cmp(&y.timestamp()));
272            let last_write = match last_write {
273                None => 0,
274                Some(x) => x.timestamp(),
275            };
276
277            let metrics = DocActivityMetrics {
278                id,
279                last_read_timestamp: StatisticValue { raw: last_read, normalized: None },
280                last_write_timestamp: StatisticValue { raw: last_write, normalized: None },
281                read_count: StatisticValue { raw: read_events.count() as i64, normalized: None },
282                write_count: StatisticValue { raw: write_events.count() as i64, normalized: None },
283            };
284            result.push(metrics);
285        }
286
287        result
288    }
289}
lb_rs/service/activity.rs

lb_rs/service/
activity.rs
