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