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