1use std::collections::HashSet;
2use std::path::Path;
3
4use anyhow::{Context, Result, bail};
5use sqlx::{Row, SqliteConnection};
6
7use super::wire::{AttachmentAddPayload, BlobUploadContract, ChangeWire};
8use crate::attachments::decode::validate_image;
9use crate::attachments::lifecycle::{
10 ByteCount, LifecyclePolicy, PruneSummary, SystemClock, acquire_lease, ensure_local_capacity,
11 prune, release_lease, release_reservation, reserve_upload,
12};
13use crate::attachments::storage::{
14 blob_inventory_row, object_path, sha256_hex, store_validated_blob,
15};
16use crate::change_log::op_type;
17use crate::db::Database;
18
19#[derive(Debug, Clone)]
20pub(super) struct MissingLocalBlob {
21 pub sha256: String,
22 pub byte_size: i64,
23 pub media_type: String,
24 pub width: Option<i64>,
25 pub height: Option<i64>,
26}
27
28#[derive(Debug)]
29pub(super) struct MissingBlobPage {
30 pub blobs: Vec<MissingLocalBlob>,
31 pub total: ByteCount,
32}
33
34#[derive(Debug)]
35pub(super) struct PreparedBlobUpload {
36 pub bytes: Vec<u8>,
37 pub lease_id: String,
38}
39
40#[derive(Debug)]
41pub struct ServerBlobDownload {
42 pub bytes: Vec<u8>,
43}
44
45pub(super) fn attachment_blob_contract(change: &ChangeWire) -> Result<Option<BlobUploadContract>> {
46 if change.op_type != op_type::ATTACHMENT_ADD {
47 return Ok(None);
48 }
49 let payload = AttachmentAddPayload::from_change(change)?;
50 Ok(Some(BlobUploadContract {
51 workspace_id: payload.workspace_id,
52 sha256: payload.sha256,
53 byte_size: payload.byte_size,
54 media_type: payload.media_type,
55 width: payload
56 .width
57 .context("validated attachment payload missing width")?,
58 height: payload
59 .height
60 .context("validated attachment payload missing height")?,
61 }))
62}
63
64pub(super) fn unique_blob_contracts(changes: &[ChangeWire]) -> Result<Vec<BlobUploadContract>> {
65 let mut seen = HashSet::new();
66 let mut contracts = Vec::new();
67 for change in changes {
68 if let Some(contract) = attachment_blob_contract(change)?
69 && seen.insert((contract.workspace_id.clone(), contract.sha256.clone()))
70 {
71 contracts.push(contract);
72 }
73 }
74 Ok(contracts)
75}
76
77pub(super) fn contract_by_hash(
78 contracts: &[BlobUploadContract],
79 sha256: &str,
80) -> Result<BlobUploadContract> {
81 contracts
82 .iter()
83 .find(|contract| contract.sha256 == sha256)
84 .cloned()
85 .context("payload missing attachment blob contract")
86}
87
88impl Database {
89 pub(super) async fn prepare_blob_upload(
90 &self,
91 blob_dir: &Path,
92 contract: &BlobUploadContract,
93 ) -> Result<PreparedBlobUpload> {
94 let (row, lease_id) = {
95 let mut conn = self.acquire_writer().await?;
96 let row = blob_inventory_row(&mut conn, &contract.sha256)
97 .await?
98 .filter(|row| row.available)
99 .context("error attachment-blob-local-missing")?;
100 let lease_id =
101 acquire_lease(&mut conn, &contract.sha256, "transfer", &SystemClock).await?;
102 (row, lease_id)
103 };
104 let result = async {
105 let bytes = tokio::fs::read(object_path(blob_dir, &contract.sha256)?)
106 .await
107 .context("error attachment-blob-local-missing")?;
108 if sha256_hex(&bytes) != contract.sha256
109 || row.byte_size
110 != i64::try_from(bytes.len()).context("attachment bytes exceed i64")?
111 {
112 bail!("error attachment-blob-local-invalid");
113 }
114 let validated = validate_image(bytes.clone(), Some(row.media_type.clone())).await?;
115 if (validated.facts.width, validated.facts.height) != (contract.width, contract.height)
116 || validated.facts.media_type != contract.media_type
117 || contract.byte_size != row.byte_size
118 {
119 bail!("error attachment-blob-local-invalid");
120 }
121 Ok(PreparedBlobUpload {
122 bytes,
123 lease_id: lease_id.clone(),
124 })
125 }
126 .await;
127 if result.is_err() {
128 let _ = self.finish_blob_upload(&lease_id).await;
129 }
130 result
131 }
132
133 pub(super) async fn finish_blob_upload(&self, lease_id: &str) -> Result<()> {
134 let mut conn = self.acquire_writer().await?;
135 release_lease(&mut conn, lease_id).await
136 }
137
138 pub(super) async fn missing_local_blob_page(
139 &self,
140 blob_dir: &Path,
141 limit: usize,
142 ) -> Result<MissingBlobPage> {
143 let mut conn = self.acquire_reader().await?;
144 missing_local_blob_page(&mut conn, blob_dir, limit).await
145 }
146
147 pub(super) async fn store_downloaded_blob(
148 &self,
149 blob_dir: &Path,
150 policy: LifecyclePolicy,
151 blob: &MissingLocalBlob,
152 bytes: Vec<u8>,
153 ) -> Result<()> {
154 let expected = usize::try_from(blob.byte_size).context("attachment bytes exceed usize")?;
155 if bytes.len() != expected || sha256_hex(&bytes) != blob.sha256 {
156 bail!("error attachment-blob-remote-invalid");
157 }
158 let validated = validate_image(bytes, Some(blob.media_type.clone()))
159 .await
160 .context("error attachment-blob-remote-invalid")?;
161 if (blob.width, blob.height) != (Some(validated.facts.width), Some(validated.facts.height))
162 {
163 bail!("error attachment-blob-remote-invalid");
164 }
165 let mut conn = self.acquire_writer().await?;
166 let reservation = ensure_local_capacity(
167 &mut conn,
168 blob_dir,
169 &blob.sha256,
170 blob.byte_size,
171 policy,
172 &SystemClock,
173 )
174 .await?;
175 let result = store_validated_blob(&mut conn, blob_dir, validated).await;
176 if let Some(reservation) = reservation {
177 release_reservation(&mut conn, &reservation).await?;
178 }
179 result.map(|_| ())
180 }
181
182 pub async fn prepare_server_blob_uploads(
183 &self,
184 blob_dir: &Path,
185 policy: LifecyclePolicy,
186 blobs: &[BlobUploadContract],
187 ) -> Result<Vec<String>> {
188 super::wire::validate_blob_contracts(blobs)?;
189 let mut conn = self.acquire_writer().await?;
190 let mut missing = Vec::new();
191 let mut missing_hashes = HashSet::new();
192 for blob in blobs {
193 if blob_available(&mut conn, blob_dir, &blob.sha256).await? {
194 reserve_upload(
195 &mut conn,
196 &blob.workspace_id,
197 &blob.sha256,
198 blob.byte_size,
199 policy.quota_bytes,
200 &SystemClock,
201 )
202 .await?;
203 } else if missing_hashes.insert(blob.sha256.as_str()) {
204 missing.push(blob.sha256.clone());
205 }
206 }
207 Ok(missing)
208 }
209
210 pub async fn store_server_blob(
211 &self,
212 blob_dir: &Path,
213 policy: LifecyclePolicy,
214 contract: &BlobUploadContract,
215 bytes: Vec<u8>,
216 ) -> Result<()> {
217 super::wire::validate_blob_contracts(std::slice::from_ref(contract))?;
218 if sha256_hex(&bytes) != contract.sha256
219 || usize::try_from(contract.byte_size).ok() != Some(bytes.len())
220 {
221 bail!("error blob-hash-or-size-mismatch");
222 }
223 let validated = validate_image(bytes, Some(contract.media_type.clone()))
224 .await
225 .context("error blob-validation-failed")?;
226 if (validated.facts.width, validated.facts.height) != (contract.width, contract.height) {
227 bail!("error blob-validation-failed");
228 }
229 let mut conn = self.acquire_writer().await?;
230 let reservation = reserve_upload(
231 &mut conn,
232 &contract.workspace_id,
233 &contract.sha256,
234 contract.byte_size,
235 policy.quota_bytes,
236 &SystemClock,
237 )
238 .await?;
239 let result = store_validated_blob(&mut conn, blob_dir, validated).await;
240 if result.is_err()
241 && let Some(reservation) = reservation
242 {
243 release_reservation(&mut conn, &reservation).await?;
244 }
245 result.map(|_| ())
246 }
247
248 pub async fn read_server_blob(
249 &self,
250 blob_dir: &Path,
251 sha256: &str,
252 ) -> Result<Option<ServerBlobDownload>> {
253 super::wire::validate_blob_hashes(&[sha256.to_string()])?;
254 let lease = {
255 let mut conn = self.acquire_writer().await?;
256 if !blob_available(&mut conn, blob_dir, sha256).await? {
257 return Ok(None);
258 }
259 acquire_lease(&mut conn, sha256, "transfer", &SystemClock).await?
260 };
261 let result = tokio::fs::read(object_path(blob_dir, sha256)?).await;
262 let mut conn = self.acquire_writer().await?;
263 release_lease(&mut conn, &lease).await?;
264 Ok(Some(ServerBlobDownload {
265 bytes: result.context("error blob-read-failed")?,
266 }))
267 }
268
269 pub async fn maintain_server_blobs(
270 &self,
271 blob_dir: &Path,
272 policy: LifecyclePolicy,
273 ) -> Result<PruneSummary> {
274 let mut conn = self.acquire_writer().await?;
275 prune(&mut conn, blob_dir, policy, true, &SystemClock).await
276 }
277}
278
279async fn missing_local_blob_page(
280 conn: &mut SqliteConnection,
281 blob_dir: &Path,
282 limit: usize,
283) -> Result<MissingBlobPage> {
284 let rows = sqlx::query(
285 "SELECT ta.sha256, MAX(ta.byte_size) AS byte_size,
286 MAX(ta.media_type) AS media_type, MAX(ta.width) AS width,
287 MAX(ta.height) AS height, MAX(COALESCE(bi.available, 0)) AS available
288 FROM task_attachments ta
289 JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
290 LEFT JOIN blob_inventory bi ON bi.sha256 = ta.sha256
291 WHERE ta.deleted = 0 AND t.deleted = 0
292 GROUP BY ta.sha256
293 ORDER BY ta.sha256",
294 )
295 .fetch_all(&mut *conn)
296 .await?;
297 let mut blobs = Vec::new();
298 let mut total = ByteCount::default();
299 for row in rows {
300 let sha256: String = row.get("sha256");
301 let available: i64 = row.get("available");
302 if available != 0 && object_path(blob_dir, &sha256)?.exists() {
303 continue;
304 }
305 let blob = MissingLocalBlob {
306 sha256,
307 byte_size: row.get("byte_size"),
308 media_type: row.get("media_type"),
309 width: row.get("width"),
310 height: row.get("height"),
311 };
312 total.count += 1;
313 total.bytes += u64::try_from(blob.byte_size)?;
314 if blobs.len() < limit {
315 blobs.push(blob);
316 }
317 }
318 Ok(MissingBlobPage { blobs, total })
319}
320
321pub(super) async fn blob_available(
322 conn: &mut SqliteConnection,
323 blob_dir: &Path,
324 sha256: &str,
325) -> Result<bool> {
326 let Some(row) = blob_inventory_row(conn, sha256).await? else {
327 return Ok(false);
328 };
329 Ok(row.available && object_path(blob_dir, sha256)?.exists())
330}
331
332#[cfg(test)]
333mod tests {
334 use serde_json::json;
335
336 use super::*;
337
338 fn attachment_change(op_type: &str) -> ChangeWire {
339 ChangeWire {
340 change_id: "AAAAAAAAAAAAAAA0".to_string(),
341 client_id: "client".to_string(),
342 local_seq: 1,
343 entity_type: "task".to_string(),
344 entity_id: "BBBBBBBBBBBBBBBB".to_string(),
345 field: Some("attachments".to_string()),
346 op_type: op_type.to_string(),
347 payload: json!({
348 "workspace_id": "0000000000000000",
349 "workspace_key": "default",
350 "attachment_id": "7KQ9A1X4MV2P8D6R",
351 "sha256": "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789",
352 "byte_size": 12,
353 "media_type": "image/png",
354 "filename": null,
355 "alt_text": null,
356 "width": 320,
357 "height": 240,
358 "created_at": "2026-06-01T00:00:00Z"
359 }),
360 base_version: None,
361 created_at: "2026-06-01T00:00:00Z".to_string(),
362 server_seq: None,
363 }
364 }
365
366 #[test]
367 fn attachment_blob_contract_uses_typed_add_payload() {
368 let add = attachment_change(op_type::ATTACHMENT_ADD);
369 let contract = attachment_blob_contract(&add).unwrap().unwrap();
370 assert_eq!(contract.workspace_id, "0000000000000000");
371 assert_eq!(
372 contract.sha256,
373 "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"
374 );
375 assert_eq!(contract.byte_size, 12);
376 assert_eq!(contract.media_type, "image/png");
377 assert_eq!((contract.width, contract.height), (320, 240));
378
379 let mut malformed = add;
380 malformed.payload["sha256"] = json!("invalid");
381 assert!(
382 attachment_blob_contract(&malformed)
383 .unwrap_err()
384 .to_string()
385 .contains("invalid-sha256")
386 );
387
388 assert!(
389 attachment_blob_contract(&attachment_change(op_type::ATTACHMENT_DELETE))
390 .unwrap()
391 .is_none()
392 );
393 assert!(
394 attachment_blob_contract(&attachment_change(op_type::NOTE_ADD))
395 .unwrap()
396 .is_none()
397 );
398 }
399
400 async fn seed_missing_attachments(database: &Database, count: usize) {
401 let mut conn = database.acquire_writer().await.unwrap();
402 for index in 0..count {
403 let task_id = format!("{index:016X}");
404 let attachment_id = format!("{:016X}", index + 100);
405 let sha256 = format!("{index:064x}");
406 sqlx::query(
407 "INSERT INTO tasks(
408 workspace_id, id, title, description, project_id, status, priority,
409 created_at, updated_at, queue_activity_at
410 ) VALUES ('0000000000000000', ?, 'task', '', 'project', 'inbox', 'none',
411 '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z',
412 '2026-01-01T00:00:00Z')",
413 )
414 .bind(&task_id)
415 .execute(&mut *conn)
416 .await
417 .unwrap();
418 sqlx::query(
419 "INSERT INTO task_attachments(
420 workspace_id, attachment_id, task_id, sha256, byte_size, media_type,
421 width, height, created_at
422 ) VALUES ('0000000000000000', ?, ?, ?, 4, 'image/png', 1, 1,
423 '2026-01-01T00:00:00Z')",
424 )
425 .bind(attachment_id)
426 .bind(task_id)
427 .bind(sha256)
428 .execute(&mut *conn)
429 .await
430 .unwrap();
431 }
432 }
433
434 #[tokio::test]
435 async fn missing_blob_page_is_bounded_and_reports_more() {
436 let temp = tempfile::tempdir().unwrap();
437 let database = Database::open(&temp.path().join("replica.sqlite"))
438 .await
439 .unwrap();
440 seed_missing_attachments(&database, 3).await;
441 {
442 let mut conn = database.acquire_writer().await.unwrap();
443 crate::attachments::storage::upsert_inventory_available(
444 &mut conn,
445 &format!("{:064x}", 0),
446 4,
447 "image/png",
448 )
449 .await
450 .unwrap();
451 }
452
453 let page = database
454 .missing_local_blob_page(&temp.path().join("blobs"), 2)
455 .await
456 .unwrap();
457 assert_eq!(page.blobs.len(), 2);
458 assert_eq!(
459 page.total,
460 ByteCount {
461 count: 3,
462 bytes: 12
463 }
464 );
465 }
466}