aven_core/
undo.rs

1use crate::ids::{ProjectId, WorkspaceId};
2use std::collections::BTreeSet;
3
4use anyhow::{Result, bail, ensure};
5use sqlx::{Row, SqliteConnection};
6
7use crate::db::{
8    Database, begin_immediate, conflict_exists, field_version, insert_change, set_field_version,
9    task_from_row,
10};
11use crate::ids::{new_id, now};
12use crate::mutation::{apply_field_value_in_workspace, apply_project_id_in_workspace};
13use crate::operations::{
14    ProjectMetadata, insert_project_metadata_change, set_project_metadata,
15    update_task_labels_in_workspace,
16};
17use crate::projects::resolve_project_for_stored_value;
18use crate::task_fields::TaskField;
19use crate::workspaces::workspace_key_for_id;
20
21tokio::task_local! {
22    static APPLYING_UNDO: ();
23}
24
25pub fn is_applying_undo() -> bool {
26    APPLYING_UNDO.try_with(|_| ()).is_ok()
27}
28
29#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
30pub struct UndoPayload {
31    pub commands: Vec<UndoCommand>,
32}
33
34#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
35#[serde(tag = "op", rename_all = "snake_case")]
36pub enum UndoCommand {
37    SetTaskField {
38        task_id: crate::ids::TaskId,
39        field: String,
40        before: String,
41        after: String,
42    },
43    SetTaskLabels {
44        task_id: crate::ids::TaskId,
45        before: Vec<String>,
46        after: Vec<String>,
47    },
48    DeleteCreatedTask {
49        task_id: crate::ids::TaskId,
50        create_change_id: Option<String>,
51        expected: TaskUndoSnapshot,
52        #[serde(default)]
53        attachment_ids: Vec<String>,
54        #[serde(default)]
55        attachment_change_ids: Vec<String>,
56    },
57    DeleteCreatedNote {
58        task_id: crate::ids::TaskId,
59        note_id: String,
60        note_add_change_id: String,
61    },
62    DeleteCreatedProject {
63        project_key: String,
64        create_change_id: String,
65        expected_name: String,
66        expected_prefix: String,
67    },
68    SetProjectMetadata {
69        project_id: ProjectId,
70        before_key: String,
71        before_name: String,
72        before_prefix: String,
73        after_key: String,
74        after_name: String,
75        after_prefix: String,
76    },
77    DeleteCreatedLabel {
78        label: String,
79        create_change_id: String,
80    },
81    RestoreConflictResolution {
82        task_id: crate::ids::TaskId,
83        field: String,
84        before: String,
85        after: String,
86        conflict_id: i64,
87    },
88    AddTaskDependency {
89        task_id: crate::ids::TaskId,
90        depends_on_task_id: crate::ids::TaskId,
91    },
92    RemoveTaskDependency {
93        task_id: crate::ids::TaskId,
94        depends_on_task_id: crate::ids::TaskId,
95    },
96}
97
98#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
99pub struct TaskUndoSnapshot {
100    pub title: String,
101    pub description: String,
102    pub project_id: ProjectId,
103    pub project_key: String,
104    pub status: String,
105    pub priority: String,
106    pub available_at: String,
107    #[serde(default)]
108    pub due_on: String,
109    pub deleted: bool,
110    pub labels: Vec<String>,
111}
112
113pub struct UndoOutcome {
114    pub summary: String,
115    pub task_id: Option<crate::ids::TaskId>,
116    pub include_deleted: Option<bool>,
117    pub project_rename: Option<ProjectRenameUndoOutcome>,
118}
119
120pub struct ProjectRenameUndoOutcome {
121    pub before_key: String,
122    pub after_key: String,
123}
124
125impl Database {
126    pub async fn task_field_value(
127        &self,
128        workspace_id: &WorkspaceId,
129        task_id: &crate::ids::TaskId,
130        field: &str,
131    ) -> Result<String> {
132        let mut conn = self.acquire().await?;
133        task_field_value(&mut conn, workspace_id, task_id, field).await
134    }
135
136    pub async fn task_labels(
137        &self,
138        workspace_id: &WorkspaceId,
139        task_id: &crate::ids::TaskId,
140    ) -> Result<Vec<String>> {
141        let mut conn = self.acquire().await?;
142        task_labels(&mut conn, workspace_id, task_id).await
143    }
144
145    pub async fn task_undo_snapshot(
146        &self,
147        workspace_id: &WorkspaceId,
148        task_id: &crate::ids::TaskId,
149    ) -> Result<TaskUndoSnapshot> {
150        let mut conn = self.acquire().await?;
151        task_snapshot(&mut conn, workspace_id, task_id).await
152    }
153
154    pub async fn conflict_row_id(
155        &self,
156        workspace_id: &WorkspaceId,
157        task_id: &crate::ids::TaskId,
158        field: &str,
159    ) -> Result<i64> {
160        let mut conn = self.acquire().await?;
161        conflict_row_id(&mut conn, workspace_id, task_id, field).await
162    }
163
164    pub async fn record_tui_undo(
165        &self,
166        workspace_id: &WorkspaceId,
167        summary: &str,
168        payload: UndoPayload,
169    ) -> Result<()> {
170        let mut conn = self.acquire().await?;
171        record_tui_undo(&mut conn, workspace_id, summary, payload).await
172    }
173
174    pub async fn clear_pending_tui_undo_entries(&self) -> Result<()> {
175        let mut conn = self.acquire().await?;
176        clear_pending_tui_undo_entries(&mut conn).await
177    }
178
179    pub async fn apply_latest_tui_undo(
180        &self,
181        workspace_id: &WorkspaceId,
182    ) -> Result<Option<UndoOutcome>> {
183        let mut conn = self.acquire().await?;
184        apply_latest_tui_undo(&mut conn, workspace_id).await
185    }
186}
187
188pub(crate) async fn task_field_value(
189    conn: &mut SqliteConnection,
190    workspace_id: &WorkspaceId,
191    task_id: &crate::ids::TaskId,
192    field: &str,
193) -> Result<String> {
194    let task_field = TaskField::parse_or_unknown(field)?;
195    task_field_value_for_field(conn, workspace_id, task_id, task_field).await
196}
197
198async fn task_field_value_for_field(
199    conn: &mut SqliteConnection,
200    workspace_id: &WorkspaceId,
201    task_id: &crate::ids::TaskId,
202    task_field: TaskField,
203) -> Result<String> {
204    let row = sqlx::query(
205        "SELECT t.id, t.workspace_id, t.title, t.description, t.project_id, p.key AS project_key, p.prefix AS project_prefix, t.status, t.priority, t.created_at, t.updated_at, t.queue_activity_at, t.available_at, t.due_on, t.deleted, t.is_epic
206         FROM tasks t JOIN projects p ON p.workspace_id = t.workspace_id AND p.id = t.project_id
207         WHERE t.workspace_id = ? AND t.id = ?",
208    )
209    .bind(workspace_id)
210    .bind(task_id)
211    .fetch_optional(&mut *conn)
212    .await?
213    .ok_or_else(|| anyhow::anyhow!("error task-not-found task_id={task_id}"))?;
214    let task = task_from_row(&row)?;
215    Ok(task_field.current_value(&task))
216}
217
218pub(crate) async fn task_labels(
219    conn: &mut SqliteConnection,
220    workspace_id: &WorkspaceId,
221    task_id: &crate::ids::TaskId,
222) -> Result<Vec<String>> {
223    let rows = sqlx::query(
224        "SELECT label FROM task_labels WHERE workspace_id = ? AND task_id = ? ORDER BY label",
225    )
226    .bind(workspace_id)
227    .bind(task_id)
228    .fetch_all(&mut *conn)
229    .await?;
230    Ok(rows.into_iter().map(|row| row.get("label")).collect())
231}
232
233pub(crate) async fn task_snapshot(
234    conn: &mut SqliteConnection,
235    workspace_id: &WorkspaceId,
236    task_id: &crate::ids::TaskId,
237) -> Result<TaskUndoSnapshot> {
238    let row = sqlx::query(
239        "SELECT t.title, t.description, t.project_id, p.key AS project_key, t.status, t.priority, t.available_at, t.due_on, t.deleted, t.is_epic
240         FROM tasks t JOIN projects p ON p.workspace_id = t.workspace_id AND p.id = t.project_id
241         WHERE t.workspace_id = ? AND t.id = ?",
242    )
243    .bind(workspace_id)
244    .bind(task_id)
245    .fetch_optional(&mut *conn)
246    .await?
247    .ok_or_else(|| anyhow::anyhow!("error task-not-found task_id={task_id}"))?;
248    let labels = task_labels(conn, workspace_id, task_id).await?;
249    Ok(TaskUndoSnapshot {
250        title: row.get("title"),
251        description: row.get("description"),
252        project_id: row.get("project_id"),
253        project_key: row.get("project_key"),
254        status: row.get("status"),
255        priority: row.get("priority"),
256        available_at: row.get("available_at"),
257        due_on: row.get("due_on"),
258        deleted: row.get::<i64, _>("deleted") != 0,
259        labels,
260    })
261}
262
263pub(crate) async fn conflict_row_id(
264    conn: &mut SqliteConnection,
265    workspace_id: &WorkspaceId,
266    task_id: &crate::ids::TaskId,
267    field: &str,
268) -> Result<i64> {
269    sqlx::query_scalar(
270        "SELECT id FROM conflicts
271         WHERE workspace_id = ? AND task_id = ? AND field = ? AND resolved = 0
272         ORDER BY id LIMIT 1",
273    )
274    .bind(workspace_id)
275    .bind(task_id)
276    .bind(field)
277    .fetch_optional(&mut *conn)
278    .await?
279    .ok_or_else(|| anyhow::anyhow!("error conflict-not-found task_id={task_id} field={field}"))
280}
281
282pub(crate) async fn record_tui_undo(
283    conn: &mut SqliteConnection,
284    workspace_id: &WorkspaceId,
285    summary: &str,
286    payload: UndoPayload,
287) -> Result<()> {
288    if is_applying_undo() || !undo_payload_has_effect(&payload) {
289        return Ok(());
290    }
291    let id = new_id();
292    let created_at = now();
293    let seq: i64 = sqlx::query_scalar(
294        "SELECT COALESCE(MAX(seq), 0) + 1 FROM tui_undo_entries WHERE workspace_id = ?",
295    )
296    .bind(workspace_id)
297    .fetch_one(&mut *conn)
298    .await?;
299    let payload = serde_json::to_string(&payload)?;
300    sqlx::query(
301        "INSERT INTO tui_undo_entries(id, workspace_id, summary, payload_version, payload, seq, created_at)
302         VALUES (?, ?, ?, 1, ?, ?, ?)",
303    )
304    .bind(&id)
305    .bind(workspace_id)
306    .bind(summary)
307    .bind(&payload)
308    .bind(seq)
309    .bind(&created_at)
310    .execute(&mut *conn)
311    .await?;
312    prune_consumed_undo_entries(conn, workspace_id).await?;
313    Ok(())
314}
315
316fn undo_payload_has_effect(payload: &UndoPayload) -> bool {
317    payload.commands.iter().any(|command| match command {
318        UndoCommand::SetTaskField { before, after, .. } => before != after,
319        UndoCommand::SetTaskLabels { before, after, .. } => !label_sets_equal(before, after),
320        UndoCommand::SetProjectMetadata {
321            before_key,
322            before_name,
323            before_prefix,
324            after_key,
325            after_name,
326            after_prefix,
327            ..
328        } => before_key != after_key || before_name != after_name || before_prefix != after_prefix,
329        UndoCommand::DeleteCreatedTask { .. }
330        | UndoCommand::DeleteCreatedNote { .. }
331        | UndoCommand::DeleteCreatedProject { .. }
332        | UndoCommand::DeleteCreatedLabel { .. }
333        | UndoCommand::RestoreConflictResolution { .. }
334        | UndoCommand::AddTaskDependency { .. }
335        | UndoCommand::RemoveTaskDependency { .. } => true,
336    })
337}
338
339fn empty_command_outcome() -> CommandOutcome {
340    CommandOutcome {
341        task_id: None,
342        include_deleted: None,
343        project_rename: None,
344    }
345}
346
347async fn prune_consumed_undo_entries(
348    conn: &mut SqliteConnection,
349    workspace_id: &WorkspaceId,
350) -> Result<()> {
351    sqlx::query(
352        "DELETE FROM tui_undo_entries
353         WHERE workspace_id = ? AND undone_at IS NOT NULL AND id NOT IN (
354             SELECT id FROM tui_undo_entries
355             WHERE workspace_id = ? AND undone_at IS NOT NULL
356             ORDER BY undone_at DESC, seq DESC
357             LIMIT 20
358         )",
359    )
360    .bind(workspace_id)
361    .bind(workspace_id)
362    .execute(&mut *conn)
363    .await?;
364    Ok(())
365}
366
367pub(crate) async fn clear_pending_tui_undo_entries(conn: &mut SqliteConnection) -> Result<()> {
368    sqlx::query("DELETE FROM tui_undo_entries WHERE undone_at IS NULL")
369        .execute(&mut *conn)
370        .await?;
371    Ok(())
372}
373
374pub(crate) async fn apply_latest_tui_undo(
375    conn: &mut SqliteConnection,
376    workspace_id: &WorkspaceId,
377) -> Result<Option<UndoOutcome>> {
378    let mut tx = begin_immediate(conn).await?;
379    let row = sqlx::query(
380        "SELECT id, summary, payload FROM tui_undo_entries
381         WHERE workspace_id = ? AND undone_at IS NULL
382         ORDER BY seq DESC
383         LIMIT 1",
384    )
385    .bind(workspace_id)
386    .fetch_optional(&mut *tx)
387    .await?;
388    let Some(row) = row else {
389        return Ok(None);
390    };
391    let entry_id: String = row.get("id");
392    let summary: String = row.get("summary");
393    let payload_text: String = row.get("payload");
394    let undone_at = now();
395    let claimed =
396        sqlx::query("UPDATE tui_undo_entries SET undone_at = ? WHERE id = ? AND undone_at IS NULL")
397            .bind(&undone_at)
398            .bind(&entry_id)
399            .execute(&mut *tx)
400            .await?;
401    ensure!(
402        claimed.rows_affected() == 1,
403        "error undo-entry-claim-failed id={entry_id}"
404    );
405    let payload: UndoPayload = serde_json::from_str(&payload_text)?;
406    let apply_result = APPLYING_UNDO
407        .scope(
408            (),
409            apply_undo_commands(&mut tx, workspace_id, &payload.commands),
410        )
411        .await;
412    match apply_result {
413        Ok(outcome) => {
414            tx.commit().await?;
415            Ok(Some(UndoOutcome {
416                summary,
417                task_id: outcome.task_id,
418                include_deleted: outcome.include_deleted,
419                project_rename: outcome.project_rename,
420            }))
421        }
422        Err(error) => {
423            tx.rollback().await?;
424            Err(error)
425        }
426    }
427}
428
429struct CommandOutcome {
430    task_id: Option<crate::ids::TaskId>,
431    include_deleted: Option<bool>,
432    project_rename: Option<ProjectRenameUndoOutcome>,
433}
434
435async fn apply_undo_commands(
436    conn: &mut SqliteConnection,
437    workspace_id: &WorkspaceId,
438    commands: &[UndoCommand],
439) -> Result<CommandOutcome> {
440    let mut task_id = None;
441    let mut include_deleted = None;
442    let mut project_rename = None;
443    for command in commands {
444        let outcome = apply_undo_command(conn, workspace_id, command).await?;
445        if outcome.task_id.is_some() {
446            task_id = outcome.task_id;
447        }
448        if outcome.include_deleted.is_some() {
449            include_deleted = outcome.include_deleted;
450        }
451        if outcome.project_rename.is_some() {
452            project_rename = outcome.project_rename;
453        }
454    }
455    Ok(CommandOutcome {
456        task_id,
457        include_deleted,
458        project_rename,
459    })
460}
461
462async fn apply_undo_command(
463    conn: &mut SqliteConnection,
464    workspace_id: &WorkspaceId,
465    command: &UndoCommand,
466) -> Result<CommandOutcome> {
467    match command {
468        UndoCommand::SetTaskField {
469            task_id,
470            field,
471            before,
472            after,
473        } => {
474            let task_field = TaskField::parse_or_unknown(field)?;
475            let current =
476                task_field_value_for_field(conn, workspace_id, task_id, task_field).await?;
477            if current != *after {
478                bail!("error undo-state-changed task_id={task_id} field={field}");
479            }
480            if before != after {
481                if task_field == TaskField::Project {
482                    let project_id = before.parse().map_err(|_| {
483                        anyhow::anyhow!("error undo-state-changed task_id={task_id} field={field}")
484                    })?;
485                    if !project_id_exists(conn, workspace_id, &project_id).await? {
486                        bail!("error undo-state-changed task_id={task_id} field={field}");
487                    }
488                }
489                set_task_field_in_workspace(conn, workspace_id, task_id, task_field, before)
490                    .await?;
491            }
492            let include_deleted = if task_field == TaskField::Deleted {
493                Some(before == "1")
494            } else {
495                None
496            };
497            Ok(CommandOutcome {
498                task_id: Some(task_id.clone()),
499                include_deleted,
500                project_rename: None,
501            })
502        }
503        UndoCommand::SetTaskLabels {
504            task_id,
505            before,
506            after,
507        } => {
508            let current = task_labels(conn, workspace_id, task_id).await?;
509            if !label_sets_equal(&current, after) {
510                bail!("error undo-state-changed task_id={task_id} field=labels");
511            }
512            let (add_labels, remove_labels) = label_delta(&current, before);
513            update_task_labels_in_workspace(
514                conn,
515                workspace_id,
516                task_id,
517                &add_labels,
518                &remove_labels,
519            )
520            .await?;
521            Ok(CommandOutcome {
522                task_id: Some(task_id.clone()),
523                include_deleted: None,
524                project_rename: None,
525            })
526        }
527        UndoCommand::DeleteCreatedTask {
528            task_id,
529            create_change_id,
530            expected,
531            attachment_ids,
532            attachment_change_ids,
533        } => {
534            let current = task_snapshot(conn, workspace_id, task_id).await?;
535            if current != *expected {
536                bail!("error undo-state-changed task_id={task_id} field=task");
537            }
538            let current_attachment_ids: Vec<String> = sqlx::query_scalar(
539                "SELECT attachment_id FROM task_attachments
540                 WHERE workspace_id = ? AND task_id = ? AND deleted = 0
541                 ORDER BY created_at, attachment_id",
542            )
543            .bind(workspace_id)
544            .bind(task_id)
545            .fetch_all(&mut *conn)
546            .await?;
547            if let Some(change_id) = create_change_id {
548                let labels_clear = expected.labels.is_empty()
549                    || labels_match_create_change(conn, change_id, &expected.labels).await?;
550                let attachment_changes_clear =
551                    all_changes_unsynced(conn, attachment_change_ids).await?;
552                if change_is_unsynced(conn, change_id).await?
553                    && labels_clear
554                    && attachment_changes_clear
555                    && current_attachment_ids == *attachment_ids
556                {
557                    hard_delete_created_task(
558                        conn,
559                        workspace_id,
560                        task_id,
561                        change_id,
562                        attachment_change_ids,
563                    )
564                    .await?;
565                    return Ok(CommandOutcome {
566                        task_id: Some(task_id.clone()),
567                        include_deleted: None,
568                        project_rename: None,
569                    });
570                }
571            }
572            set_task_field_in_workspace(conn, workspace_id, task_id, TaskField::Deleted, "1")
573                .await?;
574            Ok(CommandOutcome {
575                task_id: Some(task_id.clone()),
576                include_deleted: None,
577                project_rename: None,
578            })
579        }
580        UndoCommand::DeleteCreatedNote {
581            task_id,
582            note_id,
583            note_add_change_id,
584        } => {
585            delete_created_note(conn, workspace_id, task_id, note_id, note_add_change_id).await?;
586            Ok(CommandOutcome {
587                task_id: Some(task_id.clone()),
588                include_deleted: None,
589                project_rename: None,
590            })
591        }
592        UndoCommand::DeleteCreatedProject {
593            project_key,
594            create_change_id,
595            expected_name,
596            expected_prefix,
597        } => {
598            delete_created_project(
599                conn,
600                workspace_id,
601                project_key,
602                create_change_id,
603                expected_name,
604                expected_prefix,
605            )
606            .await?;
607            Ok(empty_command_outcome())
608        }
609        UndoCommand::SetProjectMetadata {
610            project_id,
611            before_key,
612            before_name,
613            before_prefix,
614            after_key,
615            after_name,
616            after_prefix,
617        } => {
618            set_project_metadata_for_undo(
619                conn,
620                workspace_id,
621                project_id,
622                ProjectMetadata {
623                    key: before_key,
624                    name: before_name,
625                    prefix: before_prefix,
626                },
627                ProjectMetadata {
628                    key: after_key,
629                    name: after_name,
630                    prefix: after_prefix,
631                },
632            )
633            .await?;
634            Ok(CommandOutcome {
635                task_id: None,
636                include_deleted: None,
637                project_rename: Some(ProjectRenameUndoOutcome {
638                    before_key: before_key.clone(),
639                    after_key: after_key.clone(),
640                }),
641            })
642        }
643        UndoCommand::DeleteCreatedLabel {
644            label,
645            create_change_id,
646        } => {
647            delete_created_label(conn, workspace_id, label, create_change_id).await?;
648            Ok(empty_command_outcome())
649        }
650        UndoCommand::RestoreConflictResolution {
651            task_id,
652            field,
653            before,
654            after,
655            conflict_id,
656        } => {
657            let task_field = TaskField::parse_or_unknown(field)?;
658            let current =
659                task_field_value_for_field(conn, workspace_id, task_id, task_field).await?;
660            if current != *after {
661                bail!("error undo-state-changed task_id={task_id} field={field}");
662            }
663            set_task_field_in_workspace(conn, workspace_id, task_id, task_field, before).await?;
664            let restored = sqlx::query(
665                "UPDATE conflicts SET resolved = 0 WHERE id = ? AND workspace_id = ? AND resolved = 1",
666            )
667            .bind(conflict_id)
668            .bind(workspace_id)
669            .execute(&mut *conn)
670            .await?;
671            ensure!(
672                restored.rows_affected() == 1,
673                "error undo-state-changed task_id={task_id} field={field}"
674            );
675            Ok(CommandOutcome {
676                task_id: Some(task_id.clone()),
677                include_deleted: None,
678                project_rename: None,
679            })
680        }
681        UndoCommand::AddTaskDependency {
682            task_id,
683            depends_on_task_id,
684        } => {
685            ensure!(
686                dependency_edge_exists(conn, workspace_id, task_id, depends_on_task_id).await?,
687                "error undo-state-changed task_id={task_id} field=dependency"
688            );
689            remove_dependency_for_undo(conn, workspace_id, task_id, depends_on_task_id).await?;
690            Ok(CommandOutcome {
691                task_id: Some(task_id.clone()),
692                include_deleted: None,
693                project_rename: None,
694            })
695        }
696        UndoCommand::RemoveTaskDependency {
697            task_id,
698            depends_on_task_id,
699        } => {
700            ensure!(
701                !dependency_edge_exists(conn, workspace_id, task_id, depends_on_task_id).await?,
702                "error undo-state-changed task_id={task_id} field=dependency"
703            );
704            ensure!(
705                !crate::operations::dependency_path_exists(
706                    conn,
707                    workspace_id,
708                    depends_on_task_id,
709                    task_id,
710                )
711                .await?,
712                "error dependency-cycle task_id={task_id} depends_on_task_id={depends_on_task_id}"
713            );
714            add_dependency_for_undo(conn, workspace_id, task_id, depends_on_task_id).await?;
715            Ok(CommandOutcome {
716                task_id: Some(task_id.clone()),
717                include_deleted: None,
718                project_rename: None,
719            })
720        }
721    }
722}
723
724async fn project_id_exists(
725    conn: &mut SqliteConnection,
726    workspace_id: &WorkspaceId,
727    project_id: &ProjectId,
728) -> Result<bool> {
729    Ok(sqlx::query_scalar::<_, i64>(
730        "SELECT count(*) FROM projects
731         WHERE workspace_id = ? AND id = ? AND deleted = 0",
732    )
733    .bind(workspace_id)
734    .bind(project_id)
735    .fetch_one(&mut *conn)
736    .await?
737        > 0)
738}
739
740async fn set_task_field_in_workspace(
741    conn: &mut SqliteConnection,
742    workspace_id: &WorkspaceId,
743    task_id: &crate::ids::TaskId,
744    task_field: TaskField,
745    value: &str,
746) -> Result<()> {
747    let field = task_field.as_str();
748    if conflict_exists(conn, workspace_id, task_id, field).await? {
749        bail!(
750            "error conflicted-field ref={} field={} hint=\"use conflict resolve\"",
751            task_id,
752            field
753        );
754    }
755    let base = field_version(conn, task_id, field).await?;
756    let workspace_key = workspace_key_for_id(conn, workspace_id).await?;
757    let payload = if task_field.is_project() {
758        let project = resolve_project_for_stored_value(conn, workspace_id, value).await?;
759        apply_project_id_in_workspace(conn, workspace_id, task_id, &project.id).await?;
760        TaskField::project_payload(workspace_id, &workspace_key, &project)
761    } else {
762        apply_field_value_in_workspace(conn, workspace_id, task_id, field, value).await?;
763        task_field.scalar_payload(workspace_id, &workspace_key, value)?
764    };
765    let change_id = insert_change(
766        conn,
767        "task",
768        task_id,
769        Some(field),
770        "set_field",
771        payload,
772        base.as_deref(),
773    )
774    .await?;
775    set_field_version(conn, task_id, field, &change_id).await?;
776    Ok(())
777}
778
779fn label_sets_equal(left: &[String], right: &[String]) -> bool {
780    let left: BTreeSet<_> = left.iter().collect();
781    let right: BTreeSet<_> = right.iter().collect();
782    left == right
783}
784
785fn label_delta(current: &[String], target: &[String]) -> (Vec<String>, Vec<String>) {
786    let current_set: BTreeSet<_> = current.iter().collect();
787    let target_set: BTreeSet<_> = target.iter().collect();
788    let add = target
789        .iter()
790        .filter(|label| !current_set.contains(label))
791        .cloned()
792        .collect();
793    let remove = current
794        .iter()
795        .filter(|label| !target_set.contains(label))
796        .cloned()
797        .collect();
798    (add, remove)
799}
800
801async fn change_is_unsynced(conn: &mut SqliteConnection, change_id: &str) -> Result<bool> {
802    let server_seq =
803        sqlx::query_scalar::<_, Option<i64>>("SELECT server_seq FROM changes WHERE change_id = ?")
804            .bind(change_id)
805            .fetch_optional(&mut *conn)
806            .await?;
807    Ok(matches!(server_seq, Some(None)))
808}
809
810async fn all_changes_unsynced(conn: &mut SqliteConnection, change_ids: &[String]) -> Result<bool> {
811    for change_id in change_ids {
812        if !change_is_unsynced(conn, change_id).await? {
813            return Ok(false);
814        }
815    }
816    Ok(true)
817}
818
819async fn labels_match_create_change(
820    conn: &mut SqliteConnection,
821    change_id: &str,
822    labels: &[String],
823) -> Result<bool> {
824    let payload: String = sqlx::query_scalar("SELECT payload FROM changes WHERE change_id = ?")
825        .bind(change_id)
826        .fetch_one(&mut *conn)
827        .await?;
828    let payload: serde_json::Value = serde_json::from_str(&payload)?;
829    let payload_labels = payload
830        .get("labels")
831        .and_then(|labels| labels.as_array())
832        .map(|labels| {
833            labels
834                .iter()
835                .filter_map(|label| label.as_str().map(str::to_string))
836                .collect::<Vec<_>>()
837        })
838        .unwrap_or_default();
839    Ok(label_sets_equal(labels, &payload_labels))
840}
841
842async fn hard_delete_created_task(
843    conn: &mut SqliteConnection,
844    workspace_id: &WorkspaceId,
845    task_id: &crate::ids::TaskId,
846    create_change_id: &str,
847    attachment_change_ids: &[String],
848) -> Result<()> {
849    sqlx::query("DELETE FROM task_attachments WHERE workspace_id = ? AND task_id = ?")
850        .bind(workspace_id)
851        .bind(task_id)
852        .execute(&mut *conn)
853        .await?;
854    sqlx::query("DELETE FROM task_labels WHERE workspace_id = ? AND task_id = ?")
855        .bind(workspace_id)
856        .bind(task_id)
857        .execute(&mut *conn)
858        .await?;
859    sqlx::query("DELETE FROM field_versions WHERE entity_id = ?")
860        .bind(task_id)
861        .execute(&mut *conn)
862        .await?;
863    sqlx::query("DELETE FROM tasks WHERE workspace_id = ? AND id = ?")
864        .bind(workspace_id)
865        .bind(task_id)
866        .execute(&mut *conn)
867        .await?;
868    for change_id in attachment_change_ids {
869        sqlx::query("DELETE FROM changes WHERE change_id = ?")
870            .bind(change_id)
871            .execute(&mut *conn)
872            .await?;
873    }
874    sqlx::query("DELETE FROM changes WHERE change_id = ?")
875        .bind(create_change_id)
876        .execute(&mut *conn)
877        .await?;
878    crate::attachments::lifecycle::reconcile_liveness_in_transaction(
879        conn,
880        &crate::attachments::lifecycle::SystemClock,
881    )
882    .await?;
883    Ok(())
884}
885
886async fn delete_created_note(
887    conn: &mut SqliteConnection,
888    workspace_id: &WorkspaceId,
889    task_id: &crate::ids::TaskId,
890    note_id: &str,
891    note_add_change_id: &str,
892) -> Result<()> {
893    let row = sqlx::query(
894        "SELECT change_id FROM notes WHERE workspace_id = ? AND id = ? AND task_id = ?",
895    )
896    .bind(workspace_id)
897    .bind(note_id)
898    .bind(task_id)
899    .fetch_optional(&mut *conn)
900    .await?;
901    let Some(row) = row else {
902        bail!("error undo-state-changed task_id={task_id} field=note");
903    };
904    let stored_change_id: String = row.get("change_id");
905    if stored_change_id != note_add_change_id {
906        bail!("error undo-state-changed task_id={task_id} field=note");
907    }
908    if !change_is_unsynced(conn, note_add_change_id).await? {
909        bail!("error undo-state-changed task_id={task_id} field=note");
910    }
911    sqlx::query("DELETE FROM notes WHERE workspace_id = ? AND id = ? AND task_id = ?")
912        .bind(workspace_id)
913        .bind(note_id)
914        .bind(task_id)
915        .execute(&mut *conn)
916        .await?;
917    sqlx::query("DELETE FROM changes WHERE change_id = ?")
918        .bind(note_add_change_id)
919        .execute(&mut *conn)
920        .await?;
921    Ok(())
922}
923
924async fn delete_created_project(
925    conn: &mut SqliteConnection,
926    workspace_id: &WorkspaceId,
927    project_key: &str,
928    create_change_id: &str,
929    expected_name: &str,
930    expected_prefix: &str,
931) -> Result<()> {
932    let row = sqlx::query(
933        "SELECT id, name, prefix FROM projects WHERE workspace_id = ? AND key = ? AND deleted = 0",
934    )
935    .bind(workspace_id)
936    .bind(project_key)
937    .fetch_optional(&mut *conn)
938    .await?;
939    let Some(row) = row else {
940        bail!("error undo-state-changed project_key={project_key}");
941    };
942    let project_id: ProjectId = row.get("id");
943    let name: String = row.get("name");
944    let prefix: String = row.get("prefix");
945    if name != expected_name || prefix != expected_prefix {
946        bail!("error undo-state-changed project_key={project_key}");
947    }
948    if !change_is_unsynced(conn, create_change_id).await? {
949        bail!("error undo-state-changed project_key={project_key}");
950    }
951    let task_refs: i64 =
952        sqlx::query_scalar("SELECT count(*) FROM tasks WHERE workspace_id = ? AND project_id = ?")
953            .bind(workspace_id)
954            .bind(&project_id)
955            .fetch_one(&mut *conn)
956            .await?;
957    if task_refs > 0 {
958        bail!("error undo-state-changed project_key={project_key}");
959    }
960    let path_refs: i64 = sqlx::query_scalar(
961        "SELECT count(*) FROM project_paths WHERE workspace_id = ? AND project_id = ?",
962    )
963    .bind(workspace_id)
964    .bind(&project_id)
965    .fetch_one(&mut *conn)
966    .await?;
967    if path_refs > 0 {
968        bail!("error undo-state-changed project_key={project_key}");
969    }
970    sqlx::query("DELETE FROM projects WHERE workspace_id = ? AND key = ?")
971        .bind(workspace_id)
972        .bind(project_key)
973        .execute(&mut *conn)
974        .await?;
975    sqlx::query("DELETE FROM changes WHERE change_id = ?")
976        .bind(create_change_id)
977        .execute(&mut *conn)
978        .await?;
979    Ok(())
980}
981
982async fn set_project_metadata_for_undo(
983    conn: &mut SqliteConnection,
984    workspace_id: &WorkspaceId,
985    project_id: &ProjectId,
986    before: ProjectMetadata<'_>,
987    after: ProjectMetadata<'_>,
988) -> Result<()> {
989    let workspace = crate::workspaces::workspace_for_id(conn, workspace_id).await?;
990    let row = sqlx::query(
991        "SELECT key, name, prefix
992         FROM projects
993         WHERE workspace_id = ? AND id = ? AND deleted = 0",
994    )
995    .bind(workspace_id)
996    .bind(project_id)
997    .fetch_optional(&mut *conn)
998    .await?;
999    let Some(row) = row else {
1000        bail!("error undo-state-changed project_id={project_id}");
1001    };
1002    let key: String = row.get("key");
1003    let name: String = row.get("name");
1004    let prefix: String = row.get("prefix");
1005    if key != after.key || name != after.name || prefix != after.prefix {
1006        bail!("error undo-state-changed project_id={project_id}");
1007    }
1008    let key_refs: i64 = sqlx::query_scalar(
1009        "SELECT count(*) FROM projects
1010         WHERE workspace_id = ? AND key = ? AND id != ? AND deleted = 0",
1011    )
1012    .bind(workspace_id)
1013    .bind(before.key)
1014    .bind(project_id)
1015    .fetch_one(&mut *conn)
1016    .await?;
1017    if key_refs > 0 {
1018        bail!("error undo-state-changed project_id={project_id}");
1019    }
1020    let prefix_refs: i64 = sqlx::query_scalar(
1021        "SELECT count(*) FROM projects
1022         WHERE workspace_id = ? AND prefix = ? AND id != ? AND deleted = 0",
1023    )
1024    .bind(workspace_id)
1025    .bind(before.prefix)
1026    .bind(project_id)
1027    .fetch_one(&mut *conn)
1028    .await?;
1029    if prefix_refs > 0 {
1030        bail!("error undo-state-changed project_id={project_id}");
1031    }
1032    set_project_metadata(conn, &workspace, project_id, before, false).await?;
1033    insert_project_metadata_change(conn, &workspace, project_id, before, &now()).await?;
1034    Ok(())
1035}
1036
1037async fn delete_created_label(
1038    conn: &mut SqliteConnection,
1039    workspace_id: &WorkspaceId,
1040    label: &str,
1041    create_change_id: &str,
1042) -> Result<()> {
1043    let exists: i64 =
1044        sqlx::query_scalar("SELECT count(*) FROM labels WHERE workspace_id = ? AND name = ?")
1045            .bind(workspace_id)
1046            .bind(label)
1047            .fetch_one(&mut *conn)
1048            .await?;
1049    if exists == 0 || !change_is_unsynced(conn, create_change_id).await? {
1050        bail!("error undo-state-changed label={label}");
1051    }
1052    let refs: i64 =
1053        sqlx::query_scalar("SELECT count(*) FROM task_labels WHERE workspace_id = ? AND label = ?")
1054            .bind(workspace_id)
1055            .bind(label)
1056            .fetch_one(&mut *conn)
1057            .await?;
1058    if refs > 0 {
1059        bail!("error undo-state-changed label={label}");
1060    }
1061    sqlx::query("DELETE FROM labels WHERE workspace_id = ? AND name = ?")
1062        .bind(workspace_id)
1063        .bind(label)
1064        .execute(&mut *conn)
1065        .await?;
1066    sqlx::query("DELETE FROM changes WHERE change_id = ?")
1067        .bind(create_change_id)
1068        .execute(&mut *conn)
1069        .await?;
1070    Ok(())
1071}
1072
1073async fn dependency_task_exists(
1074    conn: &mut SqliteConnection,
1075    workspace_id: &WorkspaceId,
1076    task_id: &crate::ids::TaskId,
1077) -> Result<bool> {
1078    Ok(
1079        sqlx::query_scalar::<_, i64>(
1080            "SELECT count(*) FROM tasks WHERE workspace_id = ? AND id = ?",
1081        )
1082        .bind(workspace_id)
1083        .bind(task_id)
1084        .fetch_one(&mut *conn)
1085        .await?
1086            > 0,
1087    )
1088}
1089
1090async fn dependency_edge_exists(
1091    conn: &mut SqliteConnection,
1092    workspace_id: &WorkspaceId,
1093    task_id: &crate::ids::TaskId,
1094    depends_on_task_id: &crate::ids::TaskId,
1095) -> Result<bool> {
1096    ensure!(
1097        dependency_task_exists(conn, workspace_id, task_id).await?
1098            && dependency_task_exists(conn, workspace_id, depends_on_task_id).await?,
1099        "error undo-state-changed task_id={task_id} field=dependency"
1100    );
1101    Ok(sqlx::query_scalar::<_, i64>(
1102        "SELECT count(*) FROM task_dependencies
1103         WHERE workspace_id = ? AND task_id = ? AND depends_on_task_id = ?",
1104    )
1105    .bind(workspace_id)
1106    .bind(task_id)
1107    .bind(depends_on_task_id)
1108    .fetch_one(&mut *conn)
1109    .await?
1110        > 0)
1111}
1112
1113async fn add_dependency_for_undo(
1114    conn: &mut SqliteConnection,
1115    workspace_id: &WorkspaceId,
1116    task_id: &crate::ids::TaskId,
1117    depends_on_task_id: &crate::ids::TaskId,
1118) -> Result<()> {
1119    let created_at = now();
1120    sqlx::query(
1121        "INSERT OR IGNORE INTO task_dependencies(workspace_id, task_id, depends_on_task_id, created_at)
1122         VALUES (?, ?, ?, ?)",
1123    )
1124    .bind(workspace_id)
1125    .bind(task_id)
1126    .bind(depends_on_task_id)
1127    .bind(&created_at)
1128    .execute(&mut *conn)
1129    .await?;
1130    append_dependency_change(
1131        conn,
1132        workspace_id,
1133        task_id,
1134        depends_on_task_id,
1135        crate::change_log::op_type::DEPENDENCY_ADD,
1136    )
1137    .await
1138}
1139
1140async fn remove_dependency_for_undo(
1141    conn: &mut SqliteConnection,
1142    workspace_id: &WorkspaceId,
1143    task_id: &crate::ids::TaskId,
1144    depends_on_task_id: &crate::ids::TaskId,
1145) -> Result<()> {
1146    sqlx::query(
1147        "DELETE FROM task_dependencies
1148         WHERE workspace_id = ? AND task_id = ? AND depends_on_task_id = ?",
1149    )
1150    .bind(workspace_id)
1151    .bind(task_id)
1152    .bind(depends_on_task_id)
1153    .execute(&mut *conn)
1154    .await?;
1155    append_dependency_change(
1156        conn,
1157        workspace_id,
1158        task_id,
1159        depends_on_task_id,
1160        crate::change_log::op_type::DEPENDENCY_REMOVE,
1161    )
1162    .await
1163}
1164
1165async fn append_dependency_change(
1166    conn: &mut SqliteConnection,
1167    workspace_id: &WorkspaceId,
1168    task_id: &crate::ids::TaskId,
1169    depends_on_task_id: &crate::ids::TaskId,
1170    op_type: &'static str,
1171) -> Result<()> {
1172    let workspace = crate::workspaces::workspace_for_id(conn, workspace_id).await?;
1173    crate::change_log::append_change(
1174        conn,
1175        crate::change_log::ChangeEntity::Task,
1176        task_id,
1177        Some("dependencies"),
1178        op_type,
1179        crate::change_log::ChangePayload::workspace(&workspace)
1180            .set("depends_on_task_id", depends_on_task_id.to_string()),
1181    )
1182    .await?;
1183    Ok(())
1184}
aven_core/undo.rs

aven_core/
undo.rs
