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

aven_core/operations/
attachments.rs
