Skip to main content

max / goingson

6.1 KB · 179 lines History Blame Raw
1 //! Blob sync: upload local blobs to SyncKit, download missing blobs from SyncKit.
2
3 use std::path::Path;
4
5 use goingson_core::CoreError;
6 use sha2::{Digest, Sha256};
7 use sqlx::SqlitePool;
8 use synckit_client::SyncKitClient;
9 use tracing::{debug, info, warn};
10
11 use crate::commands::attachment::blob_path;
12 use crate::state::DESKTOP_USER_ID;
13
14 /// First 8 chars of a blob hash for log lines. Panic-safe: a `blob_hash` can arrive
15 /// via sync and need not be a clean 64-hex string, so a raw `&hash[..8]` could panic
16 /// on a short or multibyte value (ultra-fuzz Run #27 Data minor).
17 fn short(hash: &str) -> &str {
18 hash.get(..8).unwrap_or(hash)
19 }
20
21 /// Upload local blobs that haven't been synced to the server yet.
22 ///
23 /// Queries all distinct blob hashes from attachments, checks which ones have local
24 /// files, and uploads them via the SyncKit blob API (presigned S3 + E2E encryption).
25 #[tracing::instrument(skip_all)]
26 pub async fn upload_pending_blobs(
27 pool: &SqlitePool,
28 data_dir: &Path,
29 client: &SyncKitClient,
30 ) -> Result<i64, CoreError> {
31 let hashes: Vec<(String, i64)> = sqlx::query_as(
32 "SELECT DISTINCT a.blob_hash, a.file_size FROM attachments a WHERE a.user_id = ?"
33 )
34 .bind(DESKTOP_USER_ID.to_string())
35 .fetch_all(pool)
36 .await
37 .map_err(CoreError::database)?;
38
39 let mut uploaded = 0i64;
40
41 for (hash, size) in &hashes {
42 let path = blob_path(data_dir, hash);
43 if !tokio::fs::try_exists(&path).await.unwrap_or(false) {
44 continue; // No local blob to upload
45 }
46
47 // Request upload URL — server tells us if blob already exists
48 let upload_resp = match client.blob_upload_url(hash, *size).await {
49 Ok(r) => r,
50 Err(e) => {
51 warn!("Failed to get upload URL for blob {}: {}", short(hash), e);
52 continue;
53 }
54 };
55
56 if upload_resp.already_exists {
57 debug!("Blob {} already on server, skipping", short(hash));
58 continue;
59 }
60
61 // Read and upload
62 let data = match tokio::fs::read(&path).await {
63 Ok(d) => d,
64 Err(e) => {
65 warn!("Failed to read blob {}: {}", short(hash), e);
66 continue;
67 }
68 };
69
70 if let Err(e) = client.blob_upload(hash, &upload_resp.upload_url, data).await {
71 warn!("Failed to upload blob {}: {}", short(hash), e);
72 continue;
73 }
74
75 // Confirm upload
76 if let Err(e) = client.blob_confirm(hash, *size).await {
77 warn!("Failed to confirm blob {}: {}", short(hash), e);
78 continue;
79 }
80
81 uploaded += 1;
82 debug!("Uploaded blob {}", short(hash));
83 }
84
85 if uploaded > 0 {
86 info!("Uploaded {} blobs", uploaded);
87 }
88 Ok(uploaded)
89 }
90
91 /// Download blobs that exist in attachment records but not on local disk.
92 ///
93 /// After metadata sync pulls attachment records from other devices, this function
94 /// downloads the actual blob data from SyncKit (presigned S3 + E2E decryption).
95 #[tracing::instrument(skip_all)]
96 pub async fn download_missing_blobs(
97 pool: &SqlitePool,
98 data_dir: &Path,
99 client: &SyncKitClient,
100 ) -> Result<i64, CoreError> {
101 let hashes: Vec<(String,)> = sqlx::query_as(
102 "SELECT DISTINCT blob_hash FROM attachments WHERE user_id = ?"
103 )
104 .bind(DESKTOP_USER_ID.to_string())
105 .fetch_all(pool)
106 .await
107 .map_err(CoreError::database)?;
108
109 let blobs_dir = data_dir.join("blobs");
110 tokio::fs::create_dir_all(&blobs_dir)
111 .await
112 .map_err(|e| CoreError::internal(format!("Failed to create blobs dir: {}", e)))?;
113
114 let mut downloaded = 0i64;
115
116 for (hash,) in &hashes {
117 let path = blob_path(data_dir, hash);
118 if tokio::fs::try_exists(&path).await.unwrap_or(false) {
119 continue; // Already have it locally
120 }
121
122 // Get download URL
123 let download_url = match client.blob_download_url(hash).await {
124 Ok(url) => url,
125 Err(e) => {
126 warn!("Failed to get download URL for blob {}: {}", short(hash), e);
127 continue;
128 }
129 };
130
131 // Download and decrypt. The SDK now AAD-binds and re-verifies the content
132 // hash itself; the local check below stays as belt-and-braces before the
133 // file is committed under its hash name.
134 let data = match client.blob_download(hash, &download_url).await {
135 Ok(d) => d,
136 Err(e) => {
137 warn!("Failed to download blob {}: {}", short(hash), e);
138 continue;
139 }
140 };
141
142 // Verify content-addressed integrity before committing the file. The store's
143 // invariant is "the file at blobs/<hash> hashes to <hash>" — enforced on write
144 // (add_attachment) but previously only assumed on read. E2E AEAD stops a network
145 // attacker, but a server-side corruption or mis-bound ciphertext would otherwise
146 // land wrong bytes under a trusted name and be handed to open::that()
147 // (ultra-fuzz Run #28 S1). Hashing here, before the rename, keeps the store clean.
148 let actual = format!("{:x}", Sha256::digest(&data));
149 if actual != *hash {
150 warn!(
151 "Blob {} failed integrity check (content hashed to {}); discarding download",
152 short(hash),
153 short(&actual)
154 );
155 continue;
156 }
157
158 // Write to disk atomically (tmp + rename) to prevent corrupt partial files
159 let tmp_path = path.with_extension("tmp");
160 if let Err(e) = tokio::fs::write(&tmp_path, &data).await {
161 warn!("Failed to write blob {}: {}", short(hash), e);
162 continue;
163 }
164 if let Err(e) = tokio::fs::rename(&tmp_path, &path).await {
165 warn!("Failed to rename blob {}: {}", short(hash), e);
166 let _ = tokio::fs::remove_file(&tmp_path).await;
167 continue;
168 }
169
170 downloaded += 1;
171 debug!("Downloaded blob {}", short(hash));
172 }
173
174 if downloaded > 0 {
175 info!("Downloaded {} blobs", downloaded);
176 }
177 Ok(downloaded)
178 }
179