mur_core/team/

crdt.rs

//! Lightweight LWW-Element-Set CRDT for workflow YAML synchronization.
//!
//! Implements a Last-Writer-Wins Element Set where each element is a
//! workflow identified by its ID, and the value is the YAML content.
//! Conflicts are resolved by comparing (timestamp, peer_id) tuples.

use std::collections::HashMap;

use chrono::{DateTime, Utc};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use super::{ConflictResolution, SyncConflict, SyncState, TeamSync};

/// A timestamped value in the LWW register.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub struct LwwEntry {
    /// The workflow YAML content.
    pub value: String,
    /// Wall-clock timestamp of the write.
    pub timestamp: DateTime<Utc>,
    /// Peer that performed the write.
    pub peer_id: Uuid,
    /// Monotonically increasing version counter (per-key).
    pub version: u64,
}

/// LWW-Element-Set for synchronizing workflow YAML across peers.
///
/// Each workflow ID maps to an `LwwEntry`. When merging, the entry
/// with the higher timestamp wins; ties are broken by peer ID ordering.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LwwElementSet {
    /// The local peer's identity.
    local_peer: Uuid,
    /// Conflict resolution strategy.
    resolution: ConflictResolution,
    /// The element set: workflow_id → LwwEntry.
    elements: HashMap<String, LwwEntry>,
    /// Tombstones for removed workflows.
    removed: HashMap<String, LwwEntry>,
}

impl LwwElementSet {
    /// Create a new empty LWW-Element-Set for the given peer.
    pub fn new(local_peer: Uuid, resolution: ConflictResolution) -> Self {
        Self {
            local_peer,
            resolution,
            elements: HashMap::new(),
            removed: HashMap::new(),
        }
    }

    /// Insert or update a workflow YAML value.
    pub fn set(&mut self, workflow_id: &str, yaml: &str) -> LwwEntry {
        let version = self
            .elements
            .get(workflow_id)
            .map(|e| e.version + 1)
            .unwrap_or(1);

        let entry = LwwEntry {
            value: yaml.to_string(),
            timestamp: Utc::now(),
            peer_id: self.local_peer,
            version,
        };

        self.elements.insert(workflow_id.to_string(), entry.clone());
        self.removed.remove(workflow_id);
        entry
    }

    /// Remove a workflow (add to tombstone set).
    pub fn remove(&mut self, workflow_id: &str) -> Option<LwwEntry> {
        if let Some(entry) = self.elements.remove(workflow_id) {
            let tombstone = LwwEntry {
                value: String::new(),
                timestamp: Utc::now(),
                peer_id: self.local_peer,
                version: entry.version + 1,
            };
            self.removed.insert(workflow_id.to_string(), tombstone.clone());
            Some(tombstone)
        } else {
            None
        }
    }

    /// Get the current value for a workflow.
    pub fn get(&self, workflow_id: &str) -> Option<&LwwEntry> {
        self.elements.get(workflow_id)
    }

    /// List all active (non-removed) workflow IDs.
    pub fn workflows(&self) -> Vec<String> {
        self.elements.keys().cloned().collect()
    }

    /// Merge a remote entry for a given workflow ID.
    /// Returns the winning entry and any conflict detected.
    pub fn merge(
        &mut self,
        workflow_id: &str,
        remote: &LwwEntry,
    ) -> (LwwEntry, Option<SyncConflict>) {
        // Check if remote is a tombstone (empty value after our element)
        if let Some(tombstone) = self.removed.get(workflow_id) {
            if Self::entry_wins(tombstone, remote) {
                return (tombstone.clone(), None);
            }
        }

        if let Some(local) = self.elements.get(workflow_id) {
            if local.timestamp == remote.timestamp && local.peer_id != remote.peer_id {
                // True conflict: same timestamp, different peers
                let conflict = SyncConflict {
                    workflow_id: workflow_id.to_string(),
                    local_version: local.version,
                    remote_version: remote.version,
                    local_peer: local.peer_id.to_string(),
                    remote_peer: remote.peer_id.to_string(),
                    detected_at: Utc::now(),
                };

                let winner = self.resolve(local, remote);
                self.elements.insert(workflow_id.to_string(), winner.clone());
                (winner, Some(conflict))
            } else if Self::entry_wins(remote, local) {
                // Remote wins
                self.elements.insert(workflow_id.to_string(), remote.clone());
                (remote.clone(), None)
            } else {
                // Local wins
                (local.clone(), None)
            }
        } else {
            // No local entry, accept remote
            self.elements.insert(workflow_id.to_string(), remote.clone());
            (remote.clone(), None)
        }
    }

    /// Merge an entire remote element set into the local one.
    pub fn merge_set(&mut self, remote_set: &LwwElementSet) -> Vec<SyncConflict> {
        let mut conflicts = Vec::new();

        for (wf_id, remote_entry) in &remote_set.elements {
            let (_, conflict) = self.merge(wf_id, remote_entry);
            if let Some(c) = conflict {
                conflicts.push(c);
            }
        }

        // Merge tombstones
        for (wf_id, remote_tombstone) in &remote_set.removed {
            if let Some(local) = self.elements.get(wf_id) {
                if Self::entry_wins(remote_tombstone, local) {
                    self.elements.remove(wf_id);
                    self.removed.insert(wf_id.clone(), remote_tombstone.clone());
                }
            }
            // Merge tombstone timestamps (keep newer)
            if let Some(local_tomb) = self.removed.get(wf_id) {
                if Self::entry_wins(remote_tombstone, local_tomb) {
                    self.removed.insert(wf_id.clone(), remote_tombstone.clone());
                }
            } else if !self.elements.contains_key(wf_id) {
                self.removed.insert(wf_id.clone(), remote_tombstone.clone());
            }
        }

        conflicts
    }

    /// Compare two entries: returns true if `a` wins over `b`.
    fn entry_wins(a: &LwwEntry, b: &LwwEntry) -> bool {
        (a.timestamp, a.peer_id) > (b.timestamp, b.peer_id)
    }

    /// Resolve a conflict between two entries using the configured strategy.
    fn resolve(&self, local: &LwwEntry, remote: &LwwEntry) -> LwwEntry {
        match self.resolution {
            ConflictResolution::LastWriterWins => {
                if Self::entry_wins(local, remote) {
                    local.clone()
                } else {
                    remote.clone()
                }
            }
            ConflictResolution::HighestNodeId => {
                if local.peer_id >= remote.peer_id {
                    local.clone()
                } else {
                    remote.clone()
                }
            }
            ConflictResolution::ForkAndMerge => {
                // In fork mode, keep local but the caller should handle the conflict
                local.clone()
            }
        }
    }
}

impl TeamSync for LwwElementSet {
    fn push(&self, workflow_id: &str, _yaml: &str) -> anyhow::Result<SyncState> {
        match self.elements.get(workflow_id) {
            Some(_) => Ok(SyncState::Synced),
            None => Ok(SyncState::LocalAhead {
                pending_changes: 1,
            }),
        }
    }

    fn pull(&self, workflow_id: &str) -> anyhow::Result<(String, SyncState)> {
        match self.elements.get(workflow_id) {
            Some(entry) => Ok((entry.value.clone(), SyncState::Synced)),
            None => anyhow::bail!("workflow '{workflow_id}' not found in element set"),
        }
    }

    fn state(&self, workflow_id: &str) -> anyhow::Result<SyncState> {
        if self.elements.contains_key(workflow_id) {
            Ok(SyncState::Synced)
        } else {
            Ok(SyncState::LocalAhead {
                pending_changes: 0,
            })
        }
    }

    fn resolve_conflict(
        &self,
        workflow_id: &str,
        _resolution: &ConflictResolution,
    ) -> anyhow::Result<String> {
        self.elements
            .get(workflow_id)
            .map(|e| e.value.clone())
            .ok_or_else(|| anyhow::anyhow!("workflow '{workflow_id}' not found"))
    }

    fn local_peer_id(&self) -> Uuid {
        self.local_peer
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn make_peer() -> Uuid {
        Uuid::new_v4()
    }

    #[test]
    fn test_set_and_get() {
        let peer = make_peer();
        let mut set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);

        set.set("wf-1", "name: test\nsteps: []");
        let entry = set.get("wf-1").unwrap();
        assert_eq!(entry.value, "name: test\nsteps: []");
        assert_eq!(entry.version, 1);
        assert_eq!(entry.peer_id, peer);
    }

    #[test]
    fn test_set_increments_version() {
        let peer = make_peer();
        let mut set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);

        set.set("wf-1", "v1");
        set.set("wf-1", "v2");
        let entry = set.get("wf-1").unwrap();
        assert_eq!(entry.value, "v2");
        assert_eq!(entry.version, 2);
    }

    #[test]
    fn test_remove() {
        let peer = make_peer();
        let mut set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);

        set.set("wf-1", "content");
        let tombstone = set.remove("wf-1");
        assert!(tombstone.is_some());
        assert!(set.get("wf-1").is_none());
        assert!(set.workflows().is_empty());
    }

    #[test]
    fn test_remove_nonexistent() {
        let peer = make_peer();
        let mut set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);
        assert!(set.remove("wf-1").is_none());
    }

    #[test]
    fn test_merge_new_entry() {
        let peer1 = make_peer();
        let peer2 = make_peer();
        let mut set = LwwElementSet::new(peer1, ConflictResolution::LastWriterWins);

        let remote = LwwEntry {
            value: "remote content".into(),
            timestamp: Utc::now(),
            peer_id: peer2,
            version: 1,
        };

        let (winner, conflict) = set.merge("wf-1", &remote);
        assert!(conflict.is_none());
        assert_eq!(winner.value, "remote content");
        assert_eq!(set.get("wf-1").unwrap().value, "remote content");
    }

    #[test]
    fn test_merge_remote_wins() {
        let peer1 = make_peer();
        let peer2 = make_peer();
        let mut set = LwwElementSet::new(peer1, ConflictResolution::LastWriterWins);

        set.set("wf-1", "old");

        // Sleep a tiny bit to ensure different timestamp
        let remote = LwwEntry {
            value: "newer".into(),
            timestamp: Utc::now() + chrono::Duration::seconds(1),
            peer_id: peer2,
            version: 2,
        };

        let (winner, _) = set.merge("wf-1", &remote);
        assert_eq!(winner.value, "newer");
    }

    #[test]
    fn test_merge_local_wins() {
        let peer1 = make_peer();
        let peer2 = make_peer();
        let mut set = LwwElementSet::new(peer1, ConflictResolution::LastWriterWins);

        set.set("wf-1", "local");

        let remote = LwwEntry {
            value: "older".into(),
            timestamp: Utc::now() - chrono::Duration::seconds(10),
            peer_id: peer2,
            version: 1,
        };

        let (winner, _) = set.merge("wf-1", &remote);
        assert_eq!(winner.value, "local");
    }

    #[test]
    fn test_merge_set() {
        let peer1 = make_peer();
        let peer2 = make_peer();
        let mut set1 = LwwElementSet::new(peer1, ConflictResolution::LastWriterWins);
        let mut set2 = LwwElementSet::new(peer2, ConflictResolution::LastWriterWins);

        set1.set("wf-1", "from peer1");
        set2.set("wf-2", "from peer2");

        let conflicts = set1.merge_set(&set2);
        assert!(conflicts.is_empty());
        assert_eq!(set1.get("wf-1").unwrap().value, "from peer1");
        assert_eq!(set1.get("wf-2").unwrap().value, "from peer2");
    }

    #[test]
    fn test_workflows_list() {
        let peer = make_peer();
        let mut set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);

        set.set("wf-a", "a");
        set.set("wf-b", "b");
        set.set("wf-c", "c");

        let mut wfs = set.workflows();
        wfs.sort();
        assert_eq!(wfs, vec!["wf-a", "wf-b", "wf-c"]);
    }

    #[test]
    fn test_lww_entry_serialization() {
        let entry = LwwEntry {
            value: "test yaml".into(),
            timestamp: Utc::now(),
            peer_id: Uuid::new_v4(),
            version: 42,
        };
        let json = serde_json::to_string(&entry).unwrap();
        let back: LwwEntry = serde_json::from_str(&json).unwrap();
        assert_eq!(entry.value, back.value);
        assert_eq!(entry.version, back.version);
    }

    #[test]
    fn test_team_sync_trait_push() {
        let peer = make_peer();
        let mut set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);
        set.set("wf-1", "content");

        let state = set.push("wf-1", "content").unwrap();
        assert_eq!(state, SyncState::Synced);
    }

    #[test]
    fn test_team_sync_trait_pull() {
        let peer = make_peer();
        let mut set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);
        set.set("wf-1", "yaml content");

        let (yaml, state) = set.pull("wf-1").unwrap();
        assert_eq!(yaml, "yaml content");
        assert_eq!(state, SyncState::Synced);
    }

    #[test]
    fn test_team_sync_trait_pull_missing() {
        let peer = make_peer();
        let set = LwwElementSet::new(peer, ConflictResolution::LastWriterWins);
        assert!(set.pull("nonexistent").is_err());
    }

    #[test]
    fn test_highest_node_id_resolution() {
        let peer1 = Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
        let peer2 = Uuid::parse_str("ffffffff-ffff-ffff-ffff-ffffffffffff").unwrap();
        let mut set = LwwElementSet::new(peer1, ConflictResolution::HighestNodeId);

        let ts = Utc::now();
        set.set("wf-1", "from low peer");

        // Force same timestamp for conflict
        let remote = LwwEntry {
            value: "from high peer".into(),
            timestamp: ts,
            peer_id: peer2,
            version: 1,
        };

        // Overwrite local entry timestamp to match
        set.elements.get_mut("wf-1").unwrap().timestamp = ts;

        let (winner, conflict) = set.merge("wf-1", &remote);
        assert!(conflict.is_some());
        assert_eq!(winner.value, "from high peer");
    }
}