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