1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
use sqlx::sqlite::SqlitePool;
use super::{Channel, Snapshot};
use crate::{
channel::repo::Provider as _, event::repo::Provider as _, login::repo::Provider as _,
message::repo::Provider as _,
};
pub struct Boot<'a> {
db: &'a SqlitePool,
}
impl<'a> Boot<'a> {
pub const fn new(db: &'a SqlitePool) -> Self {
Self { db }
}
pub async fn snapshot(&self) -> Result<Snapshot, sqlx::Error> {
let mut tx = self.db.begin().await?;
let resume_point = tx.sequence().current().await?;
let logins = tx.logins().all(resume_point.into()).await?;
let logins = logins
.into_iter()
.filter_map(|login| login.as_of(resume_point))
.collect();
let channels = tx.channels().all(resume_point.into()).await?;
let channels = {
let mut snapshots = Vec::with_capacity(channels.len());
let channels = channels.into_iter().filter_map(|channel| {
channel
.as_of(resume_point)
.map(|snapshot| (channel, snapshot))
});
for (channel, snapshot) in channels {
let messages = tx
.messages()
.in_channel(&channel, resume_point.into())
.await?;
let messages = messages
.into_iter()
.filter_map(|message| message.as_of(resume_point));
snapshots.push(Channel::new(snapshot, messages));
}
snapshots
};
tx.commit().await?;
Ok(Snapshot {
resume_point,
logins,
channels,
})
}
}
|
