aven_core/operations/
conflicts.rs

1use anyhow::{Context, Result, bail, ensure};
2use sqlx::{Row, SqliteConnection};
3use tracing::info;
4
5use crate::change_log::op_type;
6use crate::db::{Database, begin_immediate, insert_change, set_field_version};
7use crate::ids::{TaskId, now};
8use crate::mutation::{apply_field_value_in_workspace, apply_project_id_in_workspace};
9use crate::projects::{resolve_existing_project_in_workspace, resolve_project_for_stored_value};
10use crate::recurrence::{RecurrenceOutcome, RecurrenceSeriesId, RecurrenceSeriesState};
11use crate::refs::get_task_in_workspace;
12use crate::task_fields::TaskField;
13use crate::types::Task;
14use crate::undo::{UndoCommand, UndoPayload, record_tui_undo};
15use crate::workspaces::Workspace;
16
17impl Database {
18    pub async fn list_conflicts(
19        &self,
20        workspace: &Workspace,
21        project_key: Option<&str>,
22        field: Option<&str>,
23    ) -> Result<Vec<ConflictListItem>> {
24        let mut conn = self.acquire().await?;
25        list_conflicts(&mut conn, workspace, project_key, field).await
26    }
27
28    pub async fn task_conflicts(
29        &self,
30        workspace: &Workspace,
31        task_id: &TaskId,
32        field: Option<&str>,
33    ) -> Result<Vec<ConflictDetail>> {
34        let mut conn = self.acquire().await?;
35        task_conflicts(&mut conn, workspace, task_id, field).await
36    }
37
38    pub async fn conflict_variant_value(
39        &self,
40        workspace: &Workspace,
41        task_id: &TaskId,
42        field: &str,
43        token: &str,
44    ) -> Result<String> {
45        let mut conn = self.acquire().await?;
46        conflict_variant_value(&mut conn, workspace, task_id, field, token).await
47    }
48
49    pub async fn resolve_conflict_with_tui_undo(
50        &self,
51        workspace: &Workspace,
52        task_id: &TaskId,
53        field: &str,
54        value: &str,
55        summary: &str,
56    ) -> Result<ConflictResolutionOutcome> {
57        let mut conn = self.acquire().await?;
58        resolve_conflict_value(
59            &mut conn,
60            workspace,
61            task_id,
62            field,
63            ResolutionValue::Explicit(value),
64            Some(summary),
65        )
66        .await
67    }
68
69    pub async fn resolve_conflict(
70        &self,
71        workspace: &Workspace,
72        task_id: &TaskId,
73        field: &str,
74        value: &str,
75    ) -> Result<ConflictOutcome> {
76        let mut conn = self.acquire().await?;
77        resolve_conflict(&mut conn, workspace, task_id, field, value).await
78    }
79
80    pub async fn recurrence_series_conflicts(
81        &self,
82        workspace: &Workspace,
83        series_id: &RecurrenceSeriesId,
84        field: Option<&str>,
85    ) -> Result<Vec<ConflictDetail>> {
86        let mut conn = self.acquire().await?;
87        recurrence_series_conflicts(&mut conn, workspace, series_id, field).await
88    }
89
90    pub async fn recurrence_conflict_variant_value(
91        &self,
92        workspace: &Workspace,
93        series_id: &RecurrenceSeriesId,
94        field: &str,
95        token: &str,
96    ) -> Result<String> {
97        let mut conn = self.acquire().await?;
98        for detail in
99            recurrence_series_conflicts(&mut conn, workspace, series_id, Some(field)).await?
100        {
101            if token == detail.variant_a {
102                return Ok(detail.local_value);
103            }
104            if token == detail.variant_b {
105                return Ok(detail.remote_value);
106            }
107        }
108        bail!("error unknown-variant token={token}")
109    }
110
111    pub async fn resolve_recurrence_conflict(
112        &self,
113        workspace: &Workspace,
114        series_id: &RecurrenceSeriesId,
115        field: &str,
116        value: &str,
117    ) -> Result<String> {
118        let mut conn = self.acquire().await?;
119        resolve_recurrence_conflict(&mut conn, workspace, series_id, field, value).await
120    }
121}
122
123pub struct ConflictListItem {
124    pub task_id: TaskId,
125    pub recurrence_series: bool,
126    pub title: String,
127    pub project_key: String,
128    pub project_prefix: String,
129    pub field: String,
130    pub variant_a: String,
131    pub variant_b: String,
132}
133
134pub struct ConflictDetail {
135    pub field: String,
136    pub variant_a: String,
137    pub local_value: String,
138    pub variant_b: String,
139    pub remote_value: String,
140}
141
142pub struct ConflictOutcome {
143    pub task: Task,
144    pub field: String,
145}
146
147pub struct ConflictResolutionOutcome {
148    pub outcome: ConflictOutcome,
149    pub before: String,
150    pub after: String,
151    pub conflict_id: i64,
152}
153pub async fn list_conflicts(
154    conn: &mut SqliteConnection,
155    workspace: &Workspace,
156    project_key: Option<&str>,
157    field: Option<&str>,
158) -> Result<Vec<ConflictListItem>> {
159    let workspace_id = &workspace.id;
160    let project_id = if let Some(project) = project_key {
161        Some(
162            resolve_existing_project_in_workspace(conn, workspace_id, project)
163                .await?
164                .id,
165        )
166    } else {
167        None
168    };
169    let rows = sqlx::query(
170        r#"SELECT c.entity_id AS task_id, c.entity_type, c.field, c.variant_a, c.variant_b,
171                 t.title, p.prefix, p.key AS project_key
172                 FROM conflicts c
173                 JOIN tasks t ON c.entity_type = 'task' AND t.workspace_id = c.workspace_id AND t.id = c.entity_id
174                 JOIN projects p ON p.workspace_id = t.workspace_id AND p.id = t.project_id
175                 WHERE c.workspace_id = ? AND c.resolved = 0
176                 AND (? IS NULL OR t.project_id = ?)
177                 AND (? IS NULL OR c.field = ?)
178                 UNION ALL
179                 SELECT c.entity_id AS task_id, c.entity_type, c.field, c.variant_a, c.variant_b,
180                 s.title, p.prefix, p.key AS project_key
181                 FROM conflicts c
182                 JOIN recurrence_series s ON c.entity_type = 'recurrence_series'
183                    AND s.workspace_id = c.workspace_id AND s.id = c.entity_id
184                 JOIN projects p ON p.workspace_id = s.workspace_id AND p.id = s.project_id
185                 WHERE c.workspace_id = ? AND c.resolved = 0
186                 AND (? IS NULL OR s.project_id = ?)
187                 AND (? IS NULL OR c.field = ?)
188                 ORDER BY field"#,
189    )
190    .bind(workspace_id)
191    .bind(&project_id)
192    .bind(&project_id)
193    .bind(field)
194    .bind(field)
195    .bind(workspace_id)
196    .bind(&project_id)
197    .bind(&project_id)
198    .bind(field)
199    .bind(field)
200    .fetch_all(&mut *conn)
201    .await?;
202    Ok(rows
203        .into_iter()
204        .map(|row| ConflictListItem {
205            task_id: row.get("task_id"),
206            recurrence_series: row.get::<String, _>("entity_type") == "recurrence_series",
207            title: row.get("title"),
208            project_key: row.get("project_key"),
209            project_prefix: row.get("prefix"),
210            field: row.get("field"),
211            variant_a: row.get("variant_a"),
212            variant_b: row.get("variant_b"),
213        })
214        .collect())
215}
216
217pub async fn task_conflicts(
218    conn: &mut SqliteConnection,
219    workspace: &Workspace,
220    task_id: &crate::ids::TaskId,
221    field: Option<&str>,
222) -> Result<Vec<ConflictDetail>> {
223    let workspace_id = &workspace.id;
224    let rows = sqlx::query(
225        r#"SELECT field, variant_a, local_value, variant_b, remote_value
226         FROM conflicts
227         WHERE workspace_id = ? AND task_id = ? AND resolved = 0 AND (? IS NULL OR field = ?)
228         ORDER BY field, id"#,
229    )
230    .bind(workspace_id)
231    .bind(task_id)
232    .bind(field)
233    .bind(field)
234    .fetch_all(&mut *conn)
235    .await?;
236    Ok(rows
237        .into_iter()
238        .map(|row| ConflictDetail {
239            field: row.get("field"),
240            variant_a: row.get("variant_a"),
241            local_value: row.get("local_value"),
242            variant_b: row.get("variant_b"),
243            remote_value: row.get("remote_value"),
244        })
245        .collect())
246}
247
248pub async fn recurrence_series_conflicts(
249    conn: &mut SqliteConnection,
250    workspace: &Workspace,
251    series_id: &RecurrenceSeriesId,
252    field: Option<&str>,
253) -> Result<Vec<ConflictDetail>> {
254    let rows = sqlx::query(
255        "SELECT field, variant_a, local_value, variant_b, remote_value FROM conflicts
256         WHERE workspace_id = ? AND entity_type = 'recurrence_series' AND entity_id = ?
257           AND resolved = 0 AND (? IS NULL OR field = ?) ORDER BY field, id",
258    )
259    .bind(&workspace.id)
260    .bind(series_id)
261    .bind(field)
262    .bind(field)
263    .fetch_all(&mut *conn)
264    .await?;
265    Ok(rows
266        .into_iter()
267        .map(|row| ConflictDetail {
268            field: row.get("field"),
269            variant_a: row.get("variant_a"),
270            local_value: row.get("local_value"),
271            variant_b: row.get("variant_b"),
272            remote_value: row.get("remote_value"),
273        })
274        .collect())
275}
276
277async fn resolve_recurrence_conflict(
278    conn: &mut SqliteConnection,
279    workspace: &Workspace,
280    series_id: &RecurrenceSeriesId,
281    field: &str,
282    value: &str,
283) -> Result<String> {
284    let mut tx = begin_immediate(conn).await?;
285    let conflict = sqlx::query(
286        "SELECT id, local_value, remote_value, remote_change_id FROM conflicts
287         WHERE workspace_id = ? AND entity_type = 'recurrence_series'
288           AND entity_id = ? AND field = ? AND resolved = 0",
289    )
290    .bind(&workspace.id)
291    .bind(series_id)
292    .bind(field)
293    .fetch_optional(&mut *tx)
294    .await?
295    .ok_or_else(|| {
296        anyhow::anyhow!("error conflict-not-found series_id={series_id} field={field}")
297    })?;
298    let local: String = conflict.get("local_value");
299    let remote: String = conflict.get("remote_value");
300    if value != local && value != remote {
301        bail!("error invalid-conflict-value value={value}");
302    }
303    if let Some(slot) = field.strip_prefix("outcome:") {
304        resolve_recurrence_outcome_conflict(
305            &mut tx,
306            workspace,
307            series_id,
308            slot.parse()?,
309            RecurrenceOutcome::parse(value)?,
310        )
311        .await?;
312    } else if field == "state" {
313        resolve_recurrence_state_conflict(
314            &mut tx,
315            workspace,
316            series_id,
317            RecurrenceSeriesState::parse(value)?,
318            &conflict.get::<String, _>("remote_change_id"),
319        )
320        .await?;
321    } else if matches!(
322        field,
323        "title"
324            | "description"
325            | "project"
326            | "priority"
327            | "initial_status"
328            | "available_local_time"
329            | "due_policy"
330            | "labels"
331    ) {
332        resolve_recurrence_template_conflict(&mut tx, workspace, series_id, field, value).await?;
333    }
334    sqlx::query("UPDATE conflicts SET resolved = 1 WHERE id = ?")
335        .bind(conflict.get::<i64, _>("id"))
336        .execute(&mut *tx)
337        .await?;
338    tx.commit().await?;
339    Ok(field.to_string())
340}
341
342async fn resolve_recurrence_template_conflict(
343    conn: &mut SqliteConnection,
344    workspace: &Workspace,
345    series_id: &RecurrenceSeriesId,
346    field: &str,
347    value: &str,
348) -> Result<()> {
349    let updated_at = now();
350    let (fields, labels, labels_changed) = if field == "labels" {
351        let labels: Vec<String> = serde_json::from_str(value)?;
352        sqlx::query(
353            "DELETE FROM recurrence_series_labels WHERE workspace_id = ? AND series_id = ?",
354        )
355        .bind(&workspace.id)
356        .bind(series_id)
357        .execute(&mut *conn)
358        .await?;
359        for label in &labels {
360            sqlx::query(
361                "INSERT OR IGNORE INTO labels(workspace_id, name, created_at) VALUES (?, ?, ?)",
362            )
363            .bind(&workspace.id)
364            .bind(label)
365            .bind(&updated_at)
366            .execute(&mut *conn)
367            .await?;
368            sqlx::query("INSERT INTO recurrence_series_labels(workspace_id, series_id, label) VALUES (?, ?, ?)")
369                .bind(&workspace.id).bind(series_id).bind(label).execute(&mut *conn).await?;
370        }
371        (Vec::<(String, String)>::new(), labels, true)
372    } else {
373        let query = match field {
374            "title" => {
375                "UPDATE recurrence_series SET title = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
376            }
377            "description" => {
378                "UPDATE recurrence_series SET description = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
379            }
380            "project" => {
381                "UPDATE recurrence_series SET project_id = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
382            }
383            "priority" => {
384                "UPDATE recurrence_series SET priority = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
385            }
386            "initial_status" => {
387                "UPDATE recurrence_series SET initial_status = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
388            }
389            "available_local_time" => {
390                "UPDATE recurrence_series SET available_local_time = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
391            }
392            "due_policy" => {
393                "UPDATE recurrence_series SET due_policy = ?, updated_at = ? WHERE workspace_id = ? AND id = ?"
394            }
395            _ => unreachable!(),
396        };
397        sqlx::query(query)
398            .bind(value)
399            .bind(&updated_at)
400            .bind(&workspace.id)
401            .bind(series_id)
402            .execute(&mut *conn)
403            .await?;
404        let labels = sqlx::query_scalar("SELECT label FROM recurrence_series_labels WHERE workspace_id = ? AND series_id = ? ORDER BY label")
405            .bind(&workspace.id).bind(series_id).fetch_all(&mut *conn).await?;
406        (vec![(field.to_string(), value.to_string())], labels, false)
407    };
408    let mut base_versions = serde_json::Map::new();
409    base_versions.insert(field.to_string(), serde_json::Value::Null);
410    let change_id = insert_change(
411        conn,
412        "recurrence_series",
413        series_id.as_str(),
414        None,
415        op_type::UPDATE_RECURRENCE_TEMPLATE,
416        crate::change_log::ChangePayload::workspace(workspace)
417            .set("fields", fields)
418            .set("base_versions", base_versions)
419            .set("labels_changed", labels_changed)
420            .set("labels", labels)
421            .set("updated_at", &updated_at)
422            .set("conflict_resolution", true)
423            .into_value(),
424        None,
425    )
426    .await?;
427    crate::db::set_entity_field_version(
428        conn,
429        &workspace.id,
430        crate::types::MutableEntityType::RecurrenceSeries,
431        series_id.as_str(),
432        field,
433        &change_id,
434    )
435    .await?;
436    Ok(())
437}
438
439async fn resolve_recurrence_outcome_conflict(
440    conn: &mut SqliteConnection,
441    workspace: &Workspace,
442    series_id: &RecurrenceSeriesId,
443    slot_on: chrono::NaiveDate,
444    outcome: RecurrenceOutcome,
445) -> Result<()> {
446    let row = sqlx::query("SELECT task_id, resolved_at FROM recurrence_occurrences WHERE workspace_id = ? AND series_id = ? AND slot_on = ?")
447        .bind(&workspace.id).bind(series_id).bind(slot_on.to_string()).fetch_one(&mut *conn).await?;
448    let task_id: String = row.get("task_id");
449    let resolved_at: String = row.get("resolved_at");
450    let series_row = sqlx::query("SELECT workspace_id, id, title, description, project_id, priority, initial_status, frequency, interval, weekdays, timezone, start_on, available_local_time, due_policy, state, stopped_at, created_at, updated_at, deleted FROM recurrence_series WHERE workspace_id = ? AND id = ?")
451        .bind(&workspace.id).bind(series_id).fetch_one(&mut *conn).await?;
452    let series = crate::db::recurrence_series_from_row(&series_row)?;
453    ensure!(
454        !task_id.is_empty(),
455        "error recurrence-outcome-task-missing slot={slot_on}"
456    );
457    let status = if outcome == RecurrenceOutcome::Completed {
458        "done"
459    } else {
460        "canceled"
461    };
462    apply_field_value_in_workspace(conn, &workspace.id, &task_id.parse()?, "status", status)
463        .await?;
464    let change_id = insert_change(
465        conn,
466        "recurrence_series",
467        series_id.as_str(),
468        Some("outcome"),
469        op_type::RESOLVE_RECURRENCE_OCCURRENCE,
470        crate::change_log::ChangePayload::workspace(workspace)
471            .set("slot_on", slot_on.to_string())
472            .set("task_id", &task_id)
473            .set("outcome", outcome.as_str())
474            .set(
475                "task_status",
476                if outcome == RecurrenceOutcome::Completed {
477                    "done"
478                } else {
479                    "canceled"
480                },
481            )
482            .set("resolved_at", &resolved_at)
483            .set("task_status_change_id", "")
484            .set("successor_task_id", "")
485            .set("frequency", series.rule.frequency().as_str())
486            .set("interval", series.rule.interval())
487            .set("weekdays", series.rule.weekdays_set().to_string())
488            .set("timezone", series.timezone.as_str())
489            .set("start_on", series.start_on.to_string())
490            .set(
491                "available_local_time",
492                series
493                    .available_local_time
494                    .map(|time| time.format("%H:%M:%S").to_string())
495                    .unwrap_or_default(),
496            )
497            .set("due_policy", series.due_policy.as_str())
498            .set("conflict_resolution", true)
499            .into_value(),
500        None,
501    )
502    .await?;
503    sqlx::query("UPDATE recurrence_occurrences SET outcome = ?, outcome_change_id = ? WHERE workspace_id = ? AND series_id = ? AND slot_on = ?")
504        .bind(outcome.as_str()).bind(change_id).bind(&workspace.id).bind(series_id).bind(slot_on.to_string()).execute(&mut *conn).await?;
505    Ok(())
506}
507
508async fn resolve_recurrence_state_conflict(
509    conn: &mut SqliteConnection,
510    workspace: &Workspace,
511    series_id: &RecurrenceSeriesId,
512    state: RecurrenceSeriesState,
513    remote_change_id: &str,
514) -> Result<()> {
515    let local = sqlx::query(
516        "SELECT state, stopped_at FROM recurrence_series WHERE workspace_id = ? AND id = ?",
517    )
518    .bind(&workspace.id)
519    .bind(series_id)
520    .fetch_one(&mut *conn)
521    .await?;
522    let local_state: String = local.get("state");
523    let remote_payload: Option<String> =
524        sqlx::query_scalar("SELECT payload FROM changes WHERE change_id = ?")
525            .bind(remote_change_id)
526            .fetch_optional(&mut *conn)
527            .await?;
528    let remote_payload = remote_payload
529        .as_deref()
530        .map(serde_json::from_str::<serde_json::Value>)
531        .transpose()?;
532    let stopped_at = if state == RecurrenceSeriesState::Stopped {
533        if local_state == "stopped" {
534            local.get("stopped_at")
535        } else {
536            remote_payload
537                .as_ref()
538                .and_then(|payload| payload.get("stopped_at"))
539                .and_then(serde_json::Value::as_str)
540                .context("error recurrence-stop-boundary-missing")?
541                .to_string()
542        }
543    } else {
544        String::new()
545    };
546    let mut changed_at = now();
547    if state == RecurrenceSeriesState::Active {
548        let pause_boundaries: Vec<String> = sqlx::query_scalar(
549            "SELECT json_extract(payload, '$.paused_at') FROM changes
550             WHERE entity_type = 'recurrence_series' AND entity_id = ?
551               AND op_type = 'open_recurrence_pause'
552               AND json_extract(payload, '$.workspace_id') = ?",
553        )
554        .bind(series_id)
555        .bind(workspace.id.as_str())
556        .fetch_all(&mut *conn)
557        .await?;
558        for paused_at in pause_boundaries {
559            changed_at = super::recurrence::timestamp_strictly_after(&paused_at, &changed_at)?;
560        }
561    }
562    sqlx::query("UPDATE recurrence_series SET state = ?, stopped_at = ?, updated_at = ? WHERE workspace_id = ? AND id = ?")
563        .bind(state.as_str()).bind(&stopped_at).bind(&changed_at).bind(&workspace.id).bind(series_id).execute(&mut *conn).await?;
564    let operation = if state == RecurrenceSeriesState::Stopped {
565        op_type::STOP_RECURRENCE_SERIES
566    } else {
567        op_type::SET_RECURRENCE_STATE
568    };
569    let change_id = insert_change(
570        conn,
571        "recurrence_series",
572        series_id.as_str(),
573        Some("state"),
574        operation,
575        crate::change_log::ChangePayload::workspace(workspace)
576            .set("state", state.as_str())
577            .set("stopped_at", &stopped_at)
578            .set("changed_at", &changed_at)
579            .set("conflict_resolution", true)
580            .into_value(),
581        None,
582    )
583    .await?;
584    crate::db::set_entity_field_version(
585        conn,
586        &workspace.id,
587        crate::types::MutableEntityType::RecurrenceSeries,
588        series_id.as_str(),
589        "state",
590        &change_id,
591    )
592    .await?;
593    sqlx::query(
594        "UPDATE conflicts SET resolved = 1
595         WHERE workspace_id = ? AND entity_type = 'recurrence_series'
596           AND entity_id = ? AND field = 'state' AND resolved = 0",
597    )
598    .bind(&workspace.id)
599    .bind(series_id)
600    .execute(&mut *conn)
601    .await?;
602    if state == RecurrenceSeriesState::Active {
603        sqlx::query(
604            "UPDATE recurrence_pause_intervals
605             SET resumed_at = ?, resolved_by_change_id = ?
606             WHERE workspace_id = ? AND series_id = ? AND resumed_at = ''",
607        )
608        .bind(&changed_at)
609        .bind(&change_id)
610        .bind(&workspace.id)
611        .bind(series_id)
612        .execute(&mut *conn)
613        .await?;
614    }
615    cleanup_recurrence_projections(conn, workspace, series_id, state, &stopped_at).await?;
616    if state == RecurrenceSeriesState::Active {
617        super::recurrence::reconcile_recurrence_series_in_transaction(
618            conn,
619            workspace,
620            series_id,
621            chrono::DateTime::parse_from_rfc3339(&changed_at)?.with_timezone(&chrono::Utc),
622        )
623        .await?;
624    }
625    Ok(())
626}
627
628pub(crate) async fn cleanup_recurrence_projections(
629    conn: &mut SqliteConnection,
630    workspace: &Workspace,
631    series_id: &RecurrenceSeriesId,
632    state: RecurrenceSeriesState,
633    stopped_at: &str,
634) -> Result<()> {
635    let archived_at = now();
636    if state == RecurrenceSeriesState::Paused {
637        let paused_at: String = sqlx::query_scalar(
638            "SELECT paused_at FROM recurrence_pause_intervals
639             WHERE workspace_id = ? AND series_id = ? AND resumed_at = ''
640             ORDER BY paused_at DESC LIMIT 1",
641        )
642        .bind(&workspace.id)
643        .bind(series_id)
644        .fetch_one(&mut *conn)
645        .await?;
646        archive_projected_occurrences_at_or_after(
647            conn,
648            workspace,
649            series_id,
650            &paused_at,
651            &archived_at,
652        )
653        .await?;
654    } else if state == RecurrenceSeriesState::Stopped {
655        archive_projected_occurrences_at_or_after(
656            conn,
657            workspace,
658            series_id,
659            stopped_at,
660            &archived_at,
661        )
662        .await?;
663    }
664    Ok(())
665}
666
667async fn archive_projected_occurrences_at_or_after(
668    conn: &mut SqliteConnection,
669    workspace: &Workspace,
670    series_id: &RecurrenceSeriesId,
671    boundary: &str,
672    archived_at: &str,
673) -> Result<()> {
674    let row = sqlx::query("SELECT workspace_id, id, title, description, project_id, priority, initial_status, frequency, interval, weekdays, timezone, start_on, available_local_time, due_policy, state, stopped_at, created_at, updated_at, deleted FROM recurrence_series WHERE workspace_id = ? AND id = ?")
675        .bind(&workspace.id).bind(series_id).fetch_one(&mut *conn).await?;
676    let series = crate::db::recurrence_series_from_row(&row)?;
677    let schedule = crate::recurrence::RecurrenceSchedule::new(
678        series.rule,
679        series.timezone,
680        series.start_on,
681        series.available_local_time,
682        series.due_policy,
683    );
684    let rows = sqlx::query("SELECT slot_on FROM recurrence_occurrences WHERE workspace_id = ? AND series_id = ? AND projection_state = 'projected' AND outcome = ''")
685        .bind(&workspace.id).bind(series_id).fetch_all(&mut *conn).await?;
686    for row in rows {
687        let slot: chrono::NaiveDate = row.get::<String, _>("slot_on").parse()?;
688        if crate::recurrence::slot_values(&schedule, slot)?
689            .boundary_at
690            .as_str()
691            >= boundary
692        {
693            sqlx::query("UPDATE recurrence_occurrences SET projection_state = 'archived', archived_at = ? WHERE workspace_id = ? AND series_id = ? AND slot_on = ? AND outcome = ''")
694                .bind(archived_at).bind(&workspace.id).bind(series_id).bind(slot.to_string()).execute(&mut *conn).await?;
695        }
696    }
697    Ok(())
698}
699
700pub async fn conflict_variant_value(
701    conn: &mut SqliteConnection,
702    workspace: &Workspace,
703    task_id: &crate::ids::TaskId,
704    field: &str,
705    token: &str,
706) -> Result<String> {
707    for detail in task_conflicts(conn, workspace, task_id, Some(field)).await? {
708        if token == detail.variant_a {
709            return Ok(detail.local_value);
710        }
711        if token == detail.variant_b {
712            return Ok(detail.remote_value);
713        }
714    }
715    bail!("error unknown-variant token={token}")
716}
717
718pub(crate) enum ConflictValueChoice {
719    Local,
720    Remote,
721}
722
723#[derive(Debug)]
724pub(crate) struct ConflictNotFoundError {
725    task_id: TaskId,
726    field: &'static str,
727}
728
729impl std::fmt::Display for ConflictNotFoundError {
730    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
731        write!(
732            formatter,
733            "error conflict-not-found task_id={} field={}",
734            self.task_id, self.field
735        )
736    }
737}
738
739impl std::error::Error for ConflictNotFoundError {}
740
741pub async fn resolve_conflict(
742    conn: &mut SqliteConnection,
743    workspace: &Workspace,
744    task_id: &crate::ids::TaskId,
745    field: &str,
746    value: &str,
747) -> Result<ConflictOutcome> {
748    Ok(resolve_conflict_value(
749        conn,
750        workspace,
751        task_id,
752        field,
753        ResolutionValue::Explicit(value),
754        None,
755    )
756    .await?
757    .outcome)
758}
759
760pub(crate) async fn resolve_conflict_choice(
761    conn: &mut SqliteConnection,
762    workspace: &Workspace,
763    task_id: &crate::ids::TaskId,
764    field: &str,
765    choice: ConflictValueChoice,
766) -> Result<ConflictOutcome> {
767    Ok(resolve_conflict_value(
768        conn,
769        workspace,
770        task_id,
771        field,
772        ResolutionValue::Choice(choice),
773        None,
774    )
775    .await?
776    .outcome)
777}
778
779enum ResolutionValue<'a> {
780    Explicit(&'a str),
781    Choice(ConflictValueChoice),
782}
783
784async fn resolve_conflict_value(
785    conn: &mut SqliteConnection,
786    workspace: &Workspace,
787    task_id: &crate::ids::TaskId,
788    field: &str,
789    resolution: ResolutionValue<'_>,
790    tui_summary: Option<&str>,
791) -> Result<ConflictResolutionOutcome> {
792    let task_field = TaskField::parse_or_unknown(field)?;
793    let field = task_field.as_str();
794    let mut tx = begin_immediate(conn).await?;
795    let before = crate::undo::task_field_value(&mut tx, &workspace.id, task_id, field).await?;
796    let conflict_id = crate::undo::conflict_row_id(&mut tx, &workspace.id, task_id, field).await?;
797    let value = match resolution {
798        ResolutionValue::Explicit(value) => value.to_string(),
799        ResolutionValue::Choice(choice) => {
800            let values = sqlx::query_as::<_, (String, String)>(
801                "SELECT local_value, remote_value FROM conflicts
802                 WHERE workspace_id = ? AND task_id = ? AND field = ? AND resolved = 0",
803            )
804            .bind(&workspace.id)
805            .bind(task_id)
806            .bind(field)
807            .fetch_optional(&mut *tx)
808            .await?
809            .ok_or_else(|| {
810                anyhow::Error::new(ConflictNotFoundError {
811                    task_id: task_id.clone(),
812                    field,
813                })
814            })?;
815            match choice {
816                ConflictValueChoice::Local => values.0,
817                ConflictValueChoice::Remote => values.1,
818            }
819        }
820    };
821    if task_field == TaskField::IsEpic
822        && value == "0"
823        && crate::operations::task_has_epic_children(&mut tx, &workspace.id, task_id).await?
824    {
825        bail!("error epic-has-children task_id={task_id}");
826    }
827    let result = sqlx::query(
828        "UPDATE conflicts SET resolved = 1 WHERE workspace_id = ? AND task_id = ? AND field = ? AND resolved = 0",
829    )
830    .bind(&workspace.id)
831    .bind(task_id)
832    .bind(field)
833    .execute(&mut *tx)
834    .await?;
835    if result.rows_affected() != 1 {
836        return Err(anyhow::Error::new(ConflictNotFoundError {
837            task_id: task_id.clone(),
838            field,
839        }));
840    }
841    let payload = if task_field.is_project() {
842        let project = resolve_project_for_stored_value(&mut tx, &workspace.id, &value).await?;
843        apply_project_id_in_workspace(&mut tx, &workspace.id, task_id, &project.id).await?;
844        TaskField::project_payload(&workspace.id, &workspace.key, &project)
845    } else {
846        apply_field_value_in_workspace(&mut tx, &workspace.id, task_id, field, &value).await?;
847        task_field.scalar_payload(&workspace.id, &workspace.key, &value)?
848    };
849    let change_id = insert_change(
850        &mut tx,
851        "task",
852        task_id,
853        Some(field),
854        op_type::RESOLVE_FIELD,
855        payload,
856        None,
857    )
858    .await?;
859    set_field_version(&mut tx, task_id, field, &change_id).await?;
860    let task = get_task_in_workspace(&mut tx, workspace, task_id).await?;
861    let after = crate::undo::task_field_value(&mut tx, &workspace.id, task_id, field).await?;
862    if let Some(summary) = tui_summary {
863        record_tui_undo(
864            &mut tx,
865            &workspace.id,
866            summary,
867            UndoPayload {
868                commands: vec![UndoCommand::RestoreConflictResolution {
869                    task_id: task_id.clone(),
870                    field: field.to_string(),
871                    before: before.clone(),
872                    after: after.clone(),
873                    conflict_id,
874                }],
875            },
876        )
877        .await?;
878    }
879    tx.commit().await?;
880    info!(task_id = %task_id, field = %field, "conflict resolved");
881    Ok(ConflictResolutionOutcome {
882        outcome: ConflictOutcome {
883            task,
884            field: field.to_string(),
885        },
886        before,
887        after,
888        conflict_id,
889    })
890}
aven_core/operations/conflicts.rs

aven_core/operations/
conflicts.rs
