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

aven_core/
mutation.rs
