aven_core/
mutation.rs

1use crate::ids::{ProjectId, WorkspaceId};
2use anyhow::{Result, bail, ensure};
3use sqlx::SqliteConnection;
4use tracing::{debug, info};
5
6use crate::change_log::op_type;
7use crate::choices::TaskPriority;
8use crate::db::{
9    Database, begin_immediate, conflict_exists, field_version, insert_change, set_field_version,
10    task_from_row,
11};
12use crate::ids::now;
13use crate::projects::resolve_or_create_project_in_workspace;
14use crate::refs::get_task_in_workspace;
15use crate::task_fields::TaskField;
16use crate::types::{Project, Task};
17use crate::workspaces::Workspace;
18
19#[derive(Debug)]
20pub(crate) struct OpenConflictError {
21    task_id: crate::ids::TaskId,
22    field: &'static str,
23}
24
25impl std::fmt::Display for OpenConflictError {
26    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        write!(
28            formatter,
29            "error conflicted-field ref={} field={} hint=\"use conflict resolve\"",
30            self.task_id, self.field
31        )
32    }
33}
34
35impl std::error::Error for OpenConflictError {}
36
37impl Database {
38    pub async fn set_task_status(
39        &self,
40        workspace: &Workspace,
41        task: &Task,
42        status: &str,
43    ) -> Result<Task> {
44        let mut conn = self.acquire().await?;
45        let mut tx = begin_immediate(&mut conn).await?;
46        let task = set_status(&mut tx, workspace, task, status).await?;
47        tx.commit().await?;
48        Ok(task)
49    }
50
51    pub async fn set_task_priority(
52        &self,
53        workspace: &Workspace,
54        task: &Task,
55        priority: &str,
56    ) -> Result<Task> {
57        let mut conn = self.acquire().await?;
58        let mut tx = begin_immediate(&mut conn).await?;
59        let task = set_priority(&mut tx, workspace, task, priority).await?;
60        tx.commit().await?;
61        Ok(task)
62    }
63
64    pub async fn cycle_task_priority(
65        &self,
66        workspace: &Workspace,
67        task: &Task,
68        reverse: bool,
69    ) -> Result<Task> {
70        let mut conn = self.acquire().await?;
71        let mut tx = begin_immediate(&mut conn).await?;
72        let current = get_task_in_workspace(&mut tx, workspace, &task.id).await?;
73        let task = cycle_priority(&mut tx, workspace, &current, reverse).await?;
74        tx.commit().await?;
75        Ok(task)
76    }
77
78    pub async fn set_task_deleted_state(
79        &self,
80        workspace: &Workspace,
81        task: &Task,
82        deleted: bool,
83    ) -> Result<Task> {
84        let mut conn = self.acquire().await?;
85        let mut tx = begin_immediate(&mut conn).await?;
86        let task = set_deleted(&mut tx, workspace, task, deleted).await?;
87        tx.commit().await?;
88        Ok(task)
89    }
90
91    pub async fn set_task_field(
92        &self,
93        workspace: &Workspace,
94        task_id: &crate::ids::TaskId,
95        field: &str,
96        value: &str,
97    ) -> Result<bool> {
98        let mut conn = self.acquire().await?;
99        let mut tx = begin_immediate(&mut conn).await?;
100        let changed = set_task_field(&mut tx, workspace, task_id, field, value).await?;
101        tx.commit().await?;
102        Ok(changed)
103    }
104
105    pub async fn set_task_project(
106        &self,
107        workspace: &Workspace,
108        task_id: &crate::ids::TaskId,
109        project: &Project,
110    ) -> Result<bool> {
111        let mut conn = self.acquire().await?;
112        let mut tx = begin_immediate(&mut conn).await?;
113        let changed = set_task_project(&mut tx, workspace, task_id, project).await?;
114        tx.commit().await?;
115        Ok(changed)
116    }
117
118    pub async fn set_task_fields(
119        &self,
120        workspace: &Workspace,
121        updates: &[(crate::ids::TaskId, String, String)],
122    ) -> Result<Vec<bool>> {
123        let mut conn = self.acquire().await?;
124        let mut tx = begin_immediate(&mut conn).await?;
125        let mut outcomes = Vec::with_capacity(updates.len());
126        for (task_id, field, value) in updates {
127            outcomes.push(set_task_field(&mut tx, workspace, task_id, field, value).await?);
128        }
129        tx.commit().await?;
130        Ok(outcomes)
131    }
132
133    pub async fn cycle_task_priorities(
134        &self,
135        workspace: &Workspace,
136        tasks: &[Task],
137        reverse: bool,
138    ) -> Result<Vec<Task>> {
139        let mut conn = self.acquire().await?;
140        let mut tx = begin_immediate(&mut conn).await?;
141        let mut outcomes = Vec::with_capacity(tasks.len());
142        for task in tasks {
143            outcomes.push(cycle_priority(&mut tx, workspace, task, reverse).await?);
144        }
145        tx.commit().await?;
146        Ok(outcomes)
147    }
148}
149
150pub(crate) async fn set_status(
151    conn: &mut SqliteConnection,
152    workspace: &Workspace,
153    task: &Task,
154    status: &str,
155) -> Result<Task> {
156    set_task_field(conn, workspace, &task.id, "status", status).await?;
157    get_task_in_workspace(conn, workspace, &task.id).await
158}
159
160pub(crate) async fn set_priority(
161    conn: &mut SqliteConnection,
162    workspace: &Workspace,
163    task: &Task,
164    priority: &str,
165) -> Result<Task> {
166    set_task_field(conn, workspace, &task.id, "priority", priority).await?;
167    get_task_in_workspace(conn, workspace, &task.id).await
168}
169
170pub(crate) async fn cycle_priority(
171    conn: &mut SqliteConnection,
172    workspace: &Workspace,
173    task: &Task,
174    reverse: bool,
175) -> Result<Task> {
176    let index = TaskPriority::ALL
177        .iter()
178        .position(|priority| *priority == task.priority)
179        .unwrap_or(0);
180    let next = if reverse {
181        (index + TaskPriority::ALL.len() - 1) % TaskPriority::ALL.len()
182    } else {
183        (index + 1) % TaskPriority::ALL.len()
184    };
185    set_priority(conn, workspace, task, TaskPriority::ALL[next].as_str()).await
186}
187
188pub(crate) async fn set_deleted(
189    conn: &mut SqliteConnection,
190    workspace: &Workspace,
191    task: &Task,
192    deleted: bool,
193) -> Result<Task> {
194    set_task_field(
195        conn,
196        workspace,
197        &task.id,
198        "deleted",
199        if deleted { "1" } else { "0" },
200    )
201    .await?;
202    get_task_in_workspace(conn, workspace, &task.id).await
203}
204
205pub(crate) async fn set_task_field(
206    conn: &mut SqliteConnection,
207    workspace: &Workspace,
208    task_id: &crate::ids::TaskId,
209    field: &str,
210    value: &str,
211) -> Result<bool> {
212    if let Some(changed) =
213        crate::operations::route_recurrence_task_field(conn, workspace, task_id, field, value)
214            .await?
215    {
216        return Ok(changed);
217    }
218    let task_field = TaskField::parse_or_unknown(field)?;
219    if task_field.is_project() {
220        let project = resolve_or_create_project_in_workspace(conn, &workspace.id, value).await?;
221        set_task_project(conn, workspace, task_id, &project).await
222    } else {
223        set_task_scalar_field(conn, workspace, task_id, task_field, value).await
224    }
225}
226
227pub(crate) async fn set_task_project(
228    conn: &mut SqliteConnection,
229    workspace: &Workspace,
230    task_id: &crate::ids::TaskId,
231    project: &Project,
232) -> Result<bool> {
233    crate::operations::route_recurrence_task_field(
234        conn,
235        workspace,
236        task_id,
237        "project",
238        project.id.as_str(),
239    )
240    .await?;
241    let field = TaskField::Project.as_str();
242    let current = current_task(conn, &workspace.id, task_id).await?;
243    if current.project_id == project.id {
244        return Ok(false);
245    }
246    if conflict_exists(conn, &workspace.id, task_id, field).await? {
247        return Err(anyhow::Error::new(OpenConflictError {
248            task_id: task_id.clone(),
249            field,
250        }));
251    }
252    debug!(task_id = %task_id, field = %field, "task field mutation started");
253    let base = field_version(conn, task_id, field).await?;
254    apply_project_id_in_workspace(conn, &workspace.id, task_id, &project.id).await?;
255    let payload = TaskField::project_payload(&workspace.id, &workspace.key, project);
256    finish_task_field_change(conn, task_id, field, payload, base.as_deref()).await?;
257    Ok(true)
258}
259
260async fn set_task_scalar_field(
261    conn: &mut SqliteConnection,
262    workspace: &Workspace,
263    task_id: &crate::ids::TaskId,
264    task_field: TaskField,
265    value: &str,
266) -> Result<bool> {
267    task_field.validate_value(value)?;
268
269    let field = task_field.as_str();
270    let current = current_task(conn, &workspace.id, task_id).await?;
271    if task_field.current_value(&current) == value {
272        return Ok(false);
273    }
274    if conflict_exists(conn, &workspace.id, task_id, field).await? {
275        return Err(anyhow::Error::new(OpenConflictError {
276            task_id: task_id.clone(),
277            field,
278        }));
279    }
280    debug!(task_id = %task_id, field = %field, "task field mutation started");
281    let base = field_version(conn, task_id, field).await?;
282    apply_scalar_field_value_in_workspace(conn, &workspace.id, task_id, task_field, value).await?;
283    let payload = task_field.scalar_payload(&workspace.id, &workspace.key, value)?;
284    finish_task_field_change(conn, task_id, field, payload, base.as_deref()).await?;
285    Ok(true)
286}
287
288async fn finish_task_field_change(
289    conn: &mut SqliteConnection,
290    task_id: &crate::ids::TaskId,
291    field: &str,
292    payload: serde_json::Value,
293    base: Option<&str>,
294) -> Result<()> {
295    let change_id = insert_change(
296        conn,
297        "task",
298        task_id,
299        Some(field),
300        op_type::SET_FIELD,
301        payload,
302        base,
303    )
304    .await?;
305    set_field_version(conn, task_id, field, &change_id).await?;
306    info!(
307        task_id = %task_id,
308        field = %field,
309        change_id = %change_id,
310        "task field mutated"
311    );
312    Ok(())
313}
314
315async fn current_task(
316    conn: &mut SqliteConnection,
317    workspace_id: &WorkspaceId,
318    task_id: &crate::ids::TaskId,
319) -> Result<Task> {
320    let row = sqlx::query(
321        "SELECT t.id, t.workspace_id, t.title, t.description, t.project_id,
322                p.key AS project_key, p.prefix AS project_prefix, t.status,
323                t.priority, t.source, t.created_at, t.updated_at, t.queue_activity_at,
324                t.available_at, t.due_on, t.deleted, t.is_epic
325         FROM tasks t
326         JOIN projects p ON p.workspace_id = t.workspace_id AND p.id = t.project_id
327         WHERE t.workspace_id = ? AND t.id = ?",
328    )
329    .bind(workspace_id)
330    .bind(task_id)
331    .fetch_optional(&mut *conn)
332    .await?
333    .ok_or_else(|| {
334        anyhow::anyhow!(
335            "error task-not-found task_id={} workspace_id={}",
336            task_id,
337            workspace_id
338        )
339    })?;
340    task_from_row(&row)
341}
342
343#[allow(dead_code)]
344pub async fn apply_field_value(
345    conn: &mut SqliteConnection,
346    workspace_id: &WorkspaceId,
347    task_id: &crate::ids::TaskId,
348    field: &str,
349    value: &str,
350) -> Result<()> {
351    apply_field_value_in_workspace(conn, workspace_id, task_id, field, value).await
352}
353
354pub async fn apply_project_id_in_workspace(
355    conn: &mut SqliteConnection,
356    workspace_id: &WorkspaceId,
357    task_id: &crate::ids::TaskId,
358    project_id: &ProjectId,
359) -> Result<()> {
360    let project_exists = sqlx::query_scalar::<_, i64>(
361        "SELECT count(*) FROM projects WHERE workspace_id = ? AND id = ? AND deleted = 0",
362    )
363    .bind(workspace_id)
364    .bind(project_id)
365    .fetch_one(&mut *conn)
366    .await?
367        > 0;
368    if !project_exists {
369        bail!("error unknown-project-id id={project_id}");
370    }
371    let ts = now();
372    let rows_affected = sqlx::query(
373        "UPDATE tasks SET project_id = ?, updated_at = ? WHERE workspace_id = ? AND id = ?",
374    )
375    .bind(project_id)
376    .bind(&ts)
377    .bind(workspace_id)
378    .bind(task_id)
379    .execute(&mut *conn)
380    .await?
381    .rows_affected();
382    ensure!(
383        rows_affected == 1,
384        "error task-not-found task_id={} workspace_id={}",
385        task_id,
386        workspace_id
387    );
388    Ok(())
389}
390
391pub async fn apply_field_value_in_workspace(
392    conn: &mut SqliteConnection,
393    workspace_id: &WorkspaceId,
394    task_id: &crate::ids::TaskId,
395    field: &str,
396    value: &str,
397) -> Result<()> {
398    let task_field = TaskField::parse_or_unknown(field)?;
399    apply_scalar_field_value_in_workspace(conn, workspace_id, task_id, task_field, value).await
400}
401
402async fn apply_scalar_field_value_in_workspace(
403    conn: &mut SqliteConnection,
404    workspace_id: &WorkspaceId,
405    task_id: &crate::ids::TaskId,
406    task_field: TaskField,
407    value: &str,
408) -> Result<()> {
409    task_field.validate_value(value)?;
410
411    let ts = now();
412    let activity_at = if task_field.updates_queue_activity() {
413        ts.as_str()
414    } else {
415        ""
416    };
417    let deleted_value = i64::from(value == "1");
418    let epic_value = i64::from(value == "1");
419    let rows_affected = match task_field {
420        TaskField::Title => sqlx::query(
421            "UPDATE tasks SET title = ?, updated_at = ?, queue_activity_at = COALESCE(NULLIF(?, ''), queue_activity_at) WHERE workspace_id = ? AND id = ?",
422        )
423        .bind(value)
424        .bind(&ts)
425        .bind(activity_at)
426        .bind(workspace_id)
427        .bind(task_id)
428        .execute(&mut *conn)
429        .await?
430        .rows_affected(),
431        TaskField::Description => sqlx::query(
432            "UPDATE tasks SET description = ?, updated_at = ?, queue_activity_at = COALESCE(NULLIF(?, ''), queue_activity_at) WHERE workspace_id = ? AND id = ?",
433        )
434        .bind(value)
435        .bind(&ts)
436        .bind(activity_at)
437        .bind(workspace_id)
438        .bind(task_id)
439        .execute(&mut *conn)
440        .await?
441        .rows_affected(),
442        TaskField::Project => bail!("error project-update-requires-project-id"),
443        TaskField::Status => sqlx::query(
444            "UPDATE tasks SET status = ?, updated_at = ?, queue_activity_at = COALESCE(NULLIF(?, ''), queue_activity_at) WHERE workspace_id = ? AND id = ?",
445        )
446        .bind(value)
447        .bind(&ts)
448        .bind(activity_at)
449        .bind(workspace_id)
450        .bind(task_id)
451        .execute(&mut *conn)
452        .await?
453        .rows_affected(),
454        TaskField::Priority => sqlx::query(
455            "UPDATE tasks SET priority = ?, updated_at = ?, queue_activity_at = COALESCE(NULLIF(?, ''), queue_activity_at) WHERE workspace_id = ? AND id = ?",
456        )
457        .bind(value)
458        .bind(&ts)
459        .bind(activity_at)
460        .bind(workspace_id)
461        .bind(task_id)
462        .execute(&mut *conn)
463        .await?
464        .rows_affected(),
465        TaskField::AvailableAt => sqlx::query(
466            "UPDATE tasks SET available_at = ?, updated_at = ?, queue_activity_at = COALESCE(NULLIF(?, ''), queue_activity_at) WHERE workspace_id = ? AND id = ?",
467        )
468        .bind(value)
469        .bind(&ts)
470        .bind(activity_at)
471        .bind(workspace_id)
472        .bind(task_id)
473        .execute(&mut *conn)
474        .await?
475        .rows_affected(),
476        TaskField::DueOn => sqlx::query(
477            "UPDATE tasks SET due_on = ?, updated_at = ? WHERE workspace_id = ? AND id = ?",
478        )
479        .bind(value)
480        .bind(&ts)
481        .bind(workspace_id)
482        .bind(task_id)
483        .execute(&mut *conn)
484        .await?
485        .rows_affected(),
486        TaskField::Deleted => sqlx::query(
487            "UPDATE tasks SET deleted = ?, updated_at = ?, queue_activity_at = COALESCE(NULLIF(?, ''), queue_activity_at) WHERE workspace_id = ? AND id = ?",
488        )
489        .bind(deleted_value)
490        .bind(&ts)
491        .bind(activity_at)
492        .bind(workspace_id)
493        .bind(task_id)
494        .execute(&mut *conn)
495        .await?
496        .rows_affected(),
497        TaskField::IsEpic => sqlx::query(
498            "UPDATE tasks SET is_epic = ?, updated_at = ? WHERE workspace_id = ? AND id = ?",
499        )
500        .bind(epic_value)
501        .bind(&ts)
502        .bind(workspace_id)
503        .bind(task_id)
504        .execute(&mut *conn)
505        .await?
506        .rows_affected(),
507    };
508    ensure!(
509        rows_affected == 1,
510        "error task-not-found task_id={} workspace_id={}",
511        task_id,
512        workspace_id
513    );
514    Ok(())
515}
aven_core/mutation.rs

aven_core/
mutation.rs
