use super::error::{SyncError, SyncResult};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::sync::Arc;
pub const MAX_CONFLICT_ID_BYTES: usize = 128;
pub const MAX_CONFLICT_REASON_BYTES: usize = 512;
pub const MAX_SYNC_CONFLICTS: usize = 4_096;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SyncConflictKind {
SequencePayload,
CheckpointHash,
BatchIdentity,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncConflict {
pub conflict_id: String,
pub peer_id: String,
pub kind: SyncConflictKind,
pub sequence: Option<u64>,
pub local_hash: String,
pub remote_hash: String,
pub reason: String,
pub detected_at_ms: u64,
}
impl SyncConflict {
pub fn validate(&self) -> SyncResult<()> {
for (name, value, max) in [
(
"conflict_id",
self.conflict_id.as_str(),
MAX_CONFLICT_ID_BYTES,
),
("peer_id", self.peer_id.as_str(), MAX_CONFLICT_ID_BYTES),
("reason", self.reason.as_str(), MAX_CONFLICT_REASON_BYTES),
] {
validate_text(name, value, max)?;
}
validate_hash(&self.local_hash)?;
validate_hash(&self.remote_hash)?;
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SyncConflictResolution {
KeepLocal,
AcceptRemote,
Defer,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncConflictResolutionRequest {
pub conflict_id: String,
pub idempotency_key: String,
pub resolution: SyncConflictResolution,
pub resolved_at_ms: u64,
}
impl SyncConflictResolutionRequest {
pub fn validate(&self) -> SyncResult<()> {
validate_text("conflict_id", &self.conflict_id, MAX_CONFLICT_ID_BYTES)?;
validate_text(
"idempotency_key",
&self.idempotency_key,
MAX_CONFLICT_ID_BYTES,
)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SyncConflictResolutionReceipt {
pub conflict_id: String,
pub idempotency_key: String,
pub resolution: SyncConflictResolution,
pub idempotent: bool,
pub resolved_at_ms: u64,
}
#[derive(Debug, Clone)]
struct ConflictRecord {
conflict: SyncConflict,
resolution: Option<SyncConflictResolutionReceipt>,
}
pub trait SyncConflictStore: Send + Sync {
fn record(&self, conflict: SyncConflict) -> SyncResult<()>;
fn resolve(
&self,
request: SyncConflictResolutionRequest,
) -> SyncResult<SyncConflictResolutionReceipt>;
fn list(&self) -> SyncResult<Vec<SyncConflict>>;
}
#[derive(Debug, Clone)]
pub struct InMemorySyncConflictStore {
records: Arc<Mutex<BTreeMap<String, ConflictRecord>>>,
}
impl Default for InMemorySyncConflictStore {
fn default() -> Self {
Self::new()
}
}
impl InMemorySyncConflictStore {
pub fn new() -> Self {
Self {
records: Arc::new(Mutex::new(BTreeMap::new())),
}
}
}
impl SyncConflictStore for InMemorySyncConflictStore {
fn record(&self, conflict: SyncConflict) -> SyncResult<()> {
conflict.validate()?;
let mut records = self.records.lock();
if let Some(existing) = records.get(&conflict.conflict_id) {
if existing.conflict == conflict {
return Ok(());
}
return Err(SyncError::InvalidConflict(
"conflict identity was reused with different evidence".to_string(),
));
}
if records.len() >= MAX_SYNC_CONFLICTS {
return Err(SyncError::ConflictStoreFull);
}
records.insert(
conflict.conflict_id.clone(),
ConflictRecord {
conflict,
resolution: None,
},
);
Ok(())
}
fn resolve(
&self,
request: SyncConflictResolutionRequest,
) -> SyncResult<SyncConflictResolutionReceipt> {
request.validate()?;
let mut records = self.records.lock();
let record = records
.get_mut(&request.conflict_id)
.ok_or_else(|| SyncError::ConflictNotFound(request.conflict_id.clone()))?;
if let Some(existing) = &record.resolution {
if existing.idempotency_key == request.idempotency_key
&& existing.resolution == request.resolution
{
return Ok(SyncConflictResolutionReceipt {
idempotent: true,
..existing.clone()
});
}
return Err(SyncError::ConflictAlreadyResolved(
request.conflict_id.clone(),
));
}
let receipt = SyncConflictResolutionReceipt {
conflict_id: request.conflict_id,
idempotency_key: request.idempotency_key,
resolution: request.resolution,
idempotent: false,
resolved_at_ms: request.resolved_at_ms,
};
record.resolution = Some(receipt.clone());
Ok(receipt)
}
fn list(&self) -> SyncResult<Vec<SyncConflict>> {
Ok(self
.records
.lock()
.values()
.map(|record| record.conflict.clone())
.collect())
}
}
fn validate_text(name: &str, value: &str, max: usize) -> SyncResult<()> {
if value.trim().is_empty() || value.len() > max || value.chars().any(char::is_control) {
return Err(SyncError::InvalidConflict(format!(
"{name} is empty, too long or contains control characters"
)));
}
Ok(())
}
fn validate_hash(value: &str) -> SyncResult<()> {
if value.len() != 64
|| !value
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
{
return Err(SyncError::InvalidConflict(
"conflict hash must be lowercase SHA-256".to_string(),
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn conflict() -> SyncConflict {
SyncConflict {
conflict_id: "conflict-a".to_string(),
peer_id: "peer-a".to_string(),
kind: SyncConflictKind::SequencePayload,
sequence: Some(42),
local_hash: "11".repeat(32),
remote_hash: "22".repeat(32),
reason: "same sequence has different bytes".to_string(),
detected_at_ms: 100,
}
}
#[test]
fn resolution_replay_is_idempotent_and_payload_free() {
let store = InMemorySyncConflictStore::new();
store.record(conflict()).unwrap();
let request = SyncConflictResolutionRequest {
conflict_id: "conflict-a".to_string(),
idempotency_key: "resolve-a".to_string(),
resolution: SyncConflictResolution::KeepLocal,
resolved_at_ms: 200,
};
let first = store.resolve(request.clone()).unwrap();
let replay = store.resolve(request).unwrap();
assert!(!first.idempotent);
assert!(replay.idempotent);
assert_eq!(store.list().unwrap()[0].local_hash, "11".repeat(32));
}
#[test]
fn conflicting_resolution_is_rejected_without_overwrite() {
let store = InMemorySyncConflictStore::new();
store.record(conflict()).unwrap();
store
.resolve(SyncConflictResolutionRequest {
conflict_id: "conflict-a".to_string(),
idempotency_key: "resolve-a".to_string(),
resolution: SyncConflictResolution::KeepLocal,
resolved_at_ms: 200,
})
.unwrap();
assert!(matches!(
store.resolve(SyncConflictResolutionRequest {
conflict_id: "conflict-a".to_string(),
idempotency_key: "resolve-b".to_string(),
resolution: SyncConflictResolution::AcceptRemote,
resolved_at_ms: 300,
}),
Err(SyncError::ConflictAlreadyResolved(id)) if id == "conflict-a"
));
}
}