aven_core/
mutation.rs

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

aven_core/
mutation.rs
