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 let task_exists = sqlx::query_scalar::<_, bool>(
389 "SELECT EXISTS(SELECT 1 FROM tasks WHERE workspace_id = ? AND id = ?)",
390 )
391 .bind(&workspace.id)
392 .bind(task_id)
393 .fetch_one(&mut *tx)
394 .await?;
395 if !task_exists {
396 bail!("error task-not-found task_id={task_id}");
397 }
398
399 if input.dedupe_existing
400 && let Some(existing) =
401 existing_live_attachments_by_sha(&mut tx, &workspace.id, task_id, &stored.sha256)
402 .await?
403 .first()
404 {
405 let attachment_id = existing.attachment_id.clone();
406 tx.commit().await?;
407 return Ok::<_, anyhow::Error>((attachment_id, false));
408 }
409
410 let attachment_id = identity.attachment_id.unwrap_or_else(new_id);
411 let created_at = identity.created_at.unwrap_or_else(now);
412 record_attachment_add(
413 &mut tx,
414 workspace,
415 task_id,
416 AttachmentRecordDraft {
417 attachment_id: &attachment_id,
418 sha256: &stored.sha256,
419 byte_size: stored.byte_size,
420 media_type: &stored.facts.media_type,
421 filename: input.filename.as_deref(),
422 alt_text: input.alt_text.as_deref(),
423 width: stored.facts.width,
424 height: stored.facts.height,
425 created_at: &created_at,
426 },
427 )
428 .await?;
429 tx.commit().await?;
430 Ok((attachment_id, true))
431 }
432 .await;
433 let release_result = crate::attachments::lifecycle::release_lease(conn, &staging_lease).await;
434 let (attachment_id, created) = match database_result {
435 Ok(result) => result,
436 Err(error) => {
437 crate::attachments::storage::remove_staged_blob_if_unreferenced(
438 conn,
439 blob_dir,
440 &stored.sha256,
441 )
442 .await;
443 return Err(error);
444 }
445 };
446 release_result?;
447 crate::attachments::lifecycle::reconcile_liveness(
448 conn,
449 &crate::attachments::lifecycle::SystemClock,
450 )
451 .await?;
452 let outcome = attachment_by_id(conn, workspace, &attachment_id).await?;
453 Ok(AttachmentAddOutcome {
454 outcome,
455 created,
456 optimized: optimized_flag,
457 })
458}
459
460pub async fn attachment_by_id(
461 conn: &mut SqliteConnection,
462 workspace: &Workspace,
463 attachment_id: &str,
464) -> Result<AttachmentOutcome> {
465 let attachment = require_attachment(conn, &workspace.id, attachment_id).await?;
466 let has_blob = attachment_has_blob(conn, &attachment.sha256).await?;
467 Ok(AttachmentOutcome {
468 attachment,
469 has_blob,
470 })
471}
472
473pub async fn delete_task_attachment(
474 conn: &mut SqliteConnection,
475 workspace: &Workspace,
476 attachment_id: &str,
477) -> Result<AttachmentOutcome> {
478 let attachment = require_attachment(conn, &workspace.id, attachment_id).await?;
479 if attachment.deleted {
480 let has_blob = attachment_has_blob(conn, &attachment.sha256).await?;
481 return Ok(AttachmentOutcome {
482 attachment,
483 has_blob,
484 });
485 }
486
487 let task_id = attachment.task_id.clone();
488 let deleted_at = now();
489
490 let mut tx = begin_immediate(conn).await?;
491
492 let change_id = append_change(
493 &mut tx,
494 ChangeEntity::Task,
495 &task_id,
496 Some("attachments"),
497 op_type::ATTACHMENT_DELETE,
498 ChangePayload::workspace(workspace)
499 .set("attachment_id", attachment_id)
500 .set("deleted_at", &deleted_at),
501 )
502 .await?;
503
504 sqlx::query(
505 "UPDATE task_attachments SET deleted = 1, deleted_at = ?, deleted_by_change_id = ? WHERE workspace_id = ? AND attachment_id = ?",
506 )
507 .bind(&deleted_at)
508 .bind(&change_id)
509 .bind(&workspace.id)
510 .bind(attachment_id)
511 .execute(&mut *tx)
512 .await?;
513
514 tx.commit().await?;
515 crate::attachments::lifecycle::reconcile_liveness(
516 conn,
517 &crate::attachments::lifecycle::SystemClock,
518 )
519 .await?;
520
521 attachment_by_id(conn, workspace, attachment_id).await
522}
523
524pub async fn attachment_read_items_by_task(
525 conn: &mut SqliteConnection,
526 workspace_id: &str,
527 task_id: &str,
528 include_deleted: bool,
529) -> Result<Vec<AttachmentReadItem>> {
530 let attachments = attachments_by_task(conn, workspace_id, task_id, include_deleted).await?;
531 let mut items = Vec::with_capacity(attachments.len());
532 for attachment in attachments {
533 let bytes_state = attachment_bytes_state(conn, &attachment.sha256).await?;
534 items.push(AttachmentReadItem {
535 attachment,
536 has_blob: bytes_state == AttachmentBytesState::Present,
537 bytes_state,
538 });
539 }
540 Ok(items)
541}
542
543pub async fn attachments_by_task(
544 conn: &mut SqliteConnection,
545 workspace_id: &str,
546 task_id: &str,
547 include_deleted: bool,
548) -> Result<Vec<TaskAttachment>> {
549 let deleted_predicate = if include_deleted {
550 ""
551 } else {
552 " AND ta.deleted = 0"
553 };
554 let rows = sqlx::query(AssertSqlSafe(format!(
555 "SELECT {ATTACHMENT_COLUMNS}
556 FROM task_attachments ta
557 WHERE ta.workspace_id = ? AND ta.task_id = ?{deleted_predicate}
558 ORDER BY ta.created_at, ta.attachment_id"
559 )))
560 .bind(workspace_id)
561 .bind(task_id)
562 .fetch_all(&mut *conn)
563 .await?;
564
565 let mut attachments = Vec::with_capacity(rows.len());
566 for row in &rows {
567 attachments.push(attachment_from_row(row));
568 }
569 Ok(attachments)
570}
571
572impl Database {
573 pub async fn add_task_attachment(
574 &self,
575 workspace: &Workspace,
576 blob_dir: &Path,
577 policy: crate::attachments::lifecycle::LifecyclePolicy,
578 task_id: &TaskId,
579 input: AttachmentAddInput,
580 ) -> Result<AttachmentAddOutcome> {
581 let mut conn = self.acquire().await?;
582 add_task_attachment(&mut conn, workspace, blob_dir, policy, task_id, input).await
583 }
584
585 pub async fn add_ordered_task_attachment(
586 &self,
587 workspace: &Workspace,
588 blob_dir: &Path,
589 policy: crate::attachments::lifecycle::LifecyclePolicy,
590 task_id: &TaskId,
591 created_at: String,
592 input: TaskAttachmentAddInput,
593 ) -> Result<AttachmentAddOutcome> {
594 let mut conn = self.acquire().await?;
595 add_ordered_task_attachment(
596 &mut conn, workspace, blob_dir, policy, task_id, created_at, input,
597 )
598 .await
599 }
600
601 pub async fn attachment_by_id(
602 &self,
603 workspace: &Workspace,
604 attachment_id: &str,
605 ) -> Result<AttachmentOutcome> {
606 let mut conn = self.acquire().await?;
607 attachment_by_id(&mut conn, workspace, attachment_id).await
608 }
609
610 pub async fn delete_task_attachment(
611 &self,
612 workspace: &Workspace,
613 attachment_id: &str,
614 ) -> Result<AttachmentOutcome> {
615 let mut conn = self.acquire().await?;
616 delete_task_attachment(&mut conn, workspace, attachment_id).await
617 }
618
619 pub async fn attachment_read_items_by_task(
620 &self,
621 workspace_id: &WorkspaceId,
622 task_id: &TaskId,
623 include_deleted: bool,
624 ) -> Result<Vec<AttachmentReadItem>> {
625 let mut conn = self.acquire().await?;
626 attachment_read_items_by_task(
627 &mut conn,
628 workspace_id.as_str(),
629 task_id.as_str(),
630 include_deleted,
631 )
632 .await
633 }
634
635 pub async fn prune_attachments(
636 &self,
637 blob_dir: &Path,
638 policy: crate::attachments::lifecycle::LifecyclePolicy,
639 apply: bool,
640 ) -> Result<crate::attachments::lifecycle::PruneSummary> {
641 let mut conn = self.acquire().await?;
642 crate::attachments::lifecycle::prune(
643 &mut conn,
644 blob_dir,
645 policy,
646 apply,
647 &crate::attachments::lifecycle::SystemClock,
648 )
649 .await
650 }
651
652 pub async fn acquire_attachment_lease(&self, sha256: &str, purpose: &str) -> Result<String> {
653 let mut conn = self.acquire().await?;
654 crate::attachments::lifecycle::acquire_lease(
655 &mut conn,
656 sha256,
657 purpose,
658 &crate::attachments::lifecycle::SystemClock,
659 )
660 .await
661 }
662
663 pub async fn acquire_live_attachment_read_lease(
664 &self,
665 workspace: &Workspace,
666 attachment_id: &str,
667 ) -> Result<AttachmentReadLease> {
668 let mut conn = self.acquire().await?;
669 let row = sqlx::query(
670 "SELECT ta.sha256, ta.media_type, bi.available
671 FROM task_attachments ta
672 JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
673 LEFT JOIN blob_inventory bi ON bi.sha256 = ta.sha256
674 WHERE ta.workspace_id = ? AND ta.attachment_id = ?
675 AND ta.deleted = 0 AND t.deleted = 0",
676 )
677 .bind(&workspace.id)
678 .bind(attachment_id)
679 .fetch_optional(&mut *conn)
680 .await?
681 .ok_or_else(|| anyhow::anyhow!("error attachment-invalidated"))?;
682 if !row.try_get::<bool, _>("available").unwrap_or(false) {
683 bail!("error attachment-blob-unavailable");
684 }
685 let sha256: String = row.get("sha256");
686 let media_type: String = row.get("media_type");
687 let lease_id = crate::attachments::lifecycle::acquire_lease(
688 &mut conn,
689 &sha256,
690 "read",
691 &crate::attachments::lifecycle::SystemClock,
692 )
693 .await?;
694 Ok(AttachmentReadLease {
695 sha256,
696 media_type,
697 lease_id,
698 })
699 }
700
701 pub async fn release_attachment_lease(&self, lease_id: &str) -> Result<()> {
702 let mut conn = self.acquire().await?;
703 crate::attachments::lifecycle::release_lease(&mut conn, lease_id).await
704 }
705
706 pub async fn attachment_lifecycle_report(
707 &self,
708 blob_dir: &Path,
709 policy: crate::attachments::lifecycle::LifecyclePolicy,
710 ) -> Result<crate::attachments::lifecycle::LifecycleReport> {
711 let mut conn = self.acquire().await?;
712 crate::attachments::lifecycle::lifecycle_report(
713 &mut conn,
714 blob_dir,
715 policy,
716 &crate::attachments::lifecycle::SystemClock,
717 )
718 .await
719 }
720}