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