automorph/crdt/

counter.rs

//! Counter type for concurrent increment/decrement operations.
//!
//! Unlike regular integers which use last-writer-wins semantics, Counter values
//! merge all concurrent increments/decrements. This makes them ideal for:
//! - View counts
//! - Like/vote counts
//! - Inventory quantities
//! - Any numeric value that multiple users might update concurrently
//!
//! # Example
//!
//! ```rust,ignore
//! use automorph::{Automorph, crdt::Counter};
//!
//! #[derive(Automorph)]
//! struct Post {
//!     title: String,
//!     likes: Counter,  // Concurrent likes merge correctly
//! }
//!
//! // User A likes the post
//! post.likes.increment(1);
//! post.save(&mut doc_a, &ROOT, "post")?;
//!
//! // User B also likes the post (concurrently)
//! post.likes.increment(1);
//! post.save(&mut doc_b, &ROOT, "post")?;
//!
//! // After merge: likes = 2 (both increments preserved)
//! ```
//!
//! # CRDT Semantics
//!
//! Counter uses Automerge's native counter type, which is a **G-Counter** (grow-only counter)
//! extended to support decrements. Key properties:
//!
//! - **Commutative**: Order of operations doesn't matter
//! - **Associative**: Grouping of operations doesn't matter
//! - **Idempotent**: Duplicate operations are handled correctly
//!
//! This means concurrent increments ALWAYS result in all increments being applied,
//! unlike regular integers where one update would overwrite another.

use std::fmt;
use std::ops::{Add, AddAssign, Sub, SubAssign};

use automerge::{ChangeHash, ObjId, Prop, ReadDoc, ScalarValue, Value, transaction::Transactable};

use crate::{Automorph, Error, PrimitiveChanged, Result, ScalarCursor};

/// A CRDT counter that supports concurrent increment/decrement.
///
/// This wraps Automerge's native counter type, ensuring that concurrent
/// modifications merge correctly instead of using last-writer-wins.
///
/// # Important
///
/// To get the CRDT benefits, you must use [`increment()`](Self::increment) and
/// [`decrement()`](Self::decrement) methods. Setting the value directly with
/// [`set()`](Self::set) uses last-writer-wins semantics.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
pub struct Counter {
    value: i64,
    /// Pending increment to apply on next save.
    /// This is how we track increments between load and save.
    pending_delta: i64,
}

impl Counter {
    /// Creates a new counter with the given initial value.
    #[must_use]
    pub fn new(value: i64) -> Self {
        Self {
            value,
            pending_delta: 0,
        }
    }

    /// Returns the current value of the counter.
    #[must_use]
    pub fn value(&self) -> i64 {
        self.value + self.pending_delta
    }

    /// Increments the counter by the given amount.
    ///
    /// This increment will be merged with concurrent increments from other peers,
    /// rather than overwriting them.
    pub fn increment(&mut self, amount: i64) {
        self.pending_delta += amount;
    }

    /// Decrements the counter by the given amount.
    ///
    /// This is equivalent to `increment(-amount)`.
    pub fn decrement(&mut self, amount: i64) {
        self.pending_delta -= amount;
    }

    /// Sets the counter to a specific value.
    ///
    /// **Warning**: This uses last-writer-wins semantics and will overwrite
    /// concurrent changes. Prefer [`increment()`](Self::increment) and
    /// [`decrement()`](Self::decrement) for collaborative scenarios.
    pub fn set(&mut self, value: i64) {
        // Setting resets both value and delta
        self.value = value;
        self.pending_delta = 0;
    }

    /// Returns true if there are pending changes to save.
    #[must_use]
    pub fn has_pending_changes(&self) -> bool {
        self.pending_delta != 0
    }
}

impl fmt::Debug for Counter {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Counter({})", self.value())
    }
}

impl fmt::Display for Counter {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.value())
    }
}

impl From<i64> for Counter {
    fn from(value: i64) -> Self {
        Self::new(value)
    }
}

impl From<Counter> for i64 {
    fn from(counter: Counter) -> Self {
        counter.value()
    }
}

// Arithmetic operations

impl Add<i64> for Counter {
    type Output = Self;

    fn add(mut self, rhs: i64) -> Self::Output {
        self.increment(rhs);
        self
    }
}

impl AddAssign<i64> for Counter {
    fn add_assign(&mut self, rhs: i64) {
        self.increment(rhs);
    }
}

impl Sub<i64> for Counter {
    type Output = Self;

    fn sub(mut self, rhs: i64) -> Self::Output {
        self.decrement(rhs);
        self
    }
}

impl SubAssign<i64> for Counter {
    fn sub_assign(&mut self, rhs: i64) {
        self.decrement(rhs);
    }
}

impl Automorph for Counter {
    type Changes = PrimitiveChanged;
    type Cursor = ScalarCursor;

    fn save<D: Transactable + ReadDoc>(
        &self,
        doc: &mut D,
        obj: impl AsRef<ObjId>,
        prop: impl Into<Prop>,
    ) -> Result<()> {
        let prop: Prop = prop.into();
        let obj = obj.as_ref();

        // Check if counter already exists
        match doc.get(obj, prop.clone())? {
            Some((Value::Scalar(s), _)) if s.is_counter() => {
                // Counter exists - apply increment if any
                if self.pending_delta != 0 {
                    doc.increment(obj, prop, self.pending_delta)?;
                }
            }
            _ => {
                // No counter exists or wrong type - create new counter
                doc.put(obj, prop, ScalarValue::counter(self.value()))?;
            }
        }

        Ok(())
    }

    fn load<D: ReadDoc>(doc: &D, obj: impl AsRef<ObjId>, prop: impl Into<Prop>) -> Result<Self> {
        let prop: Prop = prop.into();
        let obj = obj.as_ref();

        match doc.get(obj, prop)? {
            Some((Value::Scalar(s), _)) => {
                // Counters and integers both provide i64 values
                // is_counter() tells us if it's a proper counter type
                if let Some(int_val) = s.to_i64() {
                    Ok(Self::new(int_val))
                } else {
                    Err(Error::type_mismatch("Counter", Some(format!("{:?}", s))))
                }
            }
            Some((v, _)) => Err(Error::type_mismatch("Counter", Some(format!("{:?}", v)))),
            None => Err(Error::missing_value()),
        }
    }

    fn load_at<D: ReadDoc>(
        doc: &D,
        obj: impl AsRef<ObjId>,
        prop: impl Into<Prop>,
        heads: &[ChangeHash],
    ) -> Result<Self> {
        let prop: Prop = prop.into();
        let obj = obj.as_ref();

        match doc.get_at(obj, prop, heads)? {
            Some((Value::Scalar(s), _)) => {
                // Counters and integers both provide i64 values
                // is_counter() tells us if it's a proper counter type
                if let Some(int_val) = s.to_i64() {
                    Ok(Self::new(int_val))
                } else {
                    Err(Error::type_mismatch("Counter", Some(format!("{:?}", s))))
                }
            }
            Some((v, _)) => Err(Error::type_mismatch("Counter", Some(format!("{:?}", v)))),
            None => Err(Error::missing_value()),
        }
    }

    fn diff<D: ReadDoc>(
        &self,
        doc: &D,
        obj: impl AsRef<ObjId>,
        prop: impl Into<Prop>,
    ) -> Result<Self::Changes> {
        let loaded = Self::load(doc, obj, prop)?;
        Ok(PrimitiveChanged::new(self.value() != loaded.value()))
    }

    fn diff_at<D: ReadDoc>(
        &self,
        doc: &D,
        obj: impl AsRef<ObjId>,
        prop: impl Into<Prop>,
        heads: &[ChangeHash],
    ) -> Result<Self::Changes> {
        let loaded = Self::load_at(doc, obj, prop, heads)?;
        Ok(PrimitiveChanged::new(self.value() != loaded.value()))
    }

    fn update<D: ReadDoc>(
        &mut self,
        doc: &D,
        obj: impl AsRef<ObjId>,
        prop: impl Into<Prop>,
    ) -> Result<Self::Changes> {
        let loaded = Self::load(doc, obj, prop)?;
        let changed = self.value() != loaded.value();
        self.value = loaded.value;
        self.pending_delta = 0;
        Ok(PrimitiveChanged::new(changed))
    }

    fn update_at<D: ReadDoc>(
        &mut self,
        doc: &D,
        obj: impl AsRef<ObjId>,
        prop: impl Into<Prop>,
        heads: &[ChangeHash],
    ) -> Result<Self::Changes> {
        let loaded = Self::load_at(doc, obj, prop, heads)?;
        let changed = self.value() != loaded.value();
        self.value = loaded.value;
        self.pending_delta = 0;
        Ok(PrimitiveChanged::new(changed))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use automerge::{ActorId, AutoCommit, ROOT};

    #[test]
    fn test_counter_basic_operations() {
        let mut counter = Counter::new(0);
        assert_eq!(counter.value(), 0);

        counter.increment(5);
        assert_eq!(counter.value(), 5);

        counter.decrement(2);
        assert_eq!(counter.value(), 3);
    }

    #[test]
    fn test_counter_roundtrip() {
        let mut doc = AutoCommit::new();

        let counter = Counter::new(42);
        counter.save(&mut doc, &ROOT, "count").unwrap();

        let loaded = Counter::load(&doc, &ROOT, "count").unwrap();
        assert_eq!(loaded.value(), 42);
    }

    #[test]
    fn test_counter_increment_merges() {
        // Create initial document
        let mut doc1 = AutoCommit::new();
        let counter = Counter::new(0);
        counter.save(&mut doc1, &ROOT, "count").unwrap();

        // Fork for concurrent editing
        let mut doc2 = doc1.fork().with_actor(ActorId::random());

        // User 1 increments by 5
        let mut counter1 = Counter::load(&doc1, &ROOT, "count").unwrap();
        counter1.increment(5);
        counter1.save(&mut doc1, &ROOT, "count").unwrap();

        // User 2 increments by 3 (concurrently)
        let mut counter2 = Counter::load(&doc2, &ROOT, "count").unwrap();
        counter2.increment(3);
        counter2.save(&mut doc2, &ROOT, "count").unwrap();

        // Merge
        doc1.merge(&mut doc2).unwrap();

        // Both increments should be preserved
        let merged = Counter::load(&doc1, &ROOT, "count").unwrap();
        assert_eq!(
            merged.value(),
            8,
            "Both increments should merge: 0 + 5 + 3 = 8"
        );
    }

    #[test]
    fn test_counter_arithmetic_operators() {
        let mut counter = Counter::new(10);

        counter += 5;
        assert_eq!(counter.value(), 15);

        counter -= 3;
        assert_eq!(counter.value(), 12);

        let counter2 = counter + 8;
        assert_eq!(counter2.value(), 20);
    }
}