1use std::collections::{HashMap, HashSet};
2
3use anyhow::{Result, bail};
4use sqlx::SqliteConnection;
5#[cfg(test)]
6use sqlx::{QueryBuilder, Sqlite};
7
8use crate::db::Database;
9use crate::ids::{TaskId, WorkspaceId};
10use crate::recurrence::RecurrenceSeriesId;
11use crate::types::Task;
12use crate::workspaces::Workspace;
13
14mod metadata;
15mod reads;
16mod recurrence;
17mod task;
18
19pub(crate) use recurrence::cleanup_recurrence_projections;
20use recurrence::resolve_recurrence_conflict;
21use task::resolve_conflict_value;
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}
169
170pub async fn list_conflicts(
171 conn: &mut SqliteConnection,
172 workspace: &Workspace,
173 project_key: Option<&str>,
174 field: Option<&str>,
175) -> Result<Vec<ConflictListItem>> {
176 reads::list_conflicts(conn, workspace, project_key, field).await
177}
178
179pub async fn task_conflicts(
180 conn: &mut SqliteConnection,
181 workspace: &Workspace,
182 task_id: &TaskId,
183 field: Option<&str>,
184) -> Result<Vec<ConflictDetail>> {
185 reads::task_conflicts(conn, workspace, task_id, field).await
186}
187
188#[cfg(test)]
189const CONFLICT_CANDIDATE_CHUNK_SIZE: usize = reads::CONFLICT_CANDIDATE_CHUNK_SIZE;
190
191async fn unresolved_task_conflict_fields(
192 conn: &mut SqliteConnection,
193 workspace_id: &WorkspaceId,
194 candidates: &[(TaskId, String)],
195) -> Result<HashMap<TaskId, HashSet<String>>> {
196 reads::unresolved_task_conflict_fields(conn, workspace_id, candidates).await
197}
198
199#[cfg(test)]
200fn unresolved_task_conflict_query(
201 workspace_id: &WorkspaceId,
202 candidates: &[&(TaskId, String)],
203 explain: bool,
204) -> QueryBuilder<Sqlite> {
205 reads::unresolved_task_conflict_query(workspace_id, candidates, explain)
206}
207
208pub async fn recurrence_series_conflicts(
209 conn: &mut SqliteConnection,
210 workspace: &Workspace,
211 series_id: &RecurrenceSeriesId,
212 field: Option<&str>,
213) -> Result<Vec<ConflictDetail>> {
214 reads::recurrence_series_conflicts(conn, workspace, series_id, field).await
215}
216
217pub async fn conflict_variant_value(
218 conn: &mut SqliteConnection,
219 workspace: &Workspace,
220 task_id: &TaskId,
221 field: &str,
222 token: &str,
223) -> Result<String> {
224 reads::conflict_variant_value(conn, workspace, task_id, field, token).await
225}
226
227pub(crate) enum ConflictResolutionValue<'a> {
228 Local,
229 Remote,
230 Explicit(&'a str),
231}
232
233pub(crate) struct ExpectedConflictIdentity<'a> {
234 pub variant_a: &'a str,
235 pub variant_b: &'a str,
236}
237
238pub async fn resolve_conflict(
239 conn: &mut SqliteConnection,
240 workspace: &Workspace,
241 task_id: &crate::ids::TaskId,
242 field: &str,
243 value: &str,
244) -> Result<ConflictOutcome> {
245 Ok(resolve_conflict_value(
246 conn,
247 workspace,
248 task_id,
249 field,
250 ConflictResolutionValue::Explicit(value),
251 None,
252 None,
253 )
254 .await?
255 .outcome)
256}
257
258pub(crate) async fn resolve_conflict_transaction(
259 conn: &mut SqliteConnection,
260 workspace: &Workspace,
261 task_id: &crate::ids::TaskId,
262 field: &str,
263 expected: ExpectedConflictIdentity<'_>,
264 resolution: ConflictResolutionValue<'_>,
265) -> Result<ConflictOutcome> {
266 Ok(resolve_conflict_value(
267 conn,
268 workspace,
269 task_id,
270 field,
271 resolution,
272 Some(expected),
273 None,
274 )
275 .await?
276 .outcome)
277}
278
279#[cfg(test)]
280mod tests {
281 use sqlx::Row;
282
283 use super::{
284 CONFLICT_CANDIDATE_CHUNK_SIZE, TaskId, WorkspaceId, unresolved_task_conflict_fields,
285 unresolved_task_conflict_query,
286 };
287
288 async fn insert_task_conflict(
289 conn: &mut sqlx::SqliteConnection,
290 workspace_id: &WorkspaceId,
291 task_id: &TaskId,
292 field: &str,
293 remote_change_id: &str,
294 resolved: bool,
295 ) {
296 sqlx::query(
297 "INSERT INTO conflicts(
298 workspace_id, entity_type, entity_id, task_id, field,
299 local_value, remote_value, remote_change_id, variant_a,
300 variant_b, created_at, resolved
301 ) VALUES (?, 'task', ?, ?, ?, 'local', 'remote', ?, 'a', 'b', 't', ?)",
302 )
303 .bind(workspace_id)
304 .bind(task_id)
305 .bind(task_id)
306 .bind(field)
307 .bind(remote_change_id)
308 .bind(resolved)
309 .execute(conn)
310 .await
311 .unwrap();
312 }
313
314 #[tokio::test]
315 async fn candidate_conflicts_filter_by_field_and_workspace() {
316 let (_temp, mut conn) = crate::test_support::test_conn().await;
317 let workspace_id: WorkspaceId = "0000000000000000".parse().unwrap();
318 let other_workspace_id: WorkspaceId = "0000000000000001".parse().unwrap();
319 let task_id: TaskId = "0000000000001000".parse().unwrap();
320
321 insert_task_conflict(
322 &mut conn,
323 &workspace_id,
324 &task_id,
325 "priority",
326 "priority-a",
327 false,
328 )
329 .await;
330 insert_task_conflict(
331 &mut conn,
332 &workspace_id,
333 &task_id,
334 "labels",
335 "resolved-labels",
336 true,
337 )
338 .await;
339 insert_task_conflict(
340 &mut conn,
341 &other_workspace_id,
342 &task_id,
343 "status",
344 "other-workspace",
345 false,
346 )
347 .await;
348 sqlx::query(
349 "WITH RECURSIVE seq(i) AS (
350 VALUES(0) UNION ALL SELECT i + 1 FROM seq WHERE i < 1999
351 )
352 INSERT INTO conflicts(
353 workspace_id, entity_type, entity_id, task_id, field,
354 local_value, remote_value, remote_change_id, variant_a,
355 variant_b, created_at
356 )
357 SELECT ?, 'task', ?, ?, printf('metadata:irrelevant-%04d', i),
358 'local', 'remote', printf('irrelevant-%04d', i), 'a', 'b', 't'
359 FROM seq",
360 )
361 .bind(&workspace_id)
362 .bind(&task_id)
363 .bind(&task_id)
364 .execute(&mut *conn)
365 .await
366 .unwrap();
367
368 let candidates = vec![
369 (task_id.clone(), "priority".to_string()),
370 (task_id.clone(), "labels".to_string()),
371 (task_id.clone(), "status".to_string()),
372 ];
373 let conflicts = unresolved_task_conflict_fields(&mut conn, &workspace_id, &candidates)
374 .await
375 .unwrap();
376
377 assert_eq!(conflicts.len(), 1);
378 assert_eq!(
379 conflicts.get(&task_id).unwrap(),
380 &std::collections::HashSet::from(["priority".to_string()])
381 );
382 }
383
384 #[tokio::test]
385 async fn candidate_conflicts_chunk_below_the_sqlite_bind_limit() {
386 let (_temp, mut conn) = crate::test_support::test_conn().await;
387 let workspace_id: WorkspaceId = "0000000000000000".parse().unwrap();
388 let candidates = (0..=CONFLICT_CANDIDATE_CHUNK_SIZE)
389 .map(|index| {
390 (
391 format!("{index:016X}").parse::<TaskId>().unwrap(),
392 "priority".to_string(),
393 )
394 })
395 .collect::<Vec<_>>();
396 for index in [0, candidates.len() - 1] {
397 insert_task_conflict(
398 &mut conn,
399 &workspace_id,
400 &candidates[index].0,
401 "priority",
402 &format!("chunk-{index}"),
403 false,
404 )
405 .await;
406 }
407
408 let conflicts = unresolved_task_conflict_fields(&mut conn, &workspace_id, &candidates)
409 .await
410 .unwrap();
411
412 assert_eq!(conflicts.len(), 2);
413 assert!(conflicts.contains_key(&candidates[0].0));
414 assert!(conflicts.contains_key(&candidates.last().unwrap().0));
415 }
416
417 #[tokio::test]
418 async fn candidate_conflict_plan_uses_exact_field_identity_lookups() {
419 let (_temp, mut conn) = crate::test_support::test_conn().await;
420 let workspace_id: WorkspaceId = "0000000000000000".parse().unwrap();
421 let candidates = (0..64)
422 .map(|index| {
423 (
424 format!("{index:016X}").parse::<TaskId>().unwrap(),
425 "priority".to_string(),
426 )
427 })
428 .collect::<Vec<_>>();
429 let candidate_refs = candidates.iter().collect::<Vec<_>>();
430 let mut query = unresolved_task_conflict_query(&workspace_id, &candidate_refs, true);
431 let plan = query
432 .build()
433 .fetch_all(&mut *conn)
434 .await
435 .unwrap()
436 .into_iter()
437 .map(|row| row.get::<String, _>("detail"))
438 .collect::<Vec<_>>()
439 .join("\n");
440
441 assert!(
442 plan.contains("sqlite_autoindex_conflicts_1")
443 && plan.contains("workspace_id=?")
444 && plan.contains("entity_type=?")
445 && plan.contains("entity_id=?")
446 && plan.contains("field=?"),
447 "unexpected candidate conflict plan:\n{plan}"
448 );
449 }
450}