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