aven_core/operations/tasks/
change_set.rs

1use std::collections::{BTreeMap, BTreeSet, HashMap};
2
3use anyhow::{Result, ensure};
4use sqlx::SqliteConnection;
5
6use super::creation::{initial_status, insert_task, validate_task_draft};
7use super::mutation::{
8    append_task_undo_commands, apply_task_update, update_requests_inbox_promotion,
9    validate_task_update,
10};
11use super::notes::add_note_in_transaction;
12use super::{TaskDraft, TaskUpdate};
13use crate::choices::{TaskPriority, TaskStatus};
14use crate::db::{Database, begin_immediate, conflict_exists};
15use crate::error::CoreError;
16use crate::ids::{ProjectId, TaskId, WorkspaceId};
17use crate::labels::resolve_labels_in_workspace;
18use crate::projects::{find_project_by_id_in_workspace, resolve_existing_project_in_workspace};
19use crate::refs::{DisplayRefContext, get_task_in_workspace};
20use crate::task_fields::TaskField;
21use crate::types::{Project, Task};
22use crate::undo::{TaskUndoSnapshot, UndoCommand, apply_consumer_undo_commands, task_snapshot};
23use crate::workspaces::{Workspace, workspace_for_id};
24
25#[derive(Debug, Clone, PartialEq, Eq, Hash)]
26pub enum ChangeTarget {
27    Existing(TaskId),
28    /// A nonempty local ID introduced by an earlier Create in the same set.
29    Local(String),
30}
31
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct CreatedTaskRef {
34    pub task_id: TaskId,
35    pub display_ref: String,
36}
37
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct ChangeSetOutcome {
40    pub created: BTreeMap<String, CreatedTaskRef>,
41    pub undo_token: String,
42}
43
44pub(crate) enum ChangeSetOperation {
45    Create {
46        local_id: String,
47        draft: TaskDraft,
48    },
49    Update {
50        target: ChangeTarget,
51        project_id: Option<ProjectId>,
52        update: TaskUpdate,
53    },
54    AddNote {
55        target: ChangeTarget,
56        body: String,
57    },
58}
59
60pub(crate) struct IndexedChangeSetError {
61    pub operation_index: Option<usize>,
62    pub error: anyhow::Error,
63}
64
65impl From<anyhow::Error> for IndexedChangeSetError {
66    fn from(error: anyhow::Error) -> Self {
67        Self {
68            operation_index: None,
69            error,
70        }
71    }
72}
73
74impl IndexedChangeSetError {
75    fn at(index: usize, error: anyhow::Error) -> Self {
76        Self {
77            operation_index: Some(index),
78            error,
79        }
80    }
81}
82
83pub(crate) struct AppliedChangeSet {
84    pub outcome: ChangeSetOutcome,
85    pub note_ids: Vec<String>,
86}
87
88#[derive(serde::Serialize, serde::Deserialize)]
89#[serde(tag = "kind", rename = "change_set")]
90pub(crate) struct ChangeSetUndoToken {
91    workspace_id: WorkspaceId,
92    expected: BTreeMap<TaskId, TaskUndoSnapshot>,
93    notes: Vec<NoteGuard>,
94    groups: Vec<InverseGroup>,
95}
96
97#[derive(serde::Serialize, serde::Deserialize)]
98struct NoteGuard {
99    task_id: TaskId,
100    note_id: String,
101    body: String,
102    change_id: String,
103}
104
105#[derive(serde::Serialize, serde::Deserialize)]
106struct InverseGroup {
107    task_id: TaskId,
108    queue_activity_before: Option<String>,
109    commands: Vec<UndoCommand>,
110}
111
112impl Database {
113    pub(crate) async fn apply_consumer_change_set(
114        &self,
115        workspace_id: &WorkspaceId,
116        ops: Vec<ChangeSetOperation>,
117    ) -> std::result::Result<AppliedChangeSet, IndexedChangeSetError> {
118        let mut conn = self.acquire_writer().await?;
119        let mut tx = begin_immediate(&mut conn)
120            .await
121            .map_err(anyhow::Error::from)?;
122        let workspace = workspace_for_id(&mut tx, workspace_id).await?;
123        validate_operations(&mut tx, &workspace, &ops).await?;
124        let mut locals = BTreeMap::new();
125        let mut token = ChangeSetUndoToken {
126            workspace_id: workspace_id.clone(),
127            expected: BTreeMap::new(),
128            notes: Vec::new(),
129            groups: Vec::new(),
130        };
131        for (index, op) in ops.into_iter().enumerate() {
132            execute_operation(&mut tx, &workspace, op, &mut locals, &mut token)
133                .await
134                .map_err(|error| IndexedChangeSetError::at(index, error))?;
135        }
136        // Final guards are checked once, not against intermediate states of repeated edits.
137        for group in &token.groups {
138            if let std::collections::btree_map::Entry::Vacant(entry) =
139                token.expected.entry(group.task_id.clone())
140            {
141                entry.insert(task_snapshot(&mut tx, workspace_id, &group.task_id).await?);
142            }
143        }
144        let refs = DisplayRefContext::for_workspace(&mut tx, workspace_id).await?;
145        let mut created = BTreeMap::new();
146        for (local_id, task_id) in locals {
147            let task = get_task_in_workspace(&mut tx, &workspace, &task_id).await?;
148            created.insert(
149                local_id,
150                CreatedTaskRef {
151                    task_id,
152                    display_ref: refs.display_ref(&task),
153                },
154            );
155        }
156        let undo_token = serde_json::to_string(&token).map_err(anyhow::Error::from)?;
157        let note_ids = token.notes.into_iter().map(|note| note.note_id).collect();
158        tx.commit().await.map_err(anyhow::Error::from)?;
159        Ok(AppliedChangeSet {
160            outcome: ChangeSetOutcome {
161                created,
162                undo_token,
163            },
164            note_ids,
165        })
166    }
167
168    pub(crate) async fn undo_consumer_change_set(&self, token: ChangeSetUndoToken) -> Result<()> {
169        let mut conn = self.acquire_writer().await?;
170        let mut tx = begin_immediate(&mut conn).await?;
171        workspace_for_id(&mut tx, &token.workspace_id).await?;
172        for (task_id, expected) in &token.expected {
173            let current = task_snapshot(&mut tx, &token.workspace_id, task_id).await?;
174            ensure!(
175                current == *expected,
176                CoreError::generation_conflict("task changed after the change set")
177            );
178        }
179        for note in &token.notes {
180            let current: Option<(String, String)> = sqlx::query_as(
181                "SELECT body, change_id FROM notes WHERE workspace_id = ? AND task_id = ? AND id = ?",
182            ).bind(&token.workspace_id).bind(&note.task_id).bind(&note.note_id)
183                .fetch_optional(&mut *tx).await?;
184            ensure!(
185                current
186                    .as_ref()
187                    .is_some_and(|(body, change)| body == &note.body && change == &note.change_id),
188                CoreError::generation_conflict("note changed after the change set")
189            );
190        }
191        for group in token.groups.into_iter().rev() {
192            apply_consumer_undo_commands(&mut tx, &token.workspace_id, &group.commands).await?;
193            if let Some(activity) = group.queue_activity_before {
194                sqlx::query(
195                    "UPDATE tasks SET queue_activity_at = ? WHERE workspace_id = ? AND id = ?",
196                )
197                .bind(activity)
198                .bind(&token.workspace_id)
199                .bind(&group.task_id)
200                .execute(&mut *tx)
201                .await?;
202            }
203        }
204        tx.commit().await?;
205        Ok(())
206    }
207}
208
209async fn validate_operations(
210    conn: &mut SqliteConnection,
211    workspace: &Workspace,
212    ops: &[ChangeSetOperation],
213) -> std::result::Result<(), IndexedChangeSetError> {
214    let mut tasks = HashMap::<ChangeTarget, Task>::new();
215    let at = crate::ids::now_utc();
216    for (index, op) in ops.iter().enumerate() {
217        let result: Result<()> = async {
218            match op {
219                ChangeSetOperation::Create { local_id, draft } => {
220                    ensure!(
221                        !local_id.trim().is_empty(),
222                        CoreError::validation("local ID must not be empty")
223                    );
224                    let target = ChangeTarget::Local(local_id.clone());
225                    ensure!(
226                        !tasks.contains_key(&target),
227                        CoreError::validation("duplicate local ID")
228                    );
229                    validate_task_draft(draft)?;
230                    let project = resolve_existing_project_in_workspace(
231                        conn,
232                        &workspace.id,
233                        draft
234                            .project
235                            .as_deref()
236                            .expect("capture requires a project"),
237                    )
238                    .await?;
239                    resolve_labels_in_workspace(conn, &workspace.id, &draft.labels).await?;
240                    let (status, priority) = initial_status(draft)?;
241                    tasks.insert(
242                        target,
243                        Task {
244                            id: TaskId::new(),
245                            workspace_id: workspace.id.clone(),
246                            title: draft.title.clone(),
247                            description: draft.description.clone(),
248                            project_id: project.id,
249                            project_key: project.key,
250                            project_prefix: project.prefix,
251                            status,
252                            priority,
253                            source: draft.source,
254                            created_at: String::new(),
255                            updated_at: String::new(),
256                            queue_activity_at: String::new(),
257                            available_at: draft.available_at.clone(),
258                            due_on: draft.due_on.clone(),
259                            deleted: false,
260                            is_epic: false,
261                        },
262                    );
263                }
264                ChangeSetOperation::Update {
265                    target,
266                    project_id,
267                    update,
268                } => {
269                    let task = effective_task(conn, workspace, target, &mut tasks).await?;
270                    validate_task_update(update)?;
271                    let project = match project_id {
272                        Some(id) => Some(require_project(conn, workspace, id).await?),
273                        None => None,
274                    };
275                    let fields = consumer_update_fields(task, project.as_ref(), update);
276                    let target_status = fields
277                        .iter()
278                        .find(|(field, _)| *field == TaskField::Status)
279                        .map(|(_, value)| TaskStatus::parse(value))
280                        .transpose()?;
281                    if matches!(target, ChangeTarget::Existing(_)) {
282                        crate::operations::recurrence::validate_consumer_recurrence_mutation(
283                            conn,
284                            workspace,
285                            &task.id,
286                            task.status,
287                            target_status,
288                            at,
289                        )
290                        .await?;
291                    }
292                    for (field, value) in fields {
293                        if field.current_value(task) != value
294                            && matches!(target, ChangeTarget::Existing(_))
295                        {
296                            ensure!(
297                                !conflict_exists(conn, &workspace.id, &task.id, field.as_str())
298                                    .await?,
299                                CoreError::open_conflict(format!(
300                                    "error conflicted-field ref={} field={}",
301                                    task.id,
302                                    field.as_str()
303                                ))
304                            );
305                        }
306                        set_effective_field(task, field, value)?;
307                    }
308                    resolve_labels_in_workspace(conn, &workspace.id, &update.add_labels).await?;
309                    resolve_labels_in_workspace(conn, &workspace.id, &update.remove_labels).await?;
310                    if matches!(target, ChangeTarget::Existing(_)) {
311                        crate::metadata::validate_task_metadata_result(
312                            conn,
313                            &workspace.id,
314                            &task.id,
315                            &[],
316                            &[],
317                        )
318                        .await?;
319                    }
320                }
321                ChangeSetOperation::AddNote { target, .. } => {
322                    let task = effective_task(conn, workspace, target, &mut tasks).await?;
323                    if matches!(target, ChangeTarget::Existing(_)) {
324                        crate::operations::recurrence::validate_consumer_recurrence_mutation(
325                            conn,
326                            workspace,
327                            &task.id,
328                            task.status,
329                            None,
330                            at,
331                        )
332                        .await?;
333                    }
334                }
335            }
336            Ok(())
337        }
338        .await;
339        result.map_err(|error| IndexedChangeSetError::at(index, error))?;
340    }
341    Ok(())
342}
343
344async fn effective_task<'a>(
345    conn: &mut SqliteConnection,
346    workspace: &Workspace,
347    target: &ChangeTarget,
348    tasks: &'a mut HashMap<ChangeTarget, Task>,
349) -> Result<&'a mut Task> {
350    if !tasks.contains_key(target) {
351        let ChangeTarget::Existing(task_id) = target else {
352            return Err(
353                CoreError::validation("local target must refer to an earlier create").into(),
354            );
355        };
356        let task = get_task_in_workspace(conn, workspace, task_id).await?;
357        ensure!(!task.deleted, CoreError::not_found("task is deleted"));
358        tasks.insert(target.clone(), task);
359    }
360    Ok(tasks.get_mut(target).expect("effective task loaded"))
361}
362
363async fn require_project(
364    conn: &mut SqliteConnection,
365    workspace: &Workspace,
366    id: &ProjectId,
367) -> Result<Project> {
368    find_project_by_id_in_workspace(conn, &workspace.id, id)
369        .await?
370        .ok_or_else(|| CoreError::not_found("selected project is unavailable").into())
371}
372
373fn consumer_update_fields(
374    task: &Task,
375    project: Option<&Project>,
376    update: &TaskUpdate,
377) -> Vec<(TaskField, String)> {
378    let mut fields = Vec::new();
379    if let Some(project) = project {
380        fields.push((TaskField::Project, project.id.to_string()));
381    }
382    if let Some(value) = &update.title {
383        fields.push((TaskField::Title, value.clone()));
384    }
385    if let Some(value) = &update.description {
386        fields.push((TaskField::Description, value.clone()));
387    }
388    let promote = task.status == TaskStatus::Inbox && update_requests_inbox_promotion(update);
389    if promote {
390        fields.push((TaskField::Status, "todo".into()));
391    } else if let Some(value) = &update.status {
392        fields.push((TaskField::Status, value.clone()));
393    }
394    if let Some(value) = &update.priority {
395        fields.push((TaskField::Priority, value.clone()));
396    }
397    if let Some(value) = &update.available_at {
398        fields.push((TaskField::AvailableAt, value.clone().unwrap_or_default()));
399    }
400    if let Some(value) = &update.due_on {
401        fields.push((TaskField::DueOn, value.clone().unwrap_or_default()));
402    }
403    fields
404}
405
406fn set_effective_field(task: &mut Task, field: TaskField, value: String) -> Result<()> {
407    match field {
408        TaskField::Title => task.title = value,
409        TaskField::Description => task.description = value,
410        TaskField::Project => task.project_id = value.parse()?,
411        TaskField::Status => task.status = TaskStatus::parse(&value)?,
412        TaskField::Priority => task.priority = TaskPriority::parse(&value)?,
413        TaskField::AvailableAt => task.available_at = (!value.is_empty()).then_some(value),
414        TaskField::DueOn => task.due_on = (!value.is_empty()).then_some(value),
415        TaskField::Deleted | TaskField::IsEpic => {
416            unreachable!("consumer edit does not expose these fields")
417        }
418    }
419    Ok(())
420}
421
422fn resolve_target(target: ChangeTarget, locals: &BTreeMap<String, TaskId>) -> TaskId {
423    match target {
424        ChangeTarget::Existing(id) => id,
425        ChangeTarget::Local(local) => locals.get(&local).expect("local target validated").clone(),
426    }
427}
428
429async fn execute_operation(
430    conn: &mut SqliteConnection,
431    workspace: &Workspace,
432    op: ChangeSetOperation,
433    locals: &mut BTreeMap<String, TaskId>,
434    token: &mut ChangeSetUndoToken,
435) -> Result<()> {
436    let group = match op {
437        ChangeSetOperation::Create { local_id, draft } => {
438            let inserted = insert_task(conn, workspace, draft, false, true).await?;
439            let expected = task_snapshot(conn, &workspace.id, &inserted.id).await?;
440            locals.insert(local_id, inserted.id.clone());
441            InverseGroup {
442                task_id: inserted.id.clone(),
443                queue_activity_before: None,
444                commands: vec![UndoCommand::DeleteCreatedTask {
445                    task_id: inserted.id,
446                    create_change_id: None,
447                    expected,
448                    attachment_ids: Vec::new(),
449                    attachment_change_ids: Vec::new(),
450                }],
451            }
452        }
453        ChangeSetOperation::Update {
454            target,
455            project_id,
456            update,
457        } => {
458            let task_id = resolve_target(target, locals);
459            let before = task_snapshot(conn, &workspace.id, &task_id).await?;
460            if let Some(id) = project_id {
461                let project = require_project(conn, workspace, &id).await?;
462                crate::mutation::set_task_project(conn, workspace, &task_id, &project).await?;
463            }
464            apply_task_update(conn, workspace, &task_id, &update, &mut BTreeSet::new()).await?;
465            let after = task_snapshot(conn, &workspace.id, &task_id).await?;
466            let mut commands = Vec::new();
467            append_task_undo_commands(&task_id, &before, &after, &mut commands);
468            InverseGroup {
469                task_id,
470                queue_activity_before: Some(before.queue_activity_at),
471                commands,
472            }
473        }
474        ChangeSetOperation::AddNote { target, body } => {
475            let task_id = resolve_target(target, locals);
476            let before = get_task_in_workspace(conn, workspace, &task_id).await?;
477            let note = add_note_in_transaction(conn, workspace, &task_id, body.clone()).await?;
478            token.notes.push(NoteGuard {
479                task_id: task_id.clone(),
480                note_id: note.note_id.clone(),
481                body,
482                change_id: note.change_id.clone(),
483            });
484            InverseGroup {
485                task_id: task_id.clone(),
486                queue_activity_before: Some(before.queue_activity_at),
487                commands: vec![UndoCommand::DeleteCreatedNote {
488                    task_id,
489                    note_id: note.note_id,
490                    note_add_change_id: note.change_id,
491                    restoration_change_ids: Vec::new(),
492                }],
493            }
494        }
495    };
496    if !group.commands.is_empty() {
497        token.groups.push(group);
498    }
499    Ok(())
500}
aven_core/operations/tasks/change_set.rs

aven_core/operations/tasks/
change_set.rs
