appcore_sync_sqlite/
conflict.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: conflict.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/10/12 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/10/12 00:00:00 by dnettoRaw
8//      ###########      S: 1.0.2-rc
9// =============================================================================
10
11//! Durable conflict evidence and idempotent resolution without domain payloads.
12use crate::{SqliteSyncError, SqliteSyncResult, SqliteSyncStore};
13use appcore_sync::{
14    SyncConflict, SyncConflictResolutionReceipt, SyncConflictResolutionRequest, SyncConflictStore,
15    SyncError, SyncResult, MAX_SYNC_CONFLICTS,
16};
17use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior};
18
19const MAX_CONFLICT_BYTES: usize = 8_192;
20const MAX_RESOLUTION_BYTES: usize = 4_096;
21
22/// SQLite-backed bounded conflict registry. It records decisions, never applies them.
23#[derive(Clone, Debug)]
24pub struct SqliteSyncConflictStore {
25    store: SqliteSyncStore,
26}
27impl SqliteSyncStore {
28    /// Returns a conflict store sharing this database and its integrity/backup boundary.
29    pub fn conflict_store(&self) -> SqliteSyncConflictStore {
30        SqliteSyncConflictStore {
31            store: self.clone(),
32        }
33    }
34}
35impl SyncConflictStore for SqliteSyncConflictStore {
36    fn record(&self, conflict: SyncConflict) -> SyncResult<()> {
37        conflict.validate()?;
38        self.store
39            .with_connection(|connection| {
40                let transaction = connection
41                    .transaction_with_behavior(TransactionBehavior::Immediate)
42                    .map_err(SqliteSyncError::database)?;
43                let existing: Option<String> = transaction
44                    .query_row(
45                        "SELECT CASE WHEN length(CAST(evidence AS BLOB)) <= 8192 THEN evidence ELSE NULL END FROM appcore_sync_conflict WHERE conflict_id = ?1",
46                        [&conflict.conflict_id],
47                        |row| row.get(0),
48                    )
49                    .optional()
50                    .map_err(SqliteSyncError::database)?;
51                if let Some(existing) = existing {
52                    if decode_conflict(&existing)? == conflict {
53                        return Ok(());
54                    }
55                    return Err(SqliteSyncError::CorruptRecord("conflict identity"));
56                }
57                let count: i64 = transaction
58                    .query_row("SELECT COUNT(*) FROM appcore_sync_conflict", [], |row| {
59                        row.get(0)
60                    })
61                    .map_err(SqliteSyncError::database)?;
62                if count >= MAX_SYNC_CONFLICTS as i64 {
63                    return Err(SqliteSyncError::CapacityExceeded("conflict"));
64                }
65                let encoded = serde_json::to_string(&conflict)
66                    .map_err(|_| SqliteSyncError::CorruptRecord("conflict"))?;
67                if encoded.len() > MAX_CONFLICT_BYTES {
68                    return Err(SqliteSyncError::CapacityExceeded("conflict record"));
69                }
70                transaction
71                    .execute(
72                        "INSERT INTO appcore_sync_conflict(conflict_id, evidence) VALUES (?1, ?2)",
73                        params![conflict.conflict_id, encoded],
74                    )
75                    .map_err(SqliteSyncError::database)?;
76                transaction.commit().map_err(SqliteSyncError::database)
77            })
78            .map_err(|error| match error {
79                SqliteSyncError::CapacityExceeded("conflict") => SyncError::ConflictStoreFull,
80                other => other.sync(),
81            })
82    }
83    fn resolve(
84        &self,
85        request: SyncConflictResolutionRequest,
86    ) -> SyncResult<SyncConflictResolutionReceipt> {
87        request.validate()?;
88        self.store
89            .with_connection(|connection| {
90                let transaction = connection
91                    .transaction_with_behavior(TransactionBehavior::Immediate)
92                    .map_err(SqliteSyncError::database)?;
93                let row: Option<(String, Option<String>)> = transaction.query_row(
94                "SELECT CASE WHEN length(CAST(evidence AS BLOB)) <= 8192 AND (resolution IS NULL OR length(CAST(resolution AS BLOB)) <= 4096) THEN evidence ELSE NULL END, CASE WHEN length(CAST(resolution AS BLOB)) <= 4096 THEN resolution ELSE NULL END FROM appcore_sync_conflict WHERE conflict_id = ?1",
95                [&request.conflict_id], |row| Ok((row.get(0)?, row.get(1)?)),
96            ).optional().map_err(SqliteSyncError::database)?;
97                let Some((evidence, resolution)) = row else {
98                    return Err(SqliteSyncError::CorruptRecord("conflict missing"));
99                };
100                let conflict = decode_conflict(&evidence)?;
101            if conflict.conflict_id != request.conflict_id { return Err(SqliteSyncError::IntegrityFailed); }
102                if let Some(resolution) = resolution {
103                    let mut receipt = decode_resolution(&resolution, &conflict.conflict_id)?;
104                    if receipt.idempotency_key != request.idempotency_key
105                        || receipt.resolution != request.resolution
106                    {
107                        return Err(SqliteSyncError::CorruptRecord("conflict resolved"));
108                    }
109                    receipt.idempotent = true;
110                    return Ok(receipt);
111                }
112                let receipt = SyncConflictResolutionReceipt {
113                    conflict_id: request.conflict_id.clone(),
114                    idempotency_key: request.idempotency_key,
115                    resolution: request.resolution,
116                    idempotent: false,
117                    resolved_at_ms: request.resolved_at_ms,
118                };
119                let encoded = serde_json::to_string(&receipt)
120                    .map_err(|_| SqliteSyncError::CorruptRecord("resolution"))?;
121                if encoded.len() > MAX_RESOLUTION_BYTES {
122                    return Err(SqliteSyncError::CapacityExceeded("resolution"));
123                }
124                transaction
125                    .execute(
126                        "UPDATE appcore_sync_conflict SET resolution = ?1 WHERE conflict_id = ?2",
127                        params![encoded, receipt.conflict_id],
128                    )
129                    .map_err(SqliteSyncError::database)?;
130                transaction.commit().map_err(SqliteSyncError::database)?;
131                Ok(receipt)
132            })
133            .map_err(|error| match error {
134                SqliteSyncError::CorruptRecord("conflict missing") => {
135                    SyncError::ConflictNotFound(request.conflict_id.clone())
136                }
137                SqliteSyncError::CorruptRecord("conflict resolved") => {
138                    SyncError::ConflictAlreadyResolved(request.conflict_id.clone())
139                }
140                other => other.sync(),
141            })
142    }
143    fn list(&self) -> SyncResult<Vec<SyncConflict>> {
144        self.store
145            .with_connection(|connection| {
146                read_conflicts(connection)
147                    .map(|records| records.into_iter().map(|(conflict, _)| conflict).collect())
148            })
149            .map_err(SqliteSyncError::sync)
150    }
151}
152
153pub(crate) fn validate_conflicts(connection: &Connection) -> SqliteSyncResult<()> {
154    read_conflicts(connection).map(|_| ())
155}
156fn read_conflicts(
157    connection: &Connection,
158) -> SqliteSyncResult<Vec<(SyncConflict, Option<SyncConflictResolutionReceipt>)>> {
159    let mut statement = connection.prepare("SELECT conflict_id, CASE WHEN length(CAST(evidence AS BLOB)) <= 8192 AND (resolution IS NULL OR length(CAST(resolution AS BLOB)) <= 4096) THEN evidence ELSE NULL END, resolution FROM appcore_sync_conflict ORDER BY conflict_id LIMIT 4097")
160        .map_err(SqliteSyncError::database)?;
161    let rows = statement
162        .query_map([], |row| {
163            Ok((
164                row.get::<_, String>(0)?,
165                row.get::<_, String>(1)?,
166                row.get::<_, Option<String>>(2)?,
167            ))
168        })
169        .map_err(SqliteSyncError::database)?;
170    let mut records = Vec::new();
171    for row in rows {
172        let (id, encoded, resolution) = row.map_err(SqliteSyncError::database)?;
173        let conflict = decode_conflict(&encoded)?;
174        if conflict.conflict_id != id || records.len() >= MAX_SYNC_CONFLICTS {
175            return Err(SqliteSyncError::IntegrityFailed);
176        }
177        let resolution = resolution
178            .map(|value| decode_resolution(&value, &id))
179            .transpose()?;
180        records.push((conflict, resolution));
181    }
182    Ok(records)
183}
184fn decode_conflict(encoded: &str) -> SqliteSyncResult<SyncConflict> {
185    if encoded.len() > MAX_CONFLICT_BYTES {
186        return Err(SqliteSyncError::IntegrityFailed);
187    }
188    let value: SyncConflict =
189        serde_json::from_str(encoded).map_err(|_| SqliteSyncError::IntegrityFailed)?;
190    value
191        .validate()
192        .map_err(|_| SqliteSyncError::IntegrityFailed)?;
193    Ok(value)
194}
195fn decode_resolution(encoded: &str, id: &str) -> SqliteSyncResult<SyncConflictResolutionReceipt> {
196    if encoded.len() > MAX_RESOLUTION_BYTES {
197        return Err(SqliteSyncError::IntegrityFailed);
198    }
199    let value: SyncConflictResolutionReceipt =
200        serde_json::from_str(encoded).map_err(|_| SqliteSyncError::IntegrityFailed)?;
201    if value.conflict_id != id || value.idempotent {
202        return Err(SqliteSyncError::IntegrityFailed);
203    }
204    SyncConflictResolutionRequest {
205        conflict_id: value.conflict_id.clone(),
206        idempotency_key: value.idempotency_key.clone(),
207        resolution: value.resolution,
208        resolved_at_ms: value.resolved_at_ms,
209    }
210    .validate()
211    .map_err(|_| SqliteSyncError::IntegrityFailed)?;
212    Ok(value)
213}
appcore_sync_sqlite/conflict.rs

appcore_sync_sqlite/
conflict.rs
