aven_core/operations/
attachments.rs

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

aven_core/operations/
attachments.rs
