aven_core/sync/
blob.rs

1use std::path::Path;
2
3use anyhow::{Context, Result};
4use sqlx::SqliteConnection;
5
6use crate::attachments::lifecycle::ByteCount;
7use crate::db::Database;
8
9impl Database {
10    /// Accepted attachments whose image bytes are missing on this device.
11    pub async fn missing_sync_attachment_counts(&self) -> Result<ByteCount> {
12        let mut conn = self.acquire_reader().await?;
13        missing_local_blob_counts(&mut conn).await
14    }
15
16    /// Live setup images whose object file is absent, regardless of the
17    /// inventory's last-known availability bit.
18    pub async fn missing_setup_attachment_counts(&self, blob_dir: &Path) -> Result<ByteCount> {
19        let mut conn = self.acquire_reader().await?;
20        let rows: Vec<(String, i64, String, String)> = sqlx::query_as(
21            "SELECT ta.sha256, MAX(ta.byte_size), MIN(t.title), MIN(ta.filename)
22             FROM task_attachments ta
23             JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
24             WHERE ta.deleted = 0 AND t.deleted = 0
25             GROUP BY ta.sha256 ORDER BY ta.sha256",
26        )
27        .fetch_all(&mut *conn)
28        .await?;
29        let mut missing = ByteCount::default();
30        for (sha256, byte_size, task, attachment) in rows {
31            let path = crate::attachments::storage::object_path(blob_dir, &sha256)?;
32            match std::fs::metadata(path) {
33                Ok(metadata) if metadata.is_file() => {}
34                Err(error) if error.kind() != std::io::ErrorKind::NotFound => {
35                    return Err(error).with_context(|| {
36                        format!("error setup-image-file task={task:?} attachment={attachment:?}")
37                    });
38                }
39                Ok(_) | Err(_) => {
40                    missing.count += 1;
41                    missing.bytes += u64::try_from(byte_size)?;
42                }
43            }
44        }
45        Ok(missing)
46    }
47}
48
49async fn missing_local_blob_counts(conn: &mut SqliteConnection) -> Result<ByteCount> {
50    let (count, bytes): (i64, i64) = sqlx::query_as(
51        "SELECT COUNT(*), COALESCE(SUM(byte_size), 0) FROM (
52           SELECT ta.sha256, MAX(ta.byte_size) AS byte_size
53           FROM task_attachments AS ta INDEXED BY idx_task_attachments_live_sha256
54           JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
55           JOIN changes c ON c.change_id = ta.created_by_change_id
56                         AND c.op_type = 'attachment_add' AND c.server_seq IS NOT NULL
57           LEFT JOIN blob_inventory bi ON bi.sha256 = ta.sha256
58           WHERE ta.deleted = 0 AND t.deleted = 0
59           GROUP BY ta.sha256
60           HAVING MAX(COALESCE(bi.available, 0)) = 0
61         )",
62    )
63    .fetch_one(&mut *conn)
64    .await?;
65    Ok(ByteCount {
66        count: u64::try_from(count)?,
67        bytes: u64::try_from(bytes)?,
68    })
69}
70
71#[cfg(test)]
72mod tests {
73    use crate::attachments::lifecycle::SystemClock;
74
75    use super::*;
76
77    async fn seed_missing_attachments(database: &Database, count: usize) {
78        let mut conn = database.acquire_writer().await.unwrap();
79        for index in 0..count {
80            let task_id = format!("{index:016X}");
81            let attachment_id = format!("{:016X}", index + 100);
82            let change_id = format!("{:016X}", index + 200);
83            let sha256 = format!("{index:064x}");
84            sqlx::query(
85                "INSERT INTO tasks(
86                   workspace_id, id, title, description, project_id, status, priority,
87                   created_at, updated_at, queue_activity_at
88                 ) VALUES ('0000000000000000', ?, 'task', '', 'project', 'inbox', 'none',
89                           '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z',
90                           '2026-01-01T00:00:00Z')",
91            )
92            .bind(&task_id)
93            .execute(&mut *conn)
94            .await
95            .unwrap();
96            sqlx::query(
97                "INSERT INTO changes(
98                   change_id, client_id, local_seq, entity_type, entity_id, field, op_type,
99                   payload, created_at, server_seq
100                 ) VALUES (?, 'remote', ?, 'task', ?, 'attachments', 'attachment_add', '{}',
101                           '2026-01-01T00:00:00Z', ?)",
102            )
103            .bind(&change_id)
104            .bind(i64::try_from(index + 1).unwrap())
105            .bind(&task_id)
106            .bind(i64::try_from(index + 1).unwrap())
107            .execute(&mut *conn)
108            .await
109            .unwrap();
110            sqlx::query(
111                "INSERT INTO task_attachments(
112                   workspace_id, attachment_id, task_id, sha256, byte_size, media_type,
113                   width, height, created_at, created_by_change_id
114                 ) VALUES ('0000000000000000', ?, ?, ?, 4, 'image/png', 1, 1,
115                           '2026-01-01T00:00:00Z', ?)",
116            )
117            .bind(attachment_id)
118            .bind(task_id)
119            .bind(sha256)
120            .bind(change_id)
121            .execute(&mut *conn)
122            .await
123            .unwrap();
124        }
125    }
126
127    #[tokio::test]
128    async fn imported_local_only_attachment_is_not_scheduled_for_download() {
129        let temp = tempfile::tempdir().unwrap();
130        let database = Database::open(&temp.path().join("replica.sqlite"))
131            .await
132            .unwrap();
133        let mut conn = database.acquire_writer().await.unwrap();
134        sqlx::query(
135            "INSERT INTO tasks(
136               workspace_id, id, title, description, project_id, status, priority,
137               created_at, updated_at, queue_activity_at
138             ) VALUES ('0000000000000000', '0000000000000001', 'task', '', 'project',
139                       'inbox', 'none', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z',
140                       '2026-01-01T00:00:00Z')",
141        )
142        .execute(&mut *conn)
143        .await
144        .unwrap();
145        sqlx::query(
146            "INSERT INTO task_attachments(
147               workspace_id, attachment_id, task_id, sha256, byte_size, media_type,
148               width, height, created_at, created_by_change_id
149             ) VALUES ('0000000000000000', '0000000000000002', '0000000000000001',
150                       'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
151                       4, 'image/png', 1, 1, '2026-01-01T00:00:00Z', NULL)",
152        )
153        .execute(&mut *conn)
154        .await
155        .unwrap();
156        drop(conn);
157
158        assert_eq!(
159            database.missing_sync_attachment_counts().await.unwrap(),
160            ByteCount { count: 0, bytes: 0 }
161        );
162    }
163
164    #[tokio::test]
165    async fn missing_counts_follow_inventory_and_object_reconciliation() {
166        let temp = tempfile::tempdir().unwrap();
167        let database = Database::open(&temp.path().join("replica.sqlite"))
168            .await
169            .unwrap();
170        seed_missing_attachments(&database, 3).await;
171        {
172            let mut conn = database.acquire_writer().await.unwrap();
173            crate::attachments::storage::upsert_inventory_available(
174                &mut conn,
175                &format!("{:064x}", 0),
176                4,
177                "image/png",
178            )
179            .await
180            .unwrap();
181        }
182
183        assert_eq!(
184            database.missing_sync_attachment_counts().await.unwrap(),
185            ByteCount { count: 2, bytes: 8 }
186        );
187
188        let blob_dir = temp.path().join("blobs");
189        assert_eq!(
190            database
191                .missing_setup_attachment_counts(&blob_dir)
192                .await
193                .unwrap(),
194            ByteCount {
195                count: 3,
196                bytes: 12
197            }
198        );
199        let mut conn = database.acquire_writer().await.unwrap();
200        let missing = crate::attachments::lifecycle::reconcile_missing_objects(
201            &mut conn,
202            &blob_dir,
203            &SystemClock,
204        )
205        .await
206        .unwrap();
207        assert_eq!(missing.count, 1);
208        drop(conn);
209
210        assert_eq!(
211            database.missing_sync_attachment_counts().await.unwrap(),
212            ByteCount {
213                count: 3,
214                bytes: 12
215            }
216        );
217    }
218}
aven_core/sync/blob.rs

aven_core/sync/
blob.rs
