1use std::collections::{BTreeMap, BTreeSet};
2use std::path::Path;
3
4use crate::ids::WorkspaceId;
5use anyhow::{Result, bail};
6use sqlx::SqliteConnection;
7use tracing::{info, warn};
8
9use crate::change_log::{ChangeEntity, ChangePayload, append_change, op_type};
10use crate::choices::{TaskPriority, TaskSource, TaskStatus};
11use crate::db::{Database, begin_immediate, set_field_version};
12use crate::ids::{TaskId, new_id, now};
13use crate::labels::resolve_labels_in_workspace;
14use crate::mutation::{set_task_field, set_task_project};
15use crate::projects::resolve_or_create_project_in_workspace;
16use crate::refs::{DisplayRefContext, get_task_in_workspace};
17use crate::task_fields::TaskField;
18use crate::types::Task;
19use crate::undo::{
20 TaskUndoSnapshot, UndoCommand, UndoContext, UndoPayload, record_tui_undo, task_snapshot,
21};
22use crate::workspaces::Workspace;
23
24pub struct TaskDraft {
25 pub title: String,
26 pub description: String,
27 pub project: Option<String>,
28 pub status: String,
29 pub priority: String,
30 pub source: TaskSource,
31 pub labels: Vec<String>,
32 pub available_at: Option<String>,
33 pub due_on: Option<String>,
34 pub is_epic: bool,
35}
36
37#[derive(Debug, Clone, Default)]
38pub enum TaskCreationUndo {
39 #[default]
40 None,
41 TuiTask,
42 TuiEpicChild {
43 epic_id: TaskId,
44 epic_display_ref: String,
45 },
46}
47
48#[derive(Debug, Clone)]
49pub struct TaskCreationOptions {
50 epic_id: Option<TaskId>,
51 undo: TaskCreationUndo,
52}
53
54impl TaskCreationOptions {
55 pub fn standalone(undo: TaskCreationUndo) -> Self {
56 Self {
57 epic_id: None,
58 undo,
59 }
60 }
61
62 pub fn for_epic(epic_id: TaskId, undo: TaskCreationUndo) -> Self {
63 Self {
64 epic_id: Some(epic_id),
65 undo,
66 }
67 }
68}
69
70#[derive(Debug)]
71pub struct TaskOutcome {
72 pub task: Task,
73 pub create_change_id: Option<String>,
74 pub attachment_change_ids: Vec<String>,
75}
76
77struct InsertedTask {
78 id: TaskId,
79 change_id: String,
80 project_key: String,
81 label_count: usize,
82}
83
84#[derive(Debug, Clone)]
85pub struct TaskLabelSelection {
86 pub selected: Vec<String>,
87 pub partial: Vec<String>,
88}
89
90#[derive(Clone, Default)]
91pub struct TaskUpdate {
92 pub title: Option<String>,
93 pub description: Option<String>,
94 pub project: Option<String>,
95 pub status: Option<String>,
96 pub priority: Option<String>,
97 pub cycle_priority: Option<bool>,
98 pub available_at: Option<Option<String>>,
99 pub due_on: Option<Option<String>>,
100 pub deleted: Option<bool>,
101 pub is_epic: Option<bool>,
102 pub add_labels: Vec<String>,
103 pub remove_labels: Vec<String>,
104 pub label_selection: Option<TaskLabelSelection>,
105}
106
107pub struct TaskUpdateOutcome {
108 pub task: Task,
109 pub changed: bool,
110}
111
112#[derive(Debug)]
113pub struct TaskMutationOutcome {
114 pub task: Task,
115 pub before: TaskUndoSnapshot,
116 pub after: TaskUndoSnapshot,
117 pub changed: bool,
118}
119
120#[derive(Debug)]
121pub struct TaskMutationReport {
122 pub outcomes: Vec<TaskMutationOutcome>,
123}
124
125#[derive(Clone, Copy, PartialEq, Eq)]
126enum TaskMutationDetail {
127 Report,
128 Compact,
129}
130
131struct TaskMutationExecutionOutcome {
132 task: Task,
133 before: Option<TaskUndoSnapshot>,
134 after: Option<TaskUndoSnapshot>,
135 changed: bool,
136}
137
138impl TaskMutationReport {
139 pub fn changed_count(&self) -> usize {
140 self.outcomes
141 .iter()
142 .filter(|outcome| outcome.changed)
143 .count()
144 }
145}
146
147pub struct NoteDeleteOutcome {
148 #[allow(dead_code)]
149 pub task_id: TaskId,
150 #[allow(dead_code)]
151 pub note_id: String,
152 pub changed: bool,
153}
154
155pub struct NoteOutcome {
156 #[allow(dead_code)]
157 pub task_id: TaskId,
158 pub note_id: String,
159 pub change_id: String,
160}
161impl Database {
162 pub async fn create_task(
163 &self,
164 workspace: &Workspace,
165 draft: TaskDraft,
166 ) -> Result<TaskOutcome> {
167 self.create_task_with_undo(workspace, draft, TaskCreationUndo::None)
168 .await
169 }
170
171 pub async fn create_task_with_undo(
172 &self,
173 workspace: &Workspace,
174 draft: TaskDraft,
175 undo: TaskCreationUndo,
176 ) -> Result<TaskOutcome> {
177 let mut conn = self.acquire().await?;
178 create_task_with_epic(&mut conn, workspace, draft, None, undo).await
179 }
180
181 pub async fn create_task_for_epic(
182 &self,
183 workspace: &Workspace,
184 draft: TaskDraft,
185 epic_id: &TaskId,
186 ) -> Result<TaskOutcome> {
187 self.create_task_for_epic_with_undo(workspace, draft, epic_id, TaskCreationUndo::None)
188 .await
189 }
190
191 pub async fn create_task_for_epic_with_undo(
192 &self,
193 workspace: &Workspace,
194 draft: TaskDraft,
195 epic_id: &TaskId,
196 undo: TaskCreationUndo,
197 ) -> Result<TaskOutcome> {
198 let mut conn = self.acquire().await?;
199 create_task_with_epic(&mut conn, workspace, draft, Some(epic_id), undo).await
200 }
201
202 pub async fn create_task_with_attachments(
203 &self,
204 workspace: &Workspace,
205 blob_dir: &Path,
206 lifecycle_policy: crate::attachments::lifecycle::LifecyclePolicy,
207 draft: TaskDraft,
208 attachments: Vec<super::attachments::TaskAttachmentAddInput>,
209 ) -> Result<TaskOutcome> {
210 self.create_task_with_attachments_and_undo(
211 workspace,
212 blob_dir,
213 lifecycle_policy,
214 draft,
215 attachments,
216 TaskCreationUndo::None,
217 )
218 .await
219 }
220
221 pub async fn create_task_with_attachments_and_undo(
222 &self,
223 workspace: &Workspace,
224 blob_dir: &Path,
225 lifecycle_policy: crate::attachments::lifecycle::LifecyclePolicy,
226 draft: TaskDraft,
227 attachments: Vec<super::attachments::TaskAttachmentAddInput>,
228 undo: TaskCreationUndo,
229 ) -> Result<TaskOutcome> {
230 let mut conn = self.acquire().await?;
231 create_task_with_attachments_and_epic(
232 &mut conn,
233 workspace,
234 blob_dir,
235 lifecycle_policy,
236 draft,
237 attachments,
238 TaskCreationOptions::standalone(undo),
239 )
240 .await
241 }
242
243 pub async fn create_task_with_attachments_for_epic(
244 &self,
245 workspace: &Workspace,
246 blob_dir: &Path,
247 lifecycle_policy: crate::attachments::lifecycle::LifecyclePolicy,
248 draft: TaskDraft,
249 attachments: Vec<super::attachments::TaskAttachmentAddInput>,
250 epic_id: &TaskId,
251 ) -> Result<TaskOutcome> {
252 self.create_task_with_attachments_for_epic_and_undo(
253 workspace,
254 blob_dir,
255 lifecycle_policy,
256 draft,
257 attachments,
258 TaskCreationOptions::for_epic(epic_id.clone(), TaskCreationUndo::None),
259 )
260 .await
261 }
262
263 pub async fn create_task_with_attachments_for_epic_and_undo(
264 &self,
265 workspace: &Workspace,
266 blob_dir: &Path,
267 lifecycle_policy: crate::attachments::lifecycle::LifecyclePolicy,
268 draft: TaskDraft,
269 attachments: Vec<super::attachments::TaskAttachmentAddInput>,
270 options: TaskCreationOptions,
271 ) -> Result<TaskOutcome> {
272 let mut conn = self.acquire().await?;
273 create_task_with_attachments_and_epic(
274 &mut conn,
275 workspace,
276 blob_dir,
277 lifecycle_policy,
278 draft,
279 attachments,
280 options,
281 )
282 .await
283 }
284
285 pub async fn update_task(
286 &self,
287 workspace: &Workspace,
288 task_id: &TaskId,
289 update: TaskUpdate,
290 ) -> Result<TaskUpdateOutcome> {
291 let outcomes = self
292 .mutate_tasks_owned(
293 workspace,
294 vec![(task_id.clone(), update)],
295 UndoContext::None,
296 TaskMutationDetail::Compact,
297 )
298 .await?;
299 let outcome = outcomes.into_iter().next().unwrap();
300 Ok(TaskUpdateOutcome {
301 task: outcome.task,
302 changed: outcome.changed,
303 })
304 }
305
306 pub async fn update_tasks(
307 &self,
308 workspace: &Workspace,
309 updates: Vec<(TaskId, TaskUpdate)>,
310 ) -> Result<Vec<TaskUpdateOutcome>> {
311 Ok(self
312 .mutate_tasks_owned(
313 workspace,
314 updates,
315 UndoContext::None,
316 TaskMutationDetail::Compact,
317 )
318 .await?
319 .into_iter()
320 .map(|outcome| TaskUpdateOutcome {
321 task: outcome.task,
322 changed: outcome.changed,
323 })
324 .collect())
325 }
326
327 pub async fn mutate_tasks(
328 &self,
329 workspace: &Workspace,
330 updates: Vec<(TaskId, TaskUpdate)>,
331 undo: UndoContext,
332 ) -> Result<TaskMutationReport> {
333 let outcomes = self
334 .mutate_tasks_owned(workspace, updates, undo, TaskMutationDetail::Report)
335 .await?;
336 Ok(TaskMutationReport {
337 outcomes: outcomes
338 .into_iter()
339 .map(|outcome| TaskMutationOutcome {
340 task: outcome.task,
341 before: outcome.before.expect("report mutation before snapshot"),
342 after: outcome.after.expect("report mutation after snapshot"),
343 changed: outcome.changed,
344 })
345 .collect(),
346 })
347 }
348
349 async fn mutate_tasks_owned(
350 &self,
351 workspace: &Workspace,
352 updates: Vec<(TaskId, TaskUpdate)>,
353 undo: UndoContext,
354 detail: TaskMutationDetail,
355 ) -> Result<Vec<TaskMutationExecutionOutcome>> {
356 for (_, update) in &updates {
357 validate_task_update(update)?;
358 }
359 if updates.is_empty() {
360 return Ok(Vec::new());
361 }
362 if detail == TaskMutationDetail::Compact && !matches!(&undo, UndoContext::None) {
363 bail!("compact task mutations cannot record undo");
364 }
365
366 let mut conn = self.acquire().await?;
367 let mut tx = begin_immediate(&mut conn).await?;
368 let mut outcomes = Vec::with_capacity(updates.len());
369 let mut undo_commands = Vec::new();
370 for (task_id, update) in &updates {
371 let before =
372 if detail == TaskMutationDetail::Report || task_update_requires_snapshot(update) {
373 Some(task_snapshot(&mut tx, &workspace.id, task_id).await?)
374 } else {
375 None
376 };
377 let update = match before.as_ref() {
378 Some(before) => materialize_task_update(update, before)?,
379 None => update.clone(),
380 };
381 let changed = apply_task_update(&mut tx, workspace, task_id, &update).await?;
382 let task = get_task_in_workspace(&mut tx, workspace, task_id).await?;
383 let after = if detail == TaskMutationDetail::Report {
384 Some(task_snapshot(&mut tx, &workspace.id, task_id).await?)
385 } else {
386 None
387 };
388 if let (Some(before), Some(after)) = (&before, &after) {
389 append_task_undo_commands(task_id, before, after, &mut undo_commands);
390 }
391 outcomes.push(TaskMutationExecutionOutcome {
392 task,
393 changed,
394 before,
395 after,
396 });
397 }
398 let changed_count = outcomes.iter().filter(|outcome| outcome.changed).count();
399 if let Some(summary) = undo.task_mutation_summary(changed_count) {
400 record_tui_undo(
401 &mut tx,
402 &workspace.id,
403 &summary,
404 UndoPayload {
405 commands: undo_commands,
406 },
407 )
408 .await?;
409 }
410 tx.commit().await?;
411
412 for outcome in &outcomes {
413 info!(task_id = %outcome.task.id, changed = outcome.changed, "task updated");
414 }
415 Ok(outcomes)
416 }
417
418 pub async fn set_task_deleted(
419 &self,
420 workspace: &Workspace,
421 task_id: &TaskId,
422 deleted: bool,
423 ) -> Result<TaskOutcome> {
424 let outcome = self
425 .update_task(
426 workspace,
427 task_id,
428 TaskUpdate {
429 deleted: Some(deleted),
430 ..TaskUpdate::default()
431 },
432 )
433 .await?;
434 Ok(TaskOutcome {
435 task: outcome.task,
436 create_change_id: None,
437 attachment_change_ids: Vec::new(),
438 })
439 }
440
441 pub async fn add_note(
442 &self,
443 workspace: &Workspace,
444 task_id: &TaskId,
445 body: String,
446 ) -> Result<NoteOutcome> {
447 let mut conn = self.acquire().await?;
448 add_note_operation(&mut conn, workspace, task_id, body, false).await
449 }
450
451 pub async fn add_note_with_tui_undo(
452 &self,
453 workspace: &Workspace,
454 task_id: &TaskId,
455 body: String,
456 ) -> Result<NoteOutcome> {
457 let mut conn = self.acquire().await?;
458 add_note_operation(&mut conn, workspace, task_id, body, true).await
459 }
460
461 pub async fn delete_note(
462 &self,
463 workspace: &Workspace,
464 task_id: &TaskId,
465 note_id: &str,
466 ) -> Result<NoteDeleteOutcome> {
467 let mut conn = self.acquire().await?;
468 delete_note(&mut conn, workspace, task_id, note_id).await
469 }
470}
471
472#[cfg(any(test, feature = "test-support"))]
473pub async fn create_task(
474 conn: &mut SqliteConnection,
475 workspace: &Workspace,
476 draft: TaskDraft,
477) -> Result<TaskOutcome> {
478 create_task_with_epic(conn, workspace, draft, None, TaskCreationUndo::None).await
479}
480
481async fn create_task_with_epic(
482 conn: &mut SqliteConnection,
483 workspace: &Workspace,
484 draft: TaskDraft,
485 epic_id: Option<&TaskId>,
486 undo: TaskCreationUndo,
487) -> Result<TaskOutcome> {
488 validate_task_draft(&draft)?;
489 let mut tx = begin_immediate(conn).await?;
490 let inserted = insert_task(&mut tx, workspace, draft).await?;
491 if let Some(epic_id) = epic_id {
492 super::add_task_to_epic_in_transaction(&mut tx, workspace, &inserted.id, epic_id).await?;
493 }
494 let task = get_task_in_workspace(&mut tx, workspace, &inserted.id).await?;
495 record_task_creation_undo(
496 &mut tx,
497 workspace,
498 &task,
499 Some(inserted.change_id.clone()),
500 Vec::new(),
501 Vec::new(),
502 undo,
503 )
504 .await?;
505 tx.commit().await?;
506 info!(
507 task_id = %inserted.id,
508 project_key = %inserted.project_key,
509 label_count = inserted.label_count,
510 "task created"
511 );
512 Ok(TaskOutcome {
513 task,
514 create_change_id: Some(inserted.change_id),
515 attachment_change_ids: Vec::new(),
516 })
517}
518
519async fn create_task_with_attachments_and_epic(
520 conn: &mut SqliteConnection,
521 workspace: &Workspace,
522 blob_dir: &Path,
523 lifecycle_policy: crate::attachments::lifecycle::LifecyclePolicy,
524 draft: TaskDraft,
525 attachments: Vec<super::attachments::TaskAttachmentAddInput>,
526 options: TaskCreationOptions,
527) -> Result<TaskOutcome> {
528 validate_task_draft(&draft)?;
529 let TaskCreationOptions { epic_id, undo } = options;
530 let mut prepared = Vec::with_capacity(attachments.len());
531 for attachment in attachments {
532 prepared.push(super::attachments::prepare_task_attachment(attachment).await?);
533 }
534
535 let mut unique = BTreeMap::new();
536 for attachment in &prepared {
537 unique
538 .entry(attachment.sha256.clone())
539 .or_insert_with(|| attachment.clone());
540 }
541
542 let mut capacity_reservations = Vec::new();
543 for attachment in unique.values() {
544 let available: bool = sqlx::query_scalar(
545 "SELECT EXISTS(SELECT 1 FROM blob_inventory WHERE sha256 = ? AND available = 1)",
546 )
547 .bind(&attachment.sha256)
548 .fetch_one(&mut *conn)
549 .await?;
550 if available {
551 continue;
552 }
553 match crate::attachments::lifecycle::ensure_local_capacity(
554 conn,
555 blob_dir,
556 &attachment.sha256,
557 attachment.byte_size,
558 lifecycle_policy,
559 &crate::attachments::lifecycle::SystemClock,
560 )
561 .await
562 {
563 Ok(Some(reservation_id)) => capacity_reservations.push(reservation_id),
564 Ok(None) => {}
565 Err(error) => {
566 for reservation_id in capacity_reservations {
567 let _ =
568 crate::attachments::lifecycle::release_reservation(conn, &reservation_id)
569 .await;
570 }
571 return Err(error);
572 }
573 }
574 }
575
576 let mut staging_leases = Vec::with_capacity(unique.len());
577 for attachment in unique.values() {
578 match crate::attachments::lifecycle::acquire_lease(
579 conn,
580 &attachment.sha256,
581 "staging",
582 &crate::attachments::lifecycle::SystemClock,
583 )
584 .await
585 {
586 Ok(lease_id) => staging_leases.push(lease_id),
587 Err(error) => {
588 for lease_id in staging_leases {
589 let _ = crate::attachments::lifecycle::release_lease(conn, &lease_id).await;
590 }
591 for reservation_id in capacity_reservations {
592 let _ =
593 crate::attachments::lifecycle::release_reservation(conn, &reservation_id)
594 .await;
595 }
596 return Err(error);
597 }
598 }
599 }
600
601 let mut created_hashes = Vec::new();
602 for attachment in unique.values() {
603 match crate::attachments::storage::stage_blob(
604 blob_dir,
605 &attachment.sha256,
606 &attachment.bytes,
607 )
608 .await
609 {
610 Ok(staged) if staged.byte_size == attachment.byte_size => {
611 if staged.created {
612 created_hashes.push(staged.sha256);
613 }
614 }
615 Ok(_) => {
616 cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
617 cleanup_created_objects(conn, blob_dir, &created_hashes).await;
618 bail!("error attachment-staged-size-mismatch");
619 }
620 Err(error) => {
621 cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
622 cleanup_created_objects(conn, blob_dir, &created_hashes).await;
623 return Err(error);
624 }
625 }
626 }
627
628 let database_result = async {
629 let mut tx = begin_immediate(conn).await?;
630 for attachment in unique.values() {
631 crate::attachments::storage::upsert_inventory_available(
632 &mut tx,
633 &attachment.sha256,
634 attachment.byte_size,
635 &attachment.facts.media_type,
636 )
637 .await?;
638 }
639 let inserted = insert_task(&mut tx, workspace, draft).await?;
640 if let Some(epic_id) = epic_id.as_ref() {
641 super::add_task_to_epic_in_transaction(&mut tx, workspace, &inserted.id, epic_id)
642 .await?;
643 }
644 let mut attachment_change_ids = Vec::with_capacity(prepared.len());
645 let attachment_base = chrono::DateTime::parse_from_rfc3339(&now())?.to_utc();
646 for (index, attachment) in prepared.iter().enumerate() {
647 let created_at = (attachment_base
648 + chrono::TimeDelta::microseconds(i64::try_from(index)?))
649 .to_rfc3339_opts(chrono::SecondsFormat::Micros, true);
650 attachment_change_ids.push(
651 super::attachments::insert_prepared_attachment(
652 &mut tx,
653 workspace,
654 &inserted.id,
655 attachment,
656 &created_at,
657 )
658 .await?,
659 );
660 }
661 let task = get_task_in_workspace(&mut tx, workspace, &inserted.id).await?;
662 let attachment_ids = prepared
663 .iter()
664 .map(|attachment| attachment.attachment_id.clone())
665 .collect();
666 record_task_creation_undo(
667 &mut tx,
668 workspace,
669 &task,
670 Some(inserted.change_id.clone()),
671 attachment_ids,
672 attachment_change_ids.clone(),
673 undo,
674 )
675 .await?;
676 tx.commit().await?;
677 Ok::<_, anyhow::Error>((inserted, attachment_change_ids, task))
678 }
679 .await;
680
681 let (inserted, attachment_change_ids, task) = match database_result {
682 Ok(value) => value,
683 Err(error) => {
684 cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
685 cleanup_created_objects(conn, blob_dir, &created_hashes).await;
686 return Err(error);
687 }
688 };
689 cleanup_attachment_guards(conn, &staging_leases, &capacity_reservations).await;
690 if let Err(error) = crate::attachments::lifecycle::reconcile_liveness(
691 conn,
692 &crate::attachments::lifecycle::SystemClock,
693 )
694 .await
695 {
696 warn!(%error, "failed to reconcile attachment liveness");
697 }
698 info!(
699 task_id = %inserted.id,
700 project_key = %inserted.project_key,
701 label_count = inserted.label_count,
702 attachment_count = prepared.len(),
703 "task created"
704 );
705 Ok(TaskOutcome {
706 task,
707 create_change_id: Some(inserted.change_id),
708 attachment_change_ids,
709 })
710}
711
712async fn record_task_creation_undo(
713 conn: &mut SqliteConnection,
714 workspace: &Workspace,
715 task: &Task,
716 create_change_id: Option<String>,
717 attachment_ids: Vec<String>,
718 attachment_change_ids: Vec<String>,
719 undo: TaskCreationUndo,
720) -> Result<()> {
721 let (summary, command) = match undo {
722 TaskCreationUndo::None => return Ok(()),
723 TaskCreationUndo::TuiTask => (
724 format!("task {}", task.id),
725 UndoCommand::DeleteCreatedTask {
726 task_id: task.id.clone(),
727 create_change_id,
728 expected: task_snapshot(conn, &workspace.id, &task.id).await?,
729 attachment_ids,
730 attachment_change_ids,
731 },
732 ),
733 TaskCreationUndo::TuiEpicChild {
734 epic_id,
735 epic_display_ref,
736 } => {
737 let display_refs = DisplayRefContext::for_workspace(conn, &workspace.id).await?;
738 let child_ref = display_refs.display_ref(task);
739 (
740 format!("add {child_ref} to {epic_display_ref}"),
741 UndoCommand::AddEpicChild {
742 epic_id,
743 child_id: task.id.clone(),
744 },
745 )
746 }
747 };
748 record_tui_undo(
749 conn,
750 &workspace.id,
751 &summary,
752 UndoPayload {
753 commands: vec![command],
754 },
755 )
756 .await
757}
758
759fn validate_task_draft(draft: &TaskDraft) -> Result<()> {
760 TaskStatus::parse(&draft.status)?;
761 TaskPriority::parse(&draft.priority)?;
762 if let Some(available_at) = draft.available_at.as_deref() {
763 crate::time_validation::validate_available_at_value(available_at)?;
764 }
765 if let Some(due_on) = draft.due_on.as_deref() {
766 crate::time_validation::validate_due_on_value(due_on)?;
767 }
768 Ok(())
769}
770
771async fn insert_task(
772 conn: &mut SqliteConnection,
773 workspace: &Workspace,
774 draft: TaskDraft,
775) -> Result<InsertedTask> {
776 let status = TaskStatus::parse(&draft.status)?;
777 let priority = TaskPriority::parse(&draft.priority)?;
778 let available_at = draft.available_at.as_deref().unwrap_or("");
779 let due_on = draft.due_on.as_deref().unwrap_or("");
780 let id = TaskId::new();
781 let ts = now();
782 let project = draft
783 .project
784 .as_deref()
785 .ok_or_else(|| anyhow::anyhow!("error project-required"))?;
786 let project = resolve_or_create_project_in_workspace(conn, &workspace.id, project).await?;
787 let labels = resolve_labels_in_workspace(conn, &workspace.id, &draft.labels).await?;
788 sqlx::query(
789 "INSERT INTO tasks(workspace_id, id, title, description, project_id, status, priority, source, created_at, updated_at, queue_activity_at, available_at, due_on, is_epic)
790 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
791 )
792 .bind(&workspace.id)
793 .bind(&id)
794 .bind(&draft.title)
795 .bind(&draft.description)
796 .bind(&project.id)
797 .bind(status.as_str())
798 .bind(priority.as_str())
799 .bind(draft.source.as_str())
800 .bind(&ts)
801 .bind(&ts)
802 .bind(&ts)
803 .bind(available_at)
804 .bind(due_on)
805 .bind(i64::from(draft.is_epic))
806 .execute(&mut *conn)
807 .await?;
808 for label in &labels {
809 sqlx::query(
810 "INSERT OR IGNORE INTO task_labels(workspace_id, task_id, label) VALUES (?, ?, ?)",
811 )
812 .bind(&workspace.id)
813 .bind(&id)
814 .bind(label)
815 .execute(&mut *conn)
816 .await?;
817 }
818 let change_id = append_change(
819 conn,
820 ChangeEntity::Task,
821 &id,
822 None,
823 op_type::CREATE_TASK,
824 ChangePayload::workspace(workspace)
825 .set("title", draft.title)
826 .set("description", draft.description)
827 .set("project_id", project.id.clone())
828 .set("project_key", project.key.clone())
829 .set("project_name", project.name.clone())
830 .set("project_prefix", project.prefix.clone())
831 .set("status", status.as_str())
832 .set("priority", priority.as_str())
833 .set("source", draft.source.as_str())
834 .set("available_at", available_at)
835 .set("due_on", due_on)
836 .set("is_epic", if draft.is_epic { "1" } else { "0" })
837 .set("labels", &labels)
838 .set("created_at", ts),
839 )
840 .await?;
841 for field in TaskField::VERSIONED {
842 set_field_version(conn, &id, field.as_str(), &change_id).await?;
843 }
844 Ok(InsertedTask {
845 id,
846 change_id,
847 project_key: project.key,
848 label_count: labels.len(),
849 })
850}
851
852async fn cleanup_attachment_guards(
853 conn: &mut SqliteConnection,
854 leases: &[String],
855 reservations: &[String],
856) {
857 for lease_id in leases {
858 if let Err(error) = crate::attachments::lifecycle::release_lease(conn, lease_id).await {
859 warn!(%error, "failed to release attachment staging lease");
860 }
861 }
862 for reservation_id in reservations {
863 if let Err(error) =
864 crate::attachments::lifecycle::release_reservation(conn, reservation_id).await
865 {
866 warn!(%error, "failed to release attachment capacity reservation");
867 }
868 }
869}
870
871async fn cleanup_created_objects(conn: &mut SqliteConnection, blob_dir: &Path, hashes: &[String]) {
872 for sha256 in hashes {
873 crate::attachments::storage::remove_staged_blob_if_unreferenced(conn, blob_dir, sha256)
874 .await;
875 }
876}
877
878#[cfg(any(test, feature = "test-support"))]
879pub async fn update_task(
880 conn: &mut SqliteConnection,
881 workspace: &Workspace,
882 task_id: &crate::ids::TaskId,
883 update: TaskUpdate,
884) -> Result<TaskUpdateOutcome> {
885 validate_task_update(&update)?;
886 let mut tx = begin_immediate(conn).await?;
887 let changed = apply_task_update(&mut tx, workspace, task_id, &update).await?;
888 tx.commit().await?;
889 Ok(TaskUpdateOutcome {
890 task: get_task_in_workspace(conn, workspace, task_id).await?,
891 changed,
892 })
893}
894
895fn append_task_undo_commands(
896 task_id: &TaskId,
897 before: &TaskUndoSnapshot,
898 after: &TaskUndoSnapshot,
899 commands: &mut Vec<UndoCommand>,
900) {
901 let fields = [
902 ("title", before.title.as_str(), after.title.as_str()),
903 (
904 "description",
905 before.description.as_str(),
906 after.description.as_str(),
907 ),
908 (
909 "project",
910 before.project_id.as_str(),
911 after.project_id.as_str(),
912 ),
913 ("status", before.status.as_str(), after.status.as_str()),
914 (
915 "priority",
916 before.priority.as_str(),
917 after.priority.as_str(),
918 ),
919 (
920 "available_at",
921 before.available_at.as_str(),
922 after.available_at.as_str(),
923 ),
924 ("due_on", before.due_on.as_str(), after.due_on.as_str()),
925 ];
926 for (field, before, after) in fields {
927 if before != after {
928 commands.push(UndoCommand::SetTaskField {
929 task_id: task_id.clone(),
930 field: field.to_string(),
931 before: before.to_string(),
932 after: after.to_string(),
933 });
934 }
935 }
936 let before_deleted = if before.deleted { "1" } else { "0" };
937 let after_deleted = if after.deleted { "1" } else { "0" };
938 if before_deleted != after_deleted {
939 commands.push(UndoCommand::SetTaskField {
940 task_id: task_id.clone(),
941 field: "deleted".to_string(),
942 before: before_deleted.to_string(),
943 after: after_deleted.to_string(),
944 });
945 }
946 let before_is_epic = if before.is_epic { "1" } else { "0" };
947 let after_is_epic = if after.is_epic { "1" } else { "0" };
948 if before_is_epic != after_is_epic {
949 commands.push(UndoCommand::SetTaskField {
950 task_id: task_id.clone(),
951 field: "is_epic".to_string(),
952 before: before_is_epic.to_string(),
953 after: after_is_epic.to_string(),
954 });
955 }
956 if before.labels != after.labels {
957 commands.push(UndoCommand::SetTaskLabels {
958 task_id: task_id.clone(),
959 before: before.labels.clone(),
960 after: after.labels.clone(),
961 });
962 }
963}
964
965fn cycled_priority(priority: TaskPriority, reverse: bool) -> TaskPriority {
966 let index = TaskPriority::ALL
967 .iter()
968 .position(|candidate| *candidate == priority)
969 .unwrap_or(0);
970 let next = if reverse {
971 (index + TaskPriority::ALL.len() - 1) % TaskPriority::ALL.len()
972 } else {
973 (index + 1) % TaskPriority::ALL.len()
974 };
975 TaskPriority::ALL[next]
976}
977
978fn task_update_requires_snapshot(update: &TaskUpdate) -> bool {
979 update.cycle_priority.is_some() || update.label_selection.is_some()
980}
981
982fn materialize_task_update(update: &TaskUpdate, before: &TaskUndoSnapshot) -> Result<TaskUpdate> {
983 let mut update = update.clone();
984 if let Some(reverse) = update.cycle_priority.take() {
985 let priority = TaskPriority::parse(&before.priority)?;
986 update.priority = Some(cycled_priority(priority, reverse).to_string());
987 }
988 if let Some(selection) = update.label_selection.take() {
989 let selected = selection.selected.iter().cloned().collect::<BTreeSet<_>>();
990 let partial = selection.partial.iter().cloned().collect::<BTreeSet<_>>();
991 update.add_labels = selection
992 .selected
993 .into_iter()
994 .filter(|label| !before.labels.contains(label))
995 .collect();
996 update.remove_labels = before
997 .labels
998 .iter()
999 .filter(|label| !selected.contains(label.as_str()) && !partial.contains(label.as_str()))
1000 .cloned()
1001 .collect();
1002 }
1003 Ok(update)
1004}
1005
1006fn validate_task_update(update: &TaskUpdate) -> Result<()> {
1007 if update.priority.is_some() && update.cycle_priority.is_some() {
1008 bail!("error invalid-task-update priority and priority cycle are mutually exclusive");
1009 }
1010 if update.label_selection.is_some()
1011 && (!update.add_labels.is_empty() || !update.remove_labels.is_empty())
1012 {
1013 bail!("error invalid-task-update label selection and label deltas are mutually exclusive");
1014 }
1015 if let Some(status) = update.status.as_deref() {
1016 TaskStatus::parse(status)?;
1017 }
1018 if let Some(priority) = update.priority.as_deref() {
1019 TaskPriority::parse(priority)?;
1020 }
1021 if let Some(Some(available_at)) = update.available_at.as_ref() {
1022 crate::time_validation::validate_available_at_value(available_at)?;
1023 }
1024 if let Some(Some(due_on)) = update.due_on.as_ref() {
1025 crate::time_validation::validate_due_on_value(due_on)?;
1026 }
1027 Ok(())
1028}
1029
1030async fn apply_task_update(
1031 conn: &mut SqliteConnection,
1032 workspace: &Workspace,
1033 task_id: &crate::ids::TaskId,
1034 update: &TaskUpdate,
1035) -> Result<bool> {
1036 let mut changed = false;
1037 if let Some(title) = update.title.as_deref() {
1038 changed |= update_task_field(conn, workspace, task_id, "title", title).await?;
1039 }
1040 if let Some(description) = update.description.as_deref() {
1041 changed |= update_task_field(conn, workspace, task_id, "description", description).await?;
1042 }
1043 if let Some(project) = update.project.as_deref() {
1044 let project = resolve_or_create_project_in_workspace(conn, &workspace.id, project).await?;
1045 changed |= set_task_project(conn, workspace, task_id, &project).await?;
1046 }
1047 if let Some(status) = update.status.as_deref() {
1048 changed |= update_task_field(conn, workspace, task_id, "status", status).await?;
1049 }
1050 if let Some(priority) = update.priority.as_deref() {
1051 changed |= update_task_field(conn, workspace, task_id, "priority", priority).await?;
1052 }
1053 if let Some(available_at) = update.available_at.as_ref() {
1054 changed |= update_task_field(
1055 conn,
1056 workspace,
1057 task_id,
1058 "available_at",
1059 available_at.as_deref().unwrap_or(""),
1060 )
1061 .await?;
1062 }
1063 if let Some(due_on) = update.due_on.as_ref() {
1064 changed |= update_task_field(
1065 conn,
1066 workspace,
1067 task_id,
1068 "due_on",
1069 due_on.as_deref().unwrap_or(""),
1070 )
1071 .await?;
1072 }
1073 if let Some(deleted) = update.deleted {
1074 changed |= update_task_field(
1075 conn,
1076 workspace,
1077 task_id,
1078 "deleted",
1079 if deleted { "1" } else { "0" },
1080 )
1081 .await?;
1082 crate::attachments::lifecycle::reconcile_liveness_in_transaction(
1083 conn,
1084 &crate::attachments::lifecycle::SystemClock,
1085 )
1086 .await?;
1087 }
1088 if let Some(is_epic) = update.is_epic {
1089 if !is_epic {
1090 let task = get_task_in_workspace(conn, workspace, task_id).await?;
1091 if super::epics::task_has_epic_children(conn, &task.workspace_id, task_id).await? {
1092 bail!("error epic-has-children task_id={task_id}");
1093 }
1094 }
1095 changed |= update_task_field(
1096 conn,
1097 workspace,
1098 task_id,
1099 "is_epic",
1100 if is_epic { "1" } else { "0" },
1101 )
1102 .await?;
1103 }
1104 changed |= update_task_labels_in_workspace(
1105 conn,
1106 &workspace.id,
1107 task_id,
1108 &update.add_labels,
1109 &update.remove_labels,
1110 )
1111 .await?;
1112 Ok(changed)
1113}
1114
1115pub async fn update_task_field(
1116 conn: &mut SqliteConnection,
1117 workspace: &Workspace,
1118 task_id: &crate::ids::TaskId,
1119 field: &str,
1120 value: &str,
1121) -> Result<bool> {
1122 set_task_field(conn, workspace, task_id, field, value).await
1123}
1124
1125pub async fn update_task_labels_in_workspace(
1126 conn: &mut SqliteConnection,
1127 workspace_id: &WorkspaceId,
1128 task_id: &crate::ids::TaskId,
1129 add_labels: &[String],
1130 remove_labels: &[String],
1131) -> Result<bool> {
1132 let workspace = crate::workspaces::workspace_for_id(conn, workspace_id).await?;
1133 crate::operations::route_recurrence_task_field(conn, &workspace, task_id, "labels", "").await?;
1134 let mut changed = false;
1135 for label in resolve_labels_in_workspace(conn, &workspace.id, add_labels).await? {
1136 let rows_affected = sqlx::query(
1137 "INSERT OR IGNORE INTO task_labels(workspace_id, task_id, label) VALUES (?, ?, ?)",
1138 )
1139 .bind(&workspace.id)
1140 .bind(task_id)
1141 .bind(&label)
1142 .execute(&mut *conn)
1143 .await?
1144 .rows_affected();
1145 if rows_affected > 0 {
1146 append_change(
1147 conn,
1148 ChangeEntity::Task,
1149 task_id,
1150 Some("labels"),
1151 op_type::LABEL_ADD,
1152 ChangePayload::workspace(&workspace).set("label", label),
1153 )
1154 .await?;
1155 changed = true;
1156 }
1157 }
1158 for label in resolve_labels_in_workspace(conn, &workspace.id, remove_labels).await? {
1159 let rows_affected = sqlx::query(
1160 "DELETE FROM task_labels WHERE workspace_id = ? AND task_id = ? AND label = ?",
1161 )
1162 .bind(&workspace.id)
1163 .bind(task_id)
1164 .bind(&label)
1165 .execute(&mut *conn)
1166 .await?
1167 .rows_affected();
1168 if rows_affected > 0 {
1169 append_change(
1170 conn,
1171 ChangeEntity::Task,
1172 task_id,
1173 Some("labels"),
1174 op_type::LABEL_REMOVE,
1175 ChangePayload::workspace(&workspace).set("label", label),
1176 )
1177 .await?;
1178 changed = true;
1179 }
1180 }
1181 if changed {
1182 info!(
1183 task_id = %task_id,
1184 added = add_labels.len(),
1185 removed = remove_labels.len(),
1186 "task labels changed"
1187 );
1188 }
1189 Ok(changed)
1190}
1191
1192pub(super) async fn add_note_operation(
1193 conn: &mut SqliteConnection,
1194 workspace: &Workspace,
1195 task_id: &crate::ids::TaskId,
1196 body: String,
1197 tui_undo: bool,
1198) -> Result<NoteOutcome> {
1199 let note_id = new_id();
1200 let ts = now();
1201 let mut tx = begin_immediate(conn).await?;
1202 crate::operations::route_recurrence_task_field(&mut tx, workspace, task_id, "notes", "")
1203 .await?;
1204 let change_id = append_change(
1205 &mut tx,
1206 ChangeEntity::Task,
1207 task_id,
1208 Some("notes"),
1209 op_type::NOTE_ADD,
1210 ChangePayload::workspace(workspace)
1211 .set("note_id", ¬e_id)
1212 .set("body", &body)
1213 .set("created_at", &ts),
1214 )
1215 .await?;
1216 sqlx::query(
1217 "INSERT INTO notes(workspace_id, id, task_id, body, created_at, change_id) VALUES (?, ?, ?, ?, ?, ?)",
1218 )
1219 .bind(&workspace.id)
1220 .bind(¬e_id)
1221 .bind(task_id)
1222 .bind(&body)
1223 .bind(&ts)
1224 .bind(&change_id)
1225 .execute(&mut *tx)
1226 .await?;
1227 sqlx::query("UPDATE tasks SET queue_activity_at = ? WHERE workspace_id = ? AND id = ?")
1228 .bind(&ts)
1229 .bind(&workspace.id)
1230 .bind(task_id)
1231 .execute(&mut *tx)
1232 .await?;
1233 if tui_undo {
1234 record_tui_undo(
1235 &mut tx,
1236 &workspace.id,
1237 &format!("note {note_id}"),
1238 UndoPayload {
1239 commands: vec![UndoCommand::DeleteCreatedNote {
1240 task_id: task_id.clone(),
1241 note_id: note_id.clone(),
1242 note_add_change_id: change_id.clone(),
1243 }],
1244 },
1245 )
1246 .await?;
1247 }
1248 tx.commit().await?;
1249 info!(task_id = %task_id, note_id = %note_id, "note added");
1250 Ok(NoteOutcome {
1251 task_id: task_id.clone(),
1252 note_id,
1253 change_id,
1254 })
1255}
1256
1257pub async fn delete_note(
1258 conn: &mut SqliteConnection,
1259 workspace: &Workspace,
1260 task_id: &crate::ids::TaskId,
1261 note_id: &str,
1262) -> Result<NoteDeleteOutcome> {
1263 let mut tx = begin_immediate(conn).await?;
1264 crate::operations::route_recurrence_task_field(&mut tx, workspace, task_id, "notes", "")
1265 .await?;
1266 let deleted_at = now();
1267 let deleted =
1268 sqlx::query("DELETE FROM notes WHERE workspace_id = ? AND task_id = ? AND id = ?")
1269 .bind(&workspace.id)
1270 .bind(task_id)
1271 .bind(note_id)
1272 .execute(&mut *tx)
1273 .await?
1274 .rows_affected();
1275 if deleted > 0 {
1276 sqlx::query("UPDATE tasks SET queue_activity_at = ? WHERE workspace_id = ? AND id = ?")
1277 .bind(&deleted_at)
1278 .bind(&workspace.id)
1279 .bind(task_id)
1280 .execute(&mut *tx)
1281 .await?;
1282 append_change(
1283 &mut tx,
1284 ChangeEntity::Task,
1285 task_id,
1286 Some("notes"),
1287 op_type::NOTE_DELETE,
1288 ChangePayload::workspace(workspace)
1289 .set("note_id", note_id)
1290 .set("deleted_at", deleted_at),
1291 )
1292 .await?;
1293 }
1294 tx.commit().await?;
1295 if deleted > 0 {
1296 info!(task_id = %task_id, note_id = %note_id, "note deleted");
1297 }
1298 Ok(NoteDeleteOutcome {
1299 task_id: task_id.clone(),
1300 note_id: note_id.to_string(),
1301 changed: deleted > 0,
1302 })
1303}