1use anyhow::{Result, bail};
2use sqlx::{Row, SqliteConnection};
3use tracing::info;
4
5use crate::change_log::op_type;
6use crate::db::{Database, begin_immediate, insert_change, set_field_version};
7use crate::ids::TaskId;
8use crate::mutation::{apply_field_value_in_workspace, apply_project_id_in_workspace};
9use crate::projects::{resolve_existing_project_in_workspace, resolve_project_for_stored_value};
10use crate::refs::get_task_in_workspace;
11use crate::task_fields::TaskField;
12use crate::types::Task;
13use crate::workspaces::Workspace;
14
15impl Database {
16 pub async fn list_conflicts(
17 &self,
18 workspace: &Workspace,
19 project_key: Option<&str>,
20 field: Option<&str>,
21 ) -> Result<Vec<ConflictListItem>> {
22 let mut conn = self.acquire().await?;
23 list_conflicts(&mut conn, workspace, project_key, field).await
24 }
25
26 pub async fn task_conflicts(
27 &self,
28 workspace: &Workspace,
29 task_id: &TaskId,
30 field: Option<&str>,
31 ) -> Result<Vec<ConflictDetail>> {
32 let mut conn = self.acquire().await?;
33 task_conflicts(&mut conn, workspace, task_id, field).await
34 }
35
36 pub async fn conflict_variant_value(
37 &self,
38 workspace: &Workspace,
39 task_id: &TaskId,
40 field: &str,
41 token: &str,
42 ) -> Result<String> {
43 let mut conn = self.acquire().await?;
44 conflict_variant_value(&mut conn, workspace, task_id, field, token).await
45 }
46
47 pub async fn resolve_conflict_for_undo(
48 &self,
49 workspace: &Workspace,
50 task_id: &TaskId,
51 field: &str,
52 value: &str,
53 ) -> Result<ConflictResolutionOutcome> {
54 let mut conn = self.acquire().await?;
55 let before =
56 crate::undo::task_field_value(&mut conn, &workspace.id, task_id, field).await?;
57 let conflict_id =
58 crate::undo::conflict_row_id(&mut conn, &workspace.id, task_id, field).await?;
59 let outcome = resolve_conflict(&mut conn, workspace, task_id, field, value).await?;
60 let after = crate::undo::task_field_value(&mut conn, &workspace.id, task_id, field).await?;
61 Ok(ConflictResolutionOutcome {
62 outcome,
63 before,
64 after,
65 conflict_id,
66 })
67 }
68
69 pub async fn resolve_conflict(
70 &self,
71 workspace: &Workspace,
72 task_id: &TaskId,
73 field: &str,
74 value: &str,
75 ) -> Result<ConflictOutcome> {
76 let mut conn = self.acquire().await?;
77 resolve_conflict(&mut conn, workspace, task_id, field, value).await
78 }
79}
80
81pub struct ConflictListItem {
82 pub task_id: TaskId,
83 pub title: String,
84 pub project_key: String,
85 pub project_prefix: String,
86 pub field: String,
87 pub variant_a: String,
88 pub variant_b: String,
89}
90
91pub struct ConflictDetail {
92 pub field: String,
93 pub variant_a: String,
94 pub local_value: String,
95 pub variant_b: String,
96 pub remote_value: String,
97}
98
99pub struct ConflictOutcome {
100 pub task: Task,
101 pub field: String,
102}
103
104pub struct ConflictResolutionOutcome {
105 pub outcome: ConflictOutcome,
106 pub before: String,
107 pub after: String,
108 pub conflict_id: i64,
109}
110pub async fn list_conflicts(
111 conn: &mut SqliteConnection,
112 workspace: &Workspace,
113 project_key: Option<&str>,
114 field: Option<&str>,
115) -> Result<Vec<ConflictListItem>> {
116 let workspace_id = &workspace.id;
117 let project_id = if let Some(project) = project_key {
118 Some(
119 resolve_existing_project_in_workspace(conn, workspace_id, project)
120 .await?
121 .id,
122 )
123 } else {
124 None
125 };
126 let rows = sqlx::query(
127 r#"SELECT c.task_id, c.field, c.variant_a, c.variant_b,
128 t.title, p.prefix, p.key AS project_key
129 FROM conflicts c
130 JOIN tasks t ON t.workspace_id = c.workspace_id AND t.id = c.task_id
131 JOIN projects p ON p.workspace_id = t.workspace_id AND p.id = t.project_id
132 WHERE c.workspace_id = ? AND c.resolved = 0
133 AND (? IS NULL OR t.project_id = ?)
134 AND (? IS NULL OR c.field = ?)
135 ORDER BY c.created_at"#,
136 )
137 .bind(workspace_id)
138 .bind(&project_id)
139 .bind(&project_id)
140 .bind(field)
141 .bind(field)
142 .fetch_all(&mut *conn)
143 .await?;
144 Ok(rows
145 .into_iter()
146 .map(|row| ConflictListItem {
147 task_id: row.get("task_id"),
148 title: row.get("title"),
149 project_key: row.get("project_key"),
150 project_prefix: row.get("prefix"),
151 field: row.get("field"),
152 variant_a: row.get("variant_a"),
153 variant_b: row.get("variant_b"),
154 })
155 .collect())
156}
157
158pub async fn task_conflicts(
159 conn: &mut SqliteConnection,
160 workspace: &Workspace,
161 task_id: &crate::ids::TaskId,
162 field: Option<&str>,
163) -> Result<Vec<ConflictDetail>> {
164 let workspace_id = &workspace.id;
165 let rows = sqlx::query(
166 r#"SELECT field, variant_a, local_value, variant_b, remote_value
167 FROM conflicts
168 WHERE workspace_id = ? AND task_id = ? AND resolved = 0 AND (? IS NULL OR field = ?)
169 ORDER BY field, id"#,
170 )
171 .bind(workspace_id)
172 .bind(task_id)
173 .bind(field)
174 .bind(field)
175 .fetch_all(&mut *conn)
176 .await?;
177 Ok(rows
178 .into_iter()
179 .map(|row| ConflictDetail {
180 field: row.get("field"),
181 variant_a: row.get("variant_a"),
182 local_value: row.get("local_value"),
183 variant_b: row.get("variant_b"),
184 remote_value: row.get("remote_value"),
185 })
186 .collect())
187}
188
189pub async fn conflict_variant_value(
190 conn: &mut SqliteConnection,
191 workspace: &Workspace,
192 task_id: &crate::ids::TaskId,
193 field: &str,
194 token: &str,
195) -> Result<String> {
196 for detail in task_conflicts(conn, workspace, task_id, Some(field)).await? {
197 if token == detail.variant_a {
198 return Ok(detail.local_value);
199 }
200 if token == detail.variant_b {
201 return Ok(detail.remote_value);
202 }
203 }
204 bail!("error unknown-variant token={token}")
205}
206
207pub(crate) enum ConflictValueChoice {
208 Local,
209 Remote,
210}
211
212#[derive(Debug)]
213pub(crate) struct ConflictNotFoundError {
214 task_id: TaskId,
215 field: &'static str,
216}
217
218impl std::fmt::Display for ConflictNotFoundError {
219 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
220 write!(
221 formatter,
222 "error conflict-not-found task_id={} field={}",
223 self.task_id, self.field
224 )
225 }
226}
227
228impl std::error::Error for ConflictNotFoundError {}
229
230pub async fn resolve_conflict(
231 conn: &mut SqliteConnection,
232 workspace: &Workspace,
233 task_id: &crate::ids::TaskId,
234 field: &str,
235 value: &str,
236) -> Result<ConflictOutcome> {
237 resolve_conflict_value(
238 conn,
239 workspace,
240 task_id,
241 field,
242 ResolutionValue::Explicit(value),
243 )
244 .await
245}
246
247pub(crate) async fn resolve_conflict_choice(
248 conn: &mut SqliteConnection,
249 workspace: &Workspace,
250 task_id: &crate::ids::TaskId,
251 field: &str,
252 choice: ConflictValueChoice,
253) -> Result<ConflictOutcome> {
254 resolve_conflict_value(
255 conn,
256 workspace,
257 task_id,
258 field,
259 ResolutionValue::Choice(choice),
260 )
261 .await
262}
263
264enum ResolutionValue<'a> {
265 Explicit(&'a str),
266 Choice(ConflictValueChoice),
267}
268
269async fn resolve_conflict_value(
270 conn: &mut SqliteConnection,
271 workspace: &Workspace,
272 task_id: &crate::ids::TaskId,
273 field: &str,
274 resolution: ResolutionValue<'_>,
275) -> Result<ConflictOutcome> {
276 let task_field = TaskField::parse_or_unknown(field)?;
277 let field = task_field.as_str();
278 let mut tx = begin_immediate(conn).await?;
279 let value = match resolution {
280 ResolutionValue::Explicit(value) => value.to_string(),
281 ResolutionValue::Choice(choice) => {
282 let values = sqlx::query_as::<_, (String, String)>(
283 "SELECT local_value, remote_value FROM conflicts
284 WHERE workspace_id = ? AND task_id = ? AND field = ? AND resolved = 0",
285 )
286 .bind(&workspace.id)
287 .bind(task_id)
288 .bind(field)
289 .fetch_optional(&mut *tx)
290 .await?
291 .ok_or_else(|| {
292 anyhow::Error::new(ConflictNotFoundError {
293 task_id: task_id.clone(),
294 field,
295 })
296 })?;
297 match choice {
298 ConflictValueChoice::Local => values.0,
299 ConflictValueChoice::Remote => values.1,
300 }
301 }
302 };
303 if task_field == TaskField::IsEpic
304 && value == "0"
305 && crate::operations::task_has_epic_children(&mut tx, &workspace.id, task_id).await?
306 {
307 bail!("error epic-has-children task_id={task_id}");
308 }
309 let result = sqlx::query(
310 "UPDATE conflicts SET resolved = 1 WHERE workspace_id = ? AND task_id = ? AND field = ? AND resolved = 0",
311 )
312 .bind(&workspace.id)
313 .bind(task_id)
314 .bind(field)
315 .execute(&mut *tx)
316 .await?;
317 if result.rows_affected() != 1 {
318 return Err(anyhow::Error::new(ConflictNotFoundError {
319 task_id: task_id.clone(),
320 field,
321 }));
322 }
323 let payload = if task_field.is_project() {
324 let project = resolve_project_for_stored_value(&mut tx, &workspace.id, &value).await?;
325 apply_project_id_in_workspace(&mut tx, &workspace.id, task_id, &project.id).await?;
326 TaskField::project_payload(&workspace.id, &workspace.key, &project)
327 } else {
328 apply_field_value_in_workspace(&mut tx, &workspace.id, task_id, field, &value).await?;
329 task_field.scalar_payload(&workspace.id, &workspace.key, &value)?
330 };
331 let change_id = insert_change(
332 &mut tx,
333 "task",
334 task_id,
335 Some(field),
336 op_type::RESOLVE_FIELD,
337 payload,
338 None,
339 )
340 .await?;
341 set_field_version(&mut tx, task_id, field, &change_id).await?;
342 tx.commit().await?;
343 info!(task_id = %task_id, field = %field, "conflict resolved");
344 Ok(ConflictOutcome {
345 task: get_task_in_workspace(conn, workspace, task_id).await?,
346 field: field.to_string(),
347 })
348}