aven_core/sync/persistence/
status.rs

1use anyhow::Result;
2
3use crate::db::{Database, get_meta, set_meta};
4
5use super::SyncPersistenceStatus;
6
7impl Database {
8    pub async fn sync_persistence_status(&self) -> Result<SyncPersistenceStatus> {
9        let mut conn = self.acquire_reader().await?;
10        super::sync_persistence_status(&mut conn).await
11    }
12
13    pub async fn sync_facts(&self) -> Result<crate::api::SyncFacts> {
14        let mut conn = self.acquire_reader().await?;
15        let mut tx = sqlx::Connection::begin(&mut *conn).await?;
16        let status = super::sync_persistence_status(&mut tx).await?;
17        let missing = super::super::blob::missing_local_blob_counts(&mut tx).await?;
18        let facts = crate::api::SyncFacts {
19            compatibility_block: status.blocked_protocol.map(|server_protocol| {
20                super::super::protocol::SyncCompatibilityError {
21                    server_protocol,
22                    client_protocol: super::super::wire::SYNC_PROTOCOL_VERSION,
23                }
24            }),
25            pending_changes: status.pending_changes,
26            attachment_uploads: status.pending_attachment_uploads,
27            attachment_downloads: missing.count,
28            metadata_confirmed_at: get_meta(&mut tx, "sync_metadata_confirmed_at").await?,
29            metadata_caught_up: get_meta(&mut tx, "sync_metadata_caught_up")
30                .await?
31                .as_deref()
32                == Some("1"),
33        };
34        tx.commit().await?;
35        Ok(facts)
36    }
37
38    pub async fn begin_sync_attempt(&self, attempted_at: String) -> Result<()> {
39        let mut conn = self.acquire_writer().await?;
40        set_meta(&mut conn, "sync_last_attempt_at", &attempted_at).await
41    }
42
43    pub async fn record_sync_error(&self, error: String) -> Result<()> {
44        let mut conn = self.acquire_writer().await?;
45        set_meta(&mut conn, "sync_last_error", &error).await
46    }
47}
48
49pub(super) async fn sync_persistence_status(
50    conn: &mut sqlx::SqliteConnection,
51) -> Result<SyncPersistenceStatus> {
52    let pending_changes =
53        sqlx::query_scalar("SELECT count(*) FROM changes WHERE server_seq IS NULL")
54            .fetch_one(&mut *conn)
55            .await?;
56    let (pending_attachment_uploads, pending_attachment_upload_bytes): (i64, i64) = sqlx::query_as(
57        "SELECT COUNT(*), COALESCE(SUM(byte_size), 0)
58                 FROM (
59                   SELECT json_extract(payload, '$.workspace_id') AS workspace_id,
60                          json_extract(payload, '$.sha256') AS sha256,
61                          MAX(CAST(json_extract(payload, '$.byte_size') AS INTEGER)) AS byte_size
62                   FROM changes
63                   WHERE server_seq IS NULL AND op_type = 'attachment_add'
64                   GROUP BY workspace_id, sha256
65                 )",
66    )
67    .fetch_one(&mut *conn)
68    .await?;
69    let conflicts = sqlx::query_scalar("SELECT count(*) FROM conflicts WHERE resolved = 0")
70        .fetch_one(&mut *conn)
71        .await?;
72    Ok(SyncPersistenceStatus {
73        pinned_server: get_meta(conn, "sync_server_url").await?,
74        established_protocol: super::super::protocol::replica_protocol(conn).await?,
75        blocked_protocol: get_meta(conn, "sync_blocked_protocol")
76            .await?
77            .filter(|v| !v.is_empty())
78            .map(|v| v.parse())
79            .transpose()?,
80        pending_changes,
81        pending_attachment_uploads,
82        pending_attachment_upload_bytes,
83        conflicts,
84        sync_cursor: get_meta(conn, "sync_cursor").await?,
85        local_sequence: get_meta(conn, "local_seq").await?,
86        last_attempt: get_meta(conn, "sync_last_attempt_at").await?,
87        last_success: get_meta(conn, "sync_last_success_at").await?,
88        last_error: get_meta(conn, "sync_last_error").await?,
89        last_pushed: get_meta(conn, "sync_last_pushed").await?,
90        last_pulled: get_meta(conn, "sync_last_pulled").await?,
91        last_cursor: get_meta(conn, "sync_last_cursor").await?,
92    })
93}
aven_core/sync/persistence/status.rs

aven_core/sync/persistence/
status.rs
