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