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("deleted_at", &deleted_at),
511 )
512 .await?;
513
514 sqlx::query(
515 "UPDATE task_attachments SET deleted = 1, deleted_at = ?, deleted_by_change_id = ? WHERE workspace_id = ? AND attachment_id = ?",
516 )
517 .bind(&deleted_at)
518 .bind(&change_id)
519 .bind(&workspace.id)
520 .bind(attachment_id)
521 .execute(&mut *tx)
522 .await?;
523
524 tx.commit().await?;
525 crate::attachments::lifecycle::reconcile_liveness(
526 conn,
527 &crate::attachments::lifecycle::SystemClock,
528 )
529 .await?;
530
531 attachment_by_id(conn, workspace, attachment_id).await
532}
533
534pub async fn attachment_read_items_by_task(
535 conn: &mut SqliteConnection,
536 workspace_id: &str,
537 task_id: &str,
538 include_deleted: bool,
539) -> Result<Vec<AttachmentReadItem>> {
540 let attachments = attachments_by_task(conn, workspace_id, task_id, include_deleted).await?;
541 let mut items = Vec::with_capacity(attachments.len());
542 for attachment in attachments {
543 let bytes_state = attachment_bytes_state(conn, &attachment.sha256).await?;
544 items.push(AttachmentReadItem {
545 attachment,
546 has_blob: bytes_state == AttachmentBytesState::Present,
547 bytes_state,
548 });
549 }
550 Ok(items)
551}
552
553pub async fn attachments_by_task(
554 conn: &mut SqliteConnection,
555 workspace_id: &str,
556 task_id: &str,
557 include_deleted: bool,
558) -> Result<Vec<TaskAttachment>> {
559 let deleted_predicate = if include_deleted {
560 ""
561 } else {
562 " AND ta.deleted = 0"
563 };
564 let rows = sqlx::query(AssertSqlSafe(format!(
565 "SELECT {ATTACHMENT_COLUMNS}
566 FROM task_attachments ta
567 WHERE ta.workspace_id = ? AND ta.task_id = ?{deleted_predicate}
568 ORDER BY ta.created_at, ta.attachment_id"
569 )))
570 .bind(workspace_id)
571 .bind(task_id)
572 .fetch_all(&mut *conn)
573 .await?;
574
575 let mut attachments = Vec::with_capacity(rows.len());
576 for row in &rows {
577 attachments.push(attachment_from_row(row));
578 }
579 Ok(attachments)
580}
581
582impl Database {
583 pub async fn add_task_attachment(
584 &self,
585 workspace: &Workspace,
586 blob_dir: &Path,
587 policy: crate::attachments::lifecycle::LifecyclePolicy,
588 task_id: &TaskId,
589 input: AttachmentAddInput,
590 ) -> Result<AttachmentAddOutcome> {
591 let mut conn = self.acquire_writer().await?;
592 add_task_attachment(&mut conn, workspace, blob_dir, policy, task_id, input).await
593 }
594
595 pub async fn add_ordered_task_attachment(
596 &self,
597 workspace: &Workspace,
598 blob_dir: &Path,
599 policy: crate::attachments::lifecycle::LifecyclePolicy,
600 task_id: &TaskId,
601 created_at: String,
602 input: TaskAttachmentAddInput,
603 ) -> Result<AttachmentAddOutcome> {
604 let mut conn = self.acquire_writer().await?;
605 add_ordered_task_attachment(
606 &mut conn, workspace, blob_dir, policy, task_id, created_at, input,
607 )
608 .await
609 }
610
611 pub async fn attachment_by_id(
612 &self,
613 workspace: &Workspace,
614 attachment_id: &str,
615 ) -> Result<AttachmentOutcome> {
616 let mut conn = self.acquire_reader().await?;
617 attachment_by_id(&mut conn, workspace, attachment_id).await
618 }
619
620 pub async fn delete_task_attachment(
621 &self,
622 workspace: &Workspace,
623 attachment_id: &str,
624 ) -> Result<AttachmentOutcome> {
625 let mut conn = self.acquire_writer().await?;
626 delete_task_attachment(&mut conn, workspace, attachment_id).await
627 }
628
629 pub async fn attachment_read_items_by_task(
630 &self,
631 workspace_id: &WorkspaceId,
632 task_id: &TaskId,
633 include_deleted: bool,
634 ) -> Result<Vec<AttachmentReadItem>> {
635 let mut conn = self.acquire_reader().await?;
636 attachment_read_items_by_task(
637 &mut conn,
638 workspace_id.as_str(),
639 task_id.as_str(),
640 include_deleted,
641 )
642 .await
643 }
644
645 pub async fn prune_attachments(
646 &self,
647 blob_dir: &Path,
648 policy: crate::attachments::lifecycle::LifecyclePolicy,
649 apply: bool,
650 ) -> Result<crate::attachments::lifecycle::PruneSummary> {
651 let mut conn = self.acquire_writer().await?;
652 crate::attachments::lifecycle::prune(
653 &mut conn,
654 blob_dir,
655 policy,
656 apply,
657 &crate::attachments::lifecycle::SystemClock,
658 )
659 .await
660 }
661
662 pub async fn acquire_attachment_lease(&self, sha256: &str, purpose: &str) -> Result<String> {
663 let mut conn = self.acquire_writer().await?;
664 crate::attachments::lifecycle::acquire_lease(
665 &mut conn,
666 sha256,
667 purpose,
668 &crate::attachments::lifecycle::SystemClock,
669 )
670 .await
671 }
672
673 pub async fn acquire_live_attachment_read_lease(
674 &self,
675 workspace: &Workspace,
676 attachment_id: &str,
677 ) -> Result<AttachmentReadLease> {
678 let mut conn = self.acquire_writer().await?;
679 let row = sqlx::query(
680 "SELECT ta.sha256, ta.media_type, bi.available
681 FROM task_attachments ta
682 JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
683 LEFT JOIN blob_inventory bi ON bi.sha256 = ta.sha256
684 WHERE ta.workspace_id = ? AND ta.attachment_id = ?
685 AND ta.deleted = 0 AND t.deleted = 0",
686 )
687 .bind(&workspace.id)
688 .bind(attachment_id)
689 .fetch_optional(&mut *conn)
690 .await?
691 .ok_or_else(|| anyhow::anyhow!("error attachment-invalidated"))?;
692 if !row.try_get::<bool, _>("available").unwrap_or(false) {
693 bail!("error attachment-blob-unavailable");
694 }
695 let sha256: String = row.get("sha256");
696 let media_type: String = row.get("media_type");
697 let lease_id = crate::attachments::lifecycle::acquire_lease(
698 &mut conn,
699 &sha256,
700 "read",
701 &crate::attachments::lifecycle::SystemClock,
702 )
703 .await?;
704 Ok(AttachmentReadLease {
705 sha256,
706 media_type,
707 lease_id,
708 })
709 }
710
711 pub async fn release_attachment_lease(&self, lease_id: &str) -> Result<()> {
712 let mut conn = self.acquire_writer().await?;
713 crate::attachments::lifecycle::release_lease(&mut conn, lease_id).await
714 }
715
716 pub async fn attachment_lifecycle_report(
717 &self,
718 blob_dir: &Path,
719 policy: crate::attachments::lifecycle::LifecyclePolicy,
720 ) -> Result<crate::attachments::lifecycle::LifecycleReport> {
721 let mut conn = self.acquire_writer().await?;
722 crate::attachments::lifecycle::lifecycle_report(
723 &mut conn,
724 blob_dir,
725 policy,
726 &crate::attachments::lifecycle::SystemClock,
727 )
728 .await
729 }
730}
731
732#[cfg(test)]
733mod tests {
734 use std::time::Duration;
735
736 use super::*;
737
738 #[tokio::test]
739 async fn lifecycle_report_waits_for_guarded_writer() {
740 let temp = tempfile::tempdir().unwrap();
741 let database = Database::open(&temp.path().join("lifecycle.sqlite"))
742 .await
743 .unwrap();
744 let writer = database.acquire_writer().await.unwrap();
745
746 let reporting_database = database.clone();
747 let blob_dir = temp.path().join("blobs");
748 let mut report = tokio::spawn(async move {
749 reporting_database
750 .attachment_lifecycle_report(
751 &blob_dir,
752 crate::attachments::lifecycle::LifecyclePolicy::default(),
753 )
754 .await
755 });
756
757 assert!(
758 tokio::time::timeout(Duration::from_millis(100), &mut report)
759 .await
760 .is_err(),
761 "lifecycle reporting should wait for the guarded writer"
762 );
763
764 drop(writer);
765 tokio::time::timeout(Duration::from_secs(1), report)
766 .await
767 .expect("lifecycle reporting should proceed after the writer is released")
768 .unwrap()
769 .unwrap();
770 }
771}