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