1use std::path::Path;
2
3use anyhow::{Result, bail};
4use sqlx::SqliteConnection;
5use sqlx::{AssertSqlSafe, Row};
6
7use crate::attachments::AttachmentBytesState;
8use crate::attachments::decode::{ImageFacts, ValidatedImage, validate_image};
9use crate::attachments::optimization::{ImageOptimizationPolicy, optimize_image_bytes};
10use crate::attachments::storage::{blob_inventory_row, sha256_hex, store_validated_blob};
11use crate::attachments::validation::{
12 validate_alt_text, validate_attachment_id, validate_filename,
13};
14use crate::change_log::{ChangeEntity, ChangePayload, append_change, op_type};
15use crate::db::{Database, begin_immediate};
16use crate::ids::{TaskId, WorkspaceId, new_id, now};
17use crate::types::TaskAttachment;
18use crate::workspaces::Workspace;
19
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct AttachmentAddInput {
22 pub filename: Option<String>,
23 pub alt_text: Option<String>,
24 pub declared_media_type: Option<String>,
25 pub bytes: Vec<u8>,
26 pub optimization_policy: ImageOptimizationPolicy,
27 pub dedupe_existing: bool,
28}
29
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct TaskAttachmentAddInput {
32 pub attachment_id: String,
33 pub input: AttachmentAddInput,
34}
35
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct PreparedAttachment {
38 pub attachment_id: String,
39 pub filename: Option<String>,
40 pub alt_text: Option<String>,
41 pub sha256: String,
42 pub byte_size: i64,
43 pub facts: ImageFacts,
44 pub bytes: Vec<u8>,
45 pub optimized: bool,
46}
47
48pub struct AttachmentAddOutcome {
49 pub outcome: AttachmentOutcome,
50 pub created: bool,
51 pub optimized: bool,
52}
53
54pub struct AttachmentOutcome {
55 pub attachment: TaskAttachment,
56 pub has_blob: bool,
57}
58
59pub struct AttachmentReadLease {
60 pub sha256: String,
61 pub media_type: String,
62 pub lease_id: String,
63}
64
65#[derive(Debug, Clone)]
66pub struct AttachmentReadItem {
67 pub attachment: TaskAttachment,
68 pub bytes_state: AttachmentBytesState,
69 pub has_blob: bool,
70}
71
72const ATTACHMENT_COLUMNS: &str =
73 "ta.workspace_id, ta.attachment_id, ta.task_id, ta.sha256, ta.byte_size,
74 ta.media_type, ta.filename, ta.alt_text, ta.width, ta.height,
75 ta.created_at, ta.created_by_change_id, ta.deleted, ta.deleted_at, ta.deleted_by_change_id";
76
77fn attachment_from_row(row: &sqlx::sqlite::SqliteRow) -> TaskAttachment {
78 TaskAttachment {
79 workspace_id: row.get("workspace_id"),
80 attachment_id: row.get("attachment_id"),
81 task_id: row.get("task_id"),
82 sha256: row.get("sha256"),
83 byte_size: row.get("byte_size"),
84 media_type: row.get("media_type"),
85 filename: row.get("filename"),
86 alt_text: row.get("alt_text"),
87 width: row.get("width"),
88 height: row.get("height"),
89 created_at: row.get("created_at"),
90 created_by_change_id: row.get("created_by_change_id"),
91 deleted: row.get::<i64, _>("deleted") != 0,
92 deleted_at: row.get("deleted_at"),
93 deleted_by_change_id: row.get("deleted_by_change_id"),
94 }
95}
96
97async fn require_attachment(
98 conn: &mut SqliteConnection,
99 workspace_id: &WorkspaceId,
100 attachment_id: &str,
101) -> Result<TaskAttachment> {
102 let row = sqlx::query(AssertSqlSafe(format!(
103 "SELECT {ATTACHMENT_COLUMNS}
104 FROM task_attachments ta
105 WHERE ta.workspace_id = ? AND ta.attachment_id = ?"
106 )))
107 .bind(workspace_id)
108 .bind(attachment_id)
109 .fetch_optional(&mut *conn)
110 .await?;
111
112 let Some(row) = row else {
113 bail!("error attachment-not-found id={}", attachment_id);
114 };
115 Ok(attachment_from_row(&row))
116}
117
118async fn attachment_bytes_state(
119 conn: &mut SqliteConnection,
120 sha256: &str,
121) -> Result<AttachmentBytesState> {
122 Ok(match blob_inventory_row(conn, sha256).await? {
123 Some(row) if row.available => AttachmentBytesState::Present,
124 Some(_) => AttachmentBytesState::Unavailable,
125 None => AttachmentBytesState::PendingDownload,
126 })
127}
128
129async fn attachment_has_blob(conn: &mut SqliteConnection, sha256: &str) -> Result<bool> {
130 Ok(attachment_bytes_state(conn, sha256).await? == AttachmentBytesState::Present)
131}
132
133async fn existing_live_attachments_by_sha(
134 conn: &mut SqliteConnection,
135 workspace_id: &crate::ids::WorkspaceId,
136 task_id: &TaskId,
137 sha256: &str,
138) -> Result<Vec<TaskAttachment>> {
139 let rows = sqlx::query(AssertSqlSafe(format!(
140 "SELECT {ATTACHMENT_COLUMNS}
141 FROM task_attachments ta
142 WHERE ta.workspace_id = ? AND ta.task_id = ? AND ta.sha256 = ? AND ta.deleted = 0
143 ORDER BY ta.created_at, ta.attachment_id"
144 )))
145 .bind(workspace_id)
146 .bind(task_id)
147 .bind(sha256)
148 .fetch_all(&mut *conn)
149 .await?;
150
151 Ok(rows.iter().map(attachment_from_row).collect())
152}
153
154pub async fn prepare_task_attachment(input: TaskAttachmentAddInput) -> Result<PreparedAttachment> {
155 validate_attachment_id(&input.attachment_id)?;
156 validate_filename(input.input.filename.as_deref())?;
157 validate_alt_text(input.input.alt_text.as_deref())?;
158
159 let source = validate_image(input.input.bytes, input.input.declared_media_type).await?;
160 let source_facts = source.facts.clone();
161 let optimized = optimize_image_bytes(
162 &source_facts.media_type,
163 source.bytes,
164 input.input.optimization_policy,
165 )
166 .await?;
167 let optimized_flag = optimized.optimized;
168 let stored_image = if optimized_flag {
169 validate_image(optimized.bytes, None).await?
170 } else {
171 ValidatedImage {
172 bytes: optimized.bytes,
173 facts: source_facts,
174 }
175 };
176 let sha256 = sha256_hex(&stored_image.bytes);
177 let byte_size = i64::try_from(stored_image.bytes.len())?;
178 Ok(PreparedAttachment {
179 attachment_id: input.attachment_id,
180 filename: input.input.filename,
181 alt_text: input.input.alt_text,
182 sha256,
183 byte_size,
184 facts: stored_image.facts,
185 bytes: stored_image.bytes,
186 optimized: optimized_flag,
187 })
188}
189
190struct AttachmentRecordDraft<'a> {
191 attachment_id: &'a str,
192 sha256: &'a str,
193 byte_size: i64,
194 media_type: &'a str,
195 filename: Option<&'a str>,
196 alt_text: Option<&'a str>,
197 width: i64,
198 height: i64,
199 created_at: &'a str,
200}
201
202async fn record_attachment_add(
203 conn: &mut SqliteConnection,
204 workspace: &Workspace,
205 task_id: &TaskId,
206 draft: AttachmentRecordDraft<'_>,
207) -> Result<String> {
208 let change_id = append_change(
209 conn,
210 ChangeEntity::Task,
211 task_id,
212 Some("attachments"),
213 op_type::ATTACHMENT_ADD,
214 ChangePayload::workspace(workspace)
215 .set("attachment_id", draft.attachment_id)
216 .set("sha256", draft.sha256)
217 .set("byte_size", draft.byte_size)
218 .set("media_type", draft.media_type)
219 .set("filename", draft.filename)
220 .set("alt_text", draft.alt_text)
221 .set("width", draft.width)
222 .set("height", draft.height)
223 .set("created_at", draft.created_at),
224 )
225 .await?;
226 sqlx::query(
227 "INSERT INTO task_attachments(workspace_id, attachment_id, task_id, sha256, byte_size, media_type, filename, alt_text, width, height, created_at, created_by_change_id)
228 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
229 )
230 .bind(&workspace.id)
231 .bind(draft.attachment_id)
232 .bind(task_id)
233 .bind(draft.sha256)
234 .bind(draft.byte_size)
235 .bind(draft.media_type)
236 .bind(draft.filename)
237 .bind(draft.alt_text)
238 .bind(draft.width)
239 .bind(draft.height)
240 .bind(draft.created_at)
241 .bind(&change_id)
242 .execute(&mut *conn)
243 .await?;
244 Ok(change_id)
245}
246
247pub(super) async fn insert_prepared_attachment(
248 conn: &mut SqliteConnection,
249 workspace: &Workspace,
250 task_id: &TaskId,
251 prepared: &PreparedAttachment,
252 created_at: &str,
253) -> Result<String> {
254 record_attachment_add(
255 conn,
256 workspace,
257 task_id,
258 AttachmentRecordDraft {
259 attachment_id: &prepared.attachment_id,
260 sha256: &prepared.sha256,
261 byte_size: prepared.byte_size,
262 media_type: &prepared.facts.media_type,
263 filename: prepared.filename.as_deref(),
264 alt_text: prepared.alt_text.as_deref(),
265 width: prepared.facts.width,
266 height: prepared.facts.height,
267 created_at,
268 },
269 )
270 .await
271}
272
273struct AttachmentCommitIdentity {
274 attachment_id: Option<String>,
275 created_at: Option<String>,
276}
277
278pub async fn add_task_attachment(
279 conn: &mut SqliteConnection,
280 workspace: &Workspace,
281 blob_dir: &Path,
282 policy: crate::attachments::lifecycle::LifecyclePolicy,
283 task_id: &TaskId,
284 input: AttachmentAddInput,
285) -> Result<AttachmentAddOutcome> {
286 add_task_attachment_inner(
287 conn,
288 workspace,
289 blob_dir,
290 policy,
291 task_id,
292 AttachmentCommitIdentity {
293 attachment_id: None,
294 created_at: None,
295 },
296 input,
297 )
298 .await
299}
300
301pub(crate) async fn add_ordered_task_attachment(
302 conn: &mut SqliteConnection,
303 workspace: &Workspace,
304 blob_dir: &Path,
305 policy: crate::attachments::lifecycle::LifecyclePolicy,
306 task_id: &TaskId,
307 created_at: String,
308 input: TaskAttachmentAddInput,
309) -> Result<AttachmentAddOutcome> {
310 add_task_attachment_inner(
311 conn,
312 workspace,
313 blob_dir,
314 policy,
315 task_id,
316 AttachmentCommitIdentity {
317 attachment_id: Some(input.attachment_id),
318 created_at: Some(created_at),
319 },
320 input.input,
321 )
322 .await
323}
324
325async fn add_task_attachment_inner(
326 conn: &mut SqliteConnection,
327 workspace: &Workspace,
328 blob_dir: &Path,
329 policy: crate::attachments::lifecycle::LifecyclePolicy,
330 task_id: &TaskId,
331 identity: AttachmentCommitIdentity,
332 input: AttachmentAddInput,
333) -> Result<AttachmentAddOutcome> {
334 validate_filename(input.filename.as_deref())?;
335 validate_alt_text(input.alt_text.as_deref())?;
336
337 let source = validate_image(input.bytes, input.declared_media_type).await?;
338 let source_facts = source.facts.clone();
339 let optimized = optimize_image_bytes(
340 &source_facts.media_type,
341 source.bytes,
342 input.optimization_policy,
343 )
344 .await?;
345 let optimized_flag = optimized.optimized;
346 let stored_image = if optimized_flag {
347 validate_image(optimized.bytes, Some(source_facts.media_type.clone())).await?
348 } else {
349 ValidatedImage {
350 bytes: optimized.bytes,
351 facts: source_facts,
352 }
353 };
354 let sha256 = sha256_hex(&stored_image.bytes);
355 let byte_size = i64::try_from(stored_image.bytes.len())?;
356 let capacity_reservation = crate::attachments::lifecycle::ensure_local_capacity(
357 conn,
358 blob_dir,
359 &sha256,
360 byte_size,
361 policy,
362 &crate::attachments::lifecycle::SystemClock,
363 )
364 .await?;
365 let staging_lease = crate::attachments::lifecycle::acquire_lease(
366 conn,
367 &sha256,
368 "staging",
369 &crate::attachments::lifecycle::SystemClock,
370 )
371 .await?;
372 let stored = match store_validated_blob(conn, blob_dir, stored_image).await {
373 Ok(stored) => stored,
374 Err(error) => {
375 let _ = crate::attachments::lifecycle::release_lease(conn, &staging_lease).await;
376 if let Some(reservation_id) = capacity_reservation.as_deref() {
377 let _ =
378 crate::attachments::lifecycle::release_reservation(conn, reservation_id).await;
379 }
380 return Err(error);
381 }
382 };
383 if let Some(reservation_id) = capacity_reservation {
384 crate::attachments::lifecycle::release_reservation(conn, &reservation_id).await?;
385 }
386 let database_result = async {
387 let mut tx = begin_immediate(conn).await?;
388 crate::operations::route_recurrence_task_field(
389 &mut tx,
390 workspace,
391 task_id,
392 "attachments",
393 "",
394 )
395 .await?;
396 let task_exists = sqlx::query_scalar::<_, bool>(
397 "SELECT EXISTS(SELECT 1 FROM tasks WHERE workspace_id = ? AND id = ?)",
398 )
399 .bind(&workspace.id)
400 .bind(task_id)
401 .fetch_one(&mut *tx)
402 .await?;
403 if !task_exists {
404 bail!("error task-not-found task_id={task_id}");
405 }
406
407 if input.dedupe_existing
408 && let Some(existing) =
409 existing_live_attachments_by_sha(&mut tx, &workspace.id, task_id, &stored.sha256)
410 .await?
411 .first()
412 {
413 let attachment_id = existing.attachment_id.clone();
414 tx.commit().await?;
415 return Ok::<_, anyhow::Error>((attachment_id, false));
416 }
417
418 let attachment_id = identity.attachment_id.unwrap_or_else(new_id);
419 let created_at = identity.created_at.unwrap_or_else(now);
420 record_attachment_add(
421 &mut tx,
422 workspace,
423 task_id,
424 AttachmentRecordDraft {
425 attachment_id: &attachment_id,
426 sha256: &stored.sha256,
427 byte_size: stored.byte_size,
428 media_type: &stored.facts.media_type,
429 filename: input.filename.as_deref(),
430 alt_text: input.alt_text.as_deref(),
431 width: stored.facts.width,
432 height: stored.facts.height,
433 created_at: &created_at,
434 },
435 )
436 .await?;
437 tx.commit().await?;
438 Ok((attachment_id, true))
439 }
440 .await;
441 let release_result = crate::attachments::lifecycle::release_lease(conn, &staging_lease).await;
442 let (attachment_id, created) = match database_result {
443 Ok(result) => result,
444 Err(error) => {
445 crate::attachments::storage::remove_staged_blob_if_unreferenced(
446 conn,
447 blob_dir,
448 &stored.sha256,
449 )
450 .await;
451 return Err(error);
452 }
453 };
454 release_result?;
455 crate::attachments::lifecycle::reconcile_liveness(
456 conn,
457 &crate::attachments::lifecycle::SystemClock,
458 )
459 .await?;
460 let outcome = attachment_by_id(conn, workspace, &attachment_id).await?;
461 Ok(AttachmentAddOutcome {
462 outcome,
463 created,
464 optimized: optimized_flag,
465 })
466}
467
468pub async fn attachment_by_id(
469 conn: &mut SqliteConnection,
470 workspace: &Workspace,
471 attachment_id: &str,
472) -> Result<AttachmentOutcome> {
473 let attachment = require_attachment(conn, &workspace.id, attachment_id).await?;
474 let has_blob = attachment_has_blob(conn, &attachment.sha256).await?;
475 Ok(AttachmentOutcome {
476 attachment,
477 has_blob,
478 })
479}
480
481pub async fn delete_task_attachment(
482 conn: &mut SqliteConnection,
483 workspace: &Workspace,
484 attachment_id: &str,
485) -> Result<AttachmentOutcome> {
486 let attachment = require_attachment(conn, &workspace.id, attachment_id).await?;
487 if attachment.deleted {
488 let has_blob = attachment_has_blob(conn, &attachment.sha256).await?;
489 return Ok(AttachmentOutcome {
490 attachment,
491 has_blob,
492 });
493 }
494
495 let task_id = attachment.task_id.clone();
496 let deleted_at = now();
497
498 let mut tx = begin_immediate(conn).await?;
499 crate::operations::route_recurrence_task_field(&mut tx, workspace, &task_id, "attachments", "")
500 .await?;
501
502 let change_id = append_change(
503 &mut tx,
504 ChangeEntity::Task,
505 &task_id,
506 Some("attachments"),
507 op_type::ATTACHMENT_DELETE,
508 ChangePayload::workspace(workspace)
509 .set("attachment_id", attachment_id)
510 .set("filename", attachment.filename.as_deref())
511 .set("media_type", &attachment.media_type)
512 .set("deleted_at", &deleted_at),
513 )
514 .await?;
515
516 sqlx::query(
517 "UPDATE task_attachments SET deleted = 1, deleted_at = ?, deleted_by_change_id = ? WHERE workspace_id = ? AND attachment_id = ?",
518 )
519 .bind(&deleted_at)
520 .bind(&change_id)
521 .bind(&workspace.id)
522 .bind(attachment_id)
523 .execute(&mut *tx)
524 .await?;
525
526 tx.commit().await?;
527 crate::attachments::lifecycle::reconcile_liveness(
528 conn,
529 &crate::attachments::lifecycle::SystemClock,
530 )
531 .await?;
532
533 attachment_by_id(conn, workspace, attachment_id).await
534}
535
536pub async fn attachment_read_items_by_task(
537 conn: &mut SqliteConnection,
538 workspace_id: &str,
539 task_id: &str,
540 include_deleted: bool,
541) -> Result<Vec<AttachmentReadItem>> {
542 let attachments = attachments_by_task(conn, workspace_id, task_id, include_deleted).await?;
543 let mut items = Vec::with_capacity(attachments.len());
544 for attachment in attachments {
545 let bytes_state = attachment_bytes_state(conn, &attachment.sha256).await?;
546 items.push(AttachmentReadItem {
547 attachment,
548 has_blob: bytes_state == AttachmentBytesState::Present,
549 bytes_state,
550 });
551 }
552 Ok(items)
553}
554
555pub async fn attachments_by_task(
556 conn: &mut SqliteConnection,
557 workspace_id: &str,
558 task_id: &str,
559 include_deleted: bool,
560) -> Result<Vec<TaskAttachment>> {
561 let deleted_predicate = if include_deleted {
562 ""
563 } else {
564 " AND ta.deleted = 0"
565 };
566 let rows = sqlx::query(AssertSqlSafe(format!(
567 "SELECT {ATTACHMENT_COLUMNS}
568 FROM task_attachments ta
569 WHERE ta.workspace_id = ? AND ta.task_id = ?{deleted_predicate}
570 ORDER BY ta.created_at, ta.attachment_id"
571 )))
572 .bind(workspace_id)
573 .bind(task_id)
574 .fetch_all(&mut *conn)
575 .await?;
576
577 let mut attachments = Vec::with_capacity(rows.len());
578 for row in &rows {
579 attachments.push(attachment_from_row(row));
580 }
581 Ok(attachments)
582}
583
584impl Database {
585 pub async fn add_task_attachment(
586 &self,
587 workspace: &Workspace,
588 blob_dir: &Path,
589 policy: crate::attachments::lifecycle::LifecyclePolicy,
590 task_id: &TaskId,
591 input: AttachmentAddInput,
592 ) -> Result<AttachmentAddOutcome> {
593 let mut conn = self.acquire_writer().await?;
594 add_task_attachment(&mut conn, workspace, blob_dir, policy, task_id, input).await
595 }
596
597 pub async fn add_ordered_task_attachment(
598 &self,
599 workspace: &Workspace,
600 blob_dir: &Path,
601 policy: crate::attachments::lifecycle::LifecyclePolicy,
602 task_id: &TaskId,
603 created_at: String,
604 input: TaskAttachmentAddInput,
605 ) -> Result<AttachmentAddOutcome> {
606 let mut conn = self.acquire_writer().await?;
607 add_ordered_task_attachment(
608 &mut conn, workspace, blob_dir, policy, task_id, created_at, input,
609 )
610 .await
611 }
612
613 pub async fn attachment_by_id(
614 &self,
615 workspace: &Workspace,
616 attachment_id: &str,
617 ) -> Result<AttachmentOutcome> {
618 let mut conn = self.acquire_reader().await?;
619 attachment_by_id(&mut conn, workspace, attachment_id).await
620 }
621
622 pub async fn delete_task_attachment(
623 &self,
624 workspace: &Workspace,
625 attachment_id: &str,
626 ) -> Result<AttachmentOutcome> {
627 let mut conn = self.acquire_writer().await?;
628 delete_task_attachment(&mut conn, workspace, attachment_id).await
629 }
630
631 pub async fn attachment_read_items_by_task(
632 &self,
633 workspace_id: &WorkspaceId,
634 task_id: &TaskId,
635 include_deleted: bool,
636 ) -> Result<Vec<AttachmentReadItem>> {
637 let mut conn = self.acquire_reader().await?;
638 attachment_read_items_by_task(
639 &mut conn,
640 workspace_id.as_str(),
641 task_id.as_str(),
642 include_deleted,
643 )
644 .await
645 }
646
647 pub async fn prune_attachments(
648 &self,
649 blob_dir: &Path,
650 policy: crate::attachments::lifecycle::LifecyclePolicy,
651 apply: bool,
652 ) -> Result<crate::attachments::lifecycle::PruneSummary> {
653 let mut conn = self.acquire_writer().await?;
654 crate::attachments::lifecycle::prune(
655 &mut conn,
656 blob_dir,
657 policy,
658 apply,
659 &crate::attachments::lifecycle::SystemClock,
660 )
661 .await
662 }
663
664 pub async fn acquire_attachment_lease(&self, sha256: &str, purpose: &str) -> Result<String> {
665 let mut conn = self.acquire_writer().await?;
666 crate::attachments::lifecycle::acquire_lease(
667 &mut conn,
668 sha256,
669 purpose,
670 &crate::attachments::lifecycle::SystemClock,
671 )
672 .await
673 }
674
675 pub async fn acquire_live_attachment_read_lease(
676 &self,
677 workspace: &Workspace,
678 attachment_id: &str,
679 ) -> Result<AttachmentReadLease> {
680 let mut conn = self.acquire_writer().await?;
681 let row = sqlx::query(
682 "SELECT ta.sha256, ta.media_type, bi.available
683 FROM task_attachments ta
684 JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
685 LEFT JOIN blob_inventory bi ON bi.sha256 = ta.sha256
686 WHERE ta.workspace_id = ? AND ta.attachment_id = ?
687 AND ta.deleted = 0 AND t.deleted = 0",
688 )
689 .bind(&workspace.id)
690 .bind(attachment_id)
691 .fetch_optional(&mut *conn)
692 .await?
693 .ok_or_else(|| anyhow::anyhow!("error attachment-invalidated"))?;
694 if !row.try_get::<bool, _>("available").unwrap_or(false) {
695 bail!("error attachment-blob-unavailable");
696 }
697 let sha256: String = row.get("sha256");
698 let media_type: String = row.get("media_type");
699 let lease_id = crate::attachments::lifecycle::acquire_lease(
700 &mut conn,
701 &sha256,
702 "read",
703 &crate::attachments::lifecycle::SystemClock,
704 )
705 .await?;
706 Ok(AttachmentReadLease {
707 sha256,
708 media_type,
709 lease_id,
710 })
711 }
712
713 pub async fn release_attachment_lease(&self, lease_id: &str) -> Result<()> {
714 let mut conn = self.acquire_writer().await?;
715 crate::attachments::lifecycle::release_lease(&mut conn, lease_id).await
716 }
717
718 pub async fn attachment_lifecycle_report(
719 &self,
720 blob_dir: &Path,
721 policy: crate::attachments::lifecycle::LifecyclePolicy,
722 ) -> Result<crate::attachments::lifecycle::LifecycleReport> {
723 let mut conn = self.acquire_writer().await?;
724 crate::attachments::lifecycle::lifecycle_report(
725 &mut conn,
726 blob_dir,
727 policy,
728 &crate::attachments::lifecycle::SystemClock,
729 )
730 .await
731 }
732}
733
734#[cfg(test)]
735mod tests {
736 use std::time::Duration;
737
738 use super::*;
739
740 #[tokio::test]
741 async fn lifecycle_report_waits_for_guarded_writer() {
742 let temp = tempfile::tempdir().unwrap();
743 let database = Database::open(&temp.path().join("lifecycle.sqlite"))
744 .await
745 .unwrap();
746 let writer = database.acquire_writer().await.unwrap();
747
748 let reporting_database = database.clone();
749 let blob_dir = temp.path().join("blobs");
750 let mut report = tokio::spawn(async move {
751 reporting_database
752 .attachment_lifecycle_report(
753 &blob_dir,
754 crate::attachments::lifecycle::LifecyclePolicy::default(),
755 )
756 .await
757 });
758
759 assert!(
760 tokio::time::timeout(Duration::from_millis(100), &mut report)
761 .await
762 .is_err(),
763 "lifecycle reporting should wait for the guarded writer"
764 );
765
766 drop(writer);
767 tokio::time::timeout(Duration::from_secs(1), report)
768 .await
769 .expect("lifecycle reporting should proceed after the writer is released")
770 .unwrap()
771 .unwrap();
772 }
773}