Skip to main content

max / audiofiles

4.9 KB · 172 lines History Blame Raw
1 //! Conflict resolution: apply remote changes locally with FK-safe ordering.
2
3 use rusqlite::Connection;
4 use synckit_client::{ChangeEntry, ChangeOp};
5
6 use tracing::instrument;
7
8 use crate::error::Result;
9
10 use super::{json_to_sql, pk_columns, table_columns, UPSERT_ORDER, DELETE_ORDER};
11
12 /// Apply a batch of remote changes locally, with trigger suppression.
13 ///
14 /// The flag set, data changes, and flag clear all happen inside a single
15 /// transaction so a process crash mid-apply rolls back everything — the flag
16 /// never gets stuck in the DB.
17 #[instrument(skip_all)]
18 pub(crate) fn apply_remote_changes(conn: &Connection, changes: &[ChangeEntry]) -> Result<i64> {
19 let tx = conn.unchecked_transaction()?;
20
21 // Suppress triggers while applying (inside the transaction)
22 tx.execute(
23 "UPDATE sync_state SET value = '1' WHERE key = 'applying_remote'",
24 [],
25 )?;
26
27 let mut count: i64 = 0;
28
29 // Separate upserts and deletes
30 let mut upserts: Vec<&ChangeEntry> = Vec::new();
31 let mut deletes: Vec<&ChangeEntry> = Vec::new();
32
33 for change in changes {
34 match change.op {
35 ChangeOp::Insert | ChangeOp::Update => upserts.push(change),
36 ChangeOp::Delete => deletes.push(change),
37 }
38 }
39
40 // Apply upserts in FK-safe order
41 for table in UPSERT_ORDER {
42 for change in &upserts {
43 if change.table == *table {
44 if let Some(data) = &change.data {
45 apply_upsert(&tx, table, data)?;
46 count += 1;
47 }
48 }
49 }
50 }
51
52 // Apply deletes in reverse FK order
53 for table in DELETE_ORDER {
54 for change in &deletes {
55 if change.table == *table {
56 apply_delete(&tx, table, &change.row_id)?;
57 count += 1;
58 }
59 }
60 }
61
62 // Clear the flag (still inside the transaction)
63 tx.execute(
64 "UPDATE sync_state SET value = '0' WHERE key = 'applying_remote'",
65 [],
66 )?;
67
68 tx.commit()?;
69
70 Ok(count)
71 }
72
73 /// Apply an upsert (INSERT OR REPLACE) for a single row.
74 pub(crate) fn apply_upsert(
75 conn: &Connection,
76 table: &str,
77 data: &serde_json::Value,
78 ) -> Result<()> {
79 let columns = match table_columns(table) {
80 Some(c) => c,
81 None => return Ok(()),
82 };
83
84 let obj = match data.as_object() {
85 Some(o) => o,
86 None => return Ok(()),
87 };
88
89 let mut col_names = Vec::new();
90 let mut placeholders = Vec::new();
91 let mut values: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
92
93 for (i, col) in columns.iter().enumerate() {
94 if let Some(val) = obj.get(*col) {
95 col_names.push(*col);
96 placeholders.push(format!("?{}", i + 1));
97 values.push(json_to_sql(val));
98 }
99 }
100
101 if col_names.is_empty() {
102 return Ok(());
103 }
104
105 // Use INSERT ... ON CONFLICT DO UPDATE to avoid DELETE+INSERT behavior of
106 // INSERT OR REPLACE, which would cascade FK deletes to child rows.
107 let pks = pk_columns(table);
108 let pk_set: std::collections::HashSet<&str> = pks.iter().copied().collect();
109 let non_pk_updates: Vec<String> = col_names
110 .iter()
111 .enumerate()
112 .filter(|(_, col)| !pk_set.contains(*col))
113 .map(|(i, col)| format!("{} = ?{}", col, i + 1))
114 .collect();
115
116 let sql = if non_pk_updates.is_empty() {
117 // All columns are PKs (e.g. tags) — just ignore conflicts
118 format!(
119 "INSERT OR IGNORE INTO {} ({}) VALUES ({})",
120 table,
121 col_names.join(", "),
122 placeholders.join(", "),
123 )
124 } else {
125 format!(
126 "INSERT INTO {} ({}) VALUES ({}) ON CONFLICT({}) DO UPDATE SET {}",
127 table,
128 col_names.join(", "),
129 placeholders.join(", "),
130 pks.join(", "),
131 non_pk_updates.join(", "),
132 )
133 };
134
135 conn.execute(
136 &sql,
137 values
138 .iter()
139 .map(|v| v.as_ref())
140 .collect::<Vec<_>>()
141 .as_slice(),
142 )?;
143
144 Ok(())
145 }
146
147 /// Apply a delete for a single row, handling composite primary keys.
148 pub(crate) fn apply_delete(conn: &Connection, table: &str, row_id: &str) -> Result<()> {
149 let pks = pk_columns(table);
150
151 if pks.len() == 1 {
152 let sql = format!("DELETE FROM {} WHERE {} = ?1", table, pks[0]);
153 conn.execute(&sql, [row_id])?;
154 } else if pks.len() == 2 {
155 // Composite PK: split row_id on first ':'
156 let (first, second) = match row_id.find(':') {
157 Some(pos) => (&row_id[..pos], &row_id[pos + 1..]),
158 None => {
159 tracing::warn!("Cannot split composite row_id for {table}: {row_id}");
160 return Ok(());
161 }
162 };
163 let sql = format!(
164 "DELETE FROM {} WHERE {} = ?1 AND {} = ?2",
165 table, pks[0], pks[1]
166 );
167 conn.execute(&sql, rusqlite::params![first, second])?;
168 }
169
170 Ok(())
171 }
172