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

aven_core/
mutation.rs
