aven_core/
mutation.rs

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

aven_core/
mutation.rs
