aven_core/operations/
attachments.rs

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            crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
415                &mut tx,
416                std::slice::from_ref(&stored.sha256),
417                &crate::attachments::lifecycle::SystemClock,
418            )
419            .await?;
420            tx.commit().await?;
421            return Ok::<_, anyhow::Error>((attachment_id, false));
422        }
423
424        let attachment_id = identity.attachment_id.unwrap_or_else(new_id);
425        let created_at = identity.created_at.unwrap_or_else(now);
426        record_attachment_add(
427            &mut tx,
428            workspace,
429            task_id,
430            AttachmentRecordDraft {
431                attachment_id: &attachment_id,
432                sha256: &stored.sha256,
433                byte_size: stored.byte_size,
434                media_type: &stored.facts.media_type,
435                filename: input.filename.as_deref(),
436                alt_text: input.alt_text.as_deref(),
437                width: stored.facts.width,
438                height: stored.facts.height,
439                created_at: &created_at,
440            },
441        )
442        .await?;
443        crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
444            &mut tx,
445            std::slice::from_ref(&stored.sha256),
446            &crate::attachments::lifecycle::SystemClock,
447        )
448        .await?;
449        tx.commit().await?;
450        Ok((attachment_id, true))
451    }
452    .await;
453    let release_result = crate::attachments::lifecycle::release_lease(conn, &staging_lease).await;
454    let (attachment_id, created) = match database_result {
455        Ok(result) => result,
456        Err(error) => {
457            crate::attachments::storage::remove_staged_blob_if_unreferenced(
458                conn,
459                blob_dir,
460                &stored.sha256,
461            )
462            .await;
463            return Err(error);
464        }
465    };
466    release_result?;
467    crate::attachments::lifecycle::reconcile_liveness(
468        conn,
469        &crate::attachments::lifecycle::SystemClock,
470    )
471    .await?;
472    let outcome = attachment_by_id(conn, workspace, &attachment_id).await?;
473    Ok(AttachmentAddOutcome {
474        outcome,
475        created,
476        optimized: optimized_flag,
477    })
478}
479
480pub async fn attachment_by_id(
481    conn: &mut SqliteConnection,
482    workspace: &Workspace,
483    attachment_id: &str,
484) -> Result<AttachmentOutcome> {
485    let attachment = require_attachment(conn, &workspace.id, attachment_id).await?;
486    let has_blob = attachment_has_blob(conn, &attachment.sha256).await?;
487    Ok(AttachmentOutcome {
488        attachment,
489        has_blob,
490    })
491}
492
493pub async fn delete_task_attachment(
494    conn: &mut SqliteConnection,
495    workspace: &Workspace,
496    attachment_id: &str,
497) -> Result<AttachmentOutcome> {
498    let attachment = require_attachment(conn, &workspace.id, attachment_id).await?;
499    if attachment.deleted {
500        let has_blob = attachment_has_blob(conn, &attachment.sha256).await?;
501        return Ok(AttachmentOutcome {
502            attachment,
503            has_blob,
504        });
505    }
506
507    let task_id = attachment.task_id.clone();
508    let deleted_at = now();
509
510    let mut tx = begin_immediate(conn).await?;
511    crate::operations::route_recurrence_task_field(&mut tx, workspace, &task_id, "attachments", "")
512        .await?;
513
514    let change_id = append_change(
515        &mut tx,
516        ChangeEntity::Task,
517        &task_id,
518        Some("attachments"),
519        op_type::ATTACHMENT_DELETE,
520        ChangePayload::workspace(workspace)
521            .set("attachment_id", attachment_id)
522            .set("filename", attachment.filename.as_deref())
523            .set("media_type", &attachment.media_type)
524            .set("deleted_at", &deleted_at),
525    )
526    .await?;
527
528    sqlx::query(
529        "UPDATE task_attachments SET deleted = 1, deleted_at = ?, deleted_by_change_id = ? WHERE workspace_id = ? AND attachment_id = ?",
530    )
531    .bind(&deleted_at)
532    .bind(&change_id)
533    .bind(&workspace.id)
534    .bind(attachment_id)
535    .execute(&mut *tx)
536    .await?;
537
538    crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
539        &mut tx,
540        std::slice::from_ref(&attachment.sha256),
541        &crate::attachments::lifecycle::SystemClock,
542    )
543    .await?;
544    tx.commit().await?;
545
546    attachment_by_id(conn, workspace, attachment_id).await
547}
548
549pub async fn attachment_read_items_by_task(
550    conn: &mut SqliteConnection,
551    workspace_id: &str,
552    task_id: &str,
553    include_deleted: bool,
554) -> Result<Vec<AttachmentReadItem>> {
555    let attachments = attachments_by_task(conn, workspace_id, task_id, include_deleted).await?;
556    let mut items = Vec::with_capacity(attachments.len());
557    for attachment in attachments {
558        let bytes_state = attachment_bytes_state(conn, &attachment.sha256).await?;
559        items.push(AttachmentReadItem {
560            attachment,
561            has_blob: bytes_state == AttachmentBytesState::Present,
562            bytes_state,
563        });
564    }
565    Ok(items)
566}
567
568pub async fn attachments_by_task(
569    conn: &mut SqliteConnection,
570    workspace_id: &str,
571    task_id: &str,
572    include_deleted: bool,
573) -> Result<Vec<TaskAttachment>> {
574    let deleted_predicate = if include_deleted {
575        ""
576    } else {
577        " AND ta.deleted = 0"
578    };
579    let rows = sqlx::query(AssertSqlSafe(format!(
580        "SELECT {ATTACHMENT_COLUMNS}
581         FROM task_attachments ta
582         WHERE ta.workspace_id = ? AND ta.task_id = ?{deleted_predicate}
583         ORDER BY ta.created_at, ta.attachment_id"
584    )))
585    .bind(workspace_id)
586    .bind(task_id)
587    .fetch_all(&mut *conn)
588    .await?;
589
590    let mut attachments = Vec::with_capacity(rows.len());
591    for row in &rows {
592        attachments.push(attachment_from_row(row));
593    }
594    Ok(attachments)
595}
596
597impl Database {
598    pub async fn add_task_attachment(
599        &self,
600        workspace: &Workspace,
601        blob_dir: &Path,
602        policy: crate::attachments::lifecycle::LifecyclePolicy,
603        task_id: &TaskId,
604        input: AttachmentAddInput,
605    ) -> Result<AttachmentAddOutcome> {
606        let mut conn = self.acquire_writer().await?;
607        add_task_attachment(&mut conn, workspace, blob_dir, policy, task_id, input).await
608    }
609
610    pub async fn add_ordered_task_attachment(
611        &self,
612        workspace: &Workspace,
613        blob_dir: &Path,
614        policy: crate::attachments::lifecycle::LifecyclePolicy,
615        task_id: &TaskId,
616        created_at: String,
617        input: TaskAttachmentAddInput,
618    ) -> Result<AttachmentAddOutcome> {
619        let mut conn = self.acquire_writer().await?;
620        add_ordered_task_attachment(
621            &mut conn, workspace, blob_dir, policy, task_id, created_at, input,
622        )
623        .await
624    }
625
626    pub async fn attachment_by_id(
627        &self,
628        workspace: &Workspace,
629        attachment_id: &str,
630    ) -> Result<AttachmentOutcome> {
631        let mut conn = self.acquire_reader().await?;
632        attachment_by_id(&mut conn, workspace, attachment_id).await
633    }
634
635    pub async fn delete_task_attachment(
636        &self,
637        workspace: &Workspace,
638        attachment_id: &str,
639    ) -> Result<AttachmentOutcome> {
640        let mut conn = self.acquire_writer().await?;
641        delete_task_attachment(&mut conn, workspace, attachment_id).await
642    }
643
644    pub async fn attachment_read_items_by_task(
645        &self,
646        workspace_id: &WorkspaceId,
647        task_id: &TaskId,
648        include_deleted: bool,
649    ) -> Result<Vec<AttachmentReadItem>> {
650        let mut conn = self.acquire_reader().await?;
651        attachment_read_items_by_task(
652            &mut conn,
653            workspace_id.as_str(),
654            task_id.as_str(),
655            include_deleted,
656        )
657        .await
658    }
659
660    pub async fn prune_attachments(
661        &self,
662        blob_dir: &Path,
663        policy: crate::attachments::lifecycle::LifecyclePolicy,
664        apply: bool,
665    ) -> Result<crate::attachments::lifecycle::PruneSummary> {
666        let mut conn = self.acquire_writer().await?;
667        crate::attachments::lifecycle::prune(
668            &mut conn,
669            blob_dir,
670            policy,
671            apply,
672            &crate::attachments::lifecycle::SystemClock,
673        )
674        .await
675    }
676
677    pub async fn acquire_attachment_lease(&self, sha256: &str, purpose: &str) -> Result<String> {
678        let mut conn = self.acquire_writer().await?;
679        crate::attachments::lifecycle::acquire_lease(
680            &mut conn,
681            sha256,
682            purpose,
683            &crate::attachments::lifecycle::SystemClock,
684        )
685        .await
686    }
687
688    pub async fn acquire_live_attachment_read_lease(
689        &self,
690        workspace: &Workspace,
691        attachment_id: &str,
692    ) -> Result<AttachmentReadLease> {
693        let mut conn = self.acquire_writer().await?;
694        let row = sqlx::query(
695            "SELECT ta.sha256, ta.media_type, bi.available
696             FROM task_attachments ta
697             JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
698             LEFT JOIN blob_inventory bi ON bi.sha256 = ta.sha256
699             WHERE ta.workspace_id = ? AND ta.attachment_id = ?
700               AND ta.deleted = 0 AND t.deleted = 0",
701        )
702        .bind(&workspace.id)
703        .bind(attachment_id)
704        .fetch_optional(&mut *conn)
705        .await?
706        .ok_or_else(|| anyhow::anyhow!("error attachment-invalidated"))?;
707        if !row.try_get::<bool, _>("available").unwrap_or(false) {
708            bail!("error attachment-blob-unavailable");
709        }
710        let sha256: String = row.get("sha256");
711        let media_type: String = row.get("media_type");
712        let lease_id = crate::attachments::lifecycle::acquire_lease(
713            &mut conn,
714            &sha256,
715            "read",
716            &crate::attachments::lifecycle::SystemClock,
717        )
718        .await?;
719        Ok(AttachmentReadLease {
720            sha256,
721            media_type,
722            lease_id,
723        })
724    }
725
726    pub async fn release_attachment_lease(&self, lease_id: &str) -> Result<()> {
727        let mut conn = self.acquire_writer().await?;
728        crate::attachments::lifecycle::release_lease(&mut conn, lease_id).await
729    }
730
731    pub async fn attachment_lifecycle_report(
732        &self,
733        blob_dir: &Path,
734        policy: crate::attachments::lifecycle::LifecyclePolicy,
735    ) -> Result<crate::attachments::lifecycle::LifecycleReport> {
736        let mut conn = self.acquire_writer().await?;
737        crate::attachments::lifecycle::lifecycle_report(
738            &mut conn,
739            blob_dir,
740            policy,
741            &crate::attachments::lifecycle::SystemClock,
742        )
743        .await
744    }
745}
746
747#[cfg(test)]
748mod tests {
749    use std::time::Duration;
750
751    use super::*;
752
753    #[tokio::test]
754    async fn lifecycle_report_waits_for_guarded_writer() {
755        let temp = tempfile::tempdir().unwrap();
756        let database = Database::open(&temp.path().join("lifecycle.sqlite"))
757            .await
758            .unwrap();
759        let writer = database.acquire_writer().await.unwrap();
760
761        let reporting_database = database.clone();
762        let blob_dir = temp.path().join("blobs");
763        let mut report = tokio::spawn(async move {
764            reporting_database
765                .attachment_lifecycle_report(
766                    &blob_dir,
767                    crate::attachments::lifecycle::LifecyclePolicy::default(),
768                )
769                .await
770        });
771
772        assert!(
773            tokio::time::timeout(Duration::from_millis(100), &mut report)
774                .await
775                .is_err(),
776            "lifecycle reporting should wait for the guarded writer"
777        );
778
779        drop(writer);
780        tokio::time::timeout(Duration::from_secs(1), report)
781            .await
782            .expect("lifecycle reporting should proceed after the writer is released")
783            .unwrap()
784            .unwrap();
785    }
786}
aven_core/operations/attachments.rs

aven_core/operations/
attachments.rs
