flaron-sdk 0.99.0

//! Plasma - cross-edge CRDT key/value store.
//!
//! Plasma replicates state across every edge in the Flaron mesh using a
//! gossip-backed CRDT layer. Use it for state that must be visible from any
//! edge: counters, presence, leaderboards, feature flags, ephemeral session
//! coordination. For per-site state with TTLs use [`crate::spark`] instead.
//!
//! ## Operations
//!
//! * [`get`] / [`set`] / [`delete`] - last-writer-wins register semantics.
//! * [`increment`] / [`decrement`] - PN-counter CRDT, returns the new value.
//! * [`list`] - enumerate all keys for this site.
//!
//! ## Capability gate
//!
//! Writes (`set`, `delete`, `increment`, `decrement`) require the flare's
//! `WritesPlasmaKV` capability. Without it, every write returns
//! [`PlasmaError::NoCapability`].

use crate::{ffi, mem};

/// Errors returned by Plasma write operations.
///
/// Codes match the `plasmaErr*` constants in
/// `internal/corona/hostapi_kv.go`.
#[derive(Debug, thiserror::Error)]
pub enum PlasmaError {
    /// Plasma is not configured on this edge node.
    #[error("plasma: not available")]
    NotAvailable,

    /// Per-invocation write count exceeded.
    #[error("plasma: write limit exceeded")]
    WriteLimit,

    /// Value exceeds the host's per-key size cap.
    #[error("plasma: value too large")]
    TooLarge,

    /// Key failed validation.
    #[error("plasma: invalid key")]
    BadKey,

    /// Flare lacks the `WritesPlasmaKV` capability.
    #[error("plasma: no capability")]
    NoCapability,

    /// Internal host error - see edge logs.
    #[error("plasma: internal error")]
    Internal,

    /// Unknown error code returned by the host.
    #[error("plasma: unknown error code {0}")]
    Unknown(i32),
}

impl PlasmaError {
    fn from_code(code: i32) -> Self {
        match code {
            1 => Self::NotAvailable,
            2 => Self::WriteLimit,
            3 => Self::TooLarge,
            4 => Self::BadKey,
            5 => Self::NoCapability,
            6 => Self::Internal,
            other => Self::Unknown(other),
        }
    }
}

/// Get a value from Plasma.
///
/// Returns `None` if the key does not exist, the read limit was hit, or
/// Plasma is not configured on this edge.
pub fn get(key: &str) -> Option<Vec<u8>> {
    let (key_ptr, key_len) = mem::host_arg_str(key);
    let result = unsafe { ffi::plasma_get(key_ptr, key_len) };
    // SAFETY: host writes the raw value bytes into the bump arena.
    unsafe { mem::read_packed_bytes(result) }
}

/// Convenience: get a value as a UTF-8 string. Returns `None` if the key is
/// missing or the bytes are not valid UTF-8.
pub fn get_string(key: &str) -> Option<String> {
    let bytes = get(key)?;
    String::from_utf8(bytes).ok()
}

/// Write a value to Plasma. Last-writer-wins.
///
/// Requires `WritesPlasmaKV` capability.
pub fn set(key: &str, value: &[u8]) -> Result<(), PlasmaError> {
    let (key_ptr, key_len) = mem::host_arg_str(key);
    let (val_ptr, val_len) = mem::host_arg_bytes(value);
    let code = unsafe { ffi::plasma_set(key_ptr, key_len, val_ptr, val_len) };
    if code == 0 {
        Ok(())
    } else {
        Err(PlasmaError::from_code(code))
    }
}

/// Delete a key from Plasma. Idempotent.
///
/// Requires `WritesPlasmaKV` capability.
pub fn delete(key: &str) -> Result<(), PlasmaError> {
    let (key_ptr, key_len) = mem::host_arg_str(key);
    let code = unsafe { ffi::plasma_delete(key_ptr, key_len) };
    if code == 0 {
        Ok(())
    } else {
        Err(PlasmaError::from_code(code))
    }
}

/// Atomically add `delta` to a PN-counter at `key` and return the new value.
///
/// Counters are CRDTs - concurrent increments from multiple edges merge
/// commutatively without coordination. Pass a negative delta to decrement,
/// or use [`decrement`] for clarity.
///
/// Returns `None` if the operation failed (capability missing, write limit,
/// host error).
pub fn increment(key: &str, delta: i64) -> Option<i64> {
    let (key_ptr, key_len) = mem::host_arg_str(key);
    let result = unsafe { ffi::plasma_increment(key_ptr, key_len, delta) };
    // SAFETY: host writes 8 LE bytes representing the new counter value.
    let bytes = unsafe { mem::read_packed_bytes(result) }?;
    if bytes.len() != 8 {
        return None;
    }
    let arr: [u8; 8] = [
        bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
    ];
    Some(i64::from_le_bytes(arr))
}

/// Atomically subtract `delta` from a PN-counter at `key` and return the new
/// value.
pub fn decrement(key: &str, delta: i64) -> Option<i64> {
    let (key_ptr, key_len) = mem::host_arg_str(key);
    let result = unsafe { ffi::plasma_decrement(key_ptr, key_len, delta) };
    let bytes = unsafe { mem::read_packed_bytes(result) }?;
    if bytes.len() != 8 {
        return None;
    }
    let arr: [u8; 8] = [
        bytes[0], bytes[1], bytes[2], bytes[3], bytes[4], bytes[5], bytes[6], bytes[7],
    ];
    Some(i64::from_le_bytes(arr))
}

/// List all keys in this site's Plasma store.
///
/// Returns an empty `Vec` if Plasma is not configured or the read limit was
/// hit.
pub fn list() -> Vec<String> {
    let result = unsafe { ffi::plasma_list() };
    let Some(json_bytes) = (unsafe { mem::read_packed_bytes(result) }) else {
        return Vec::new();
    };
    serde_json::from_slice(&json_bytes).unwrap_or_default()
}

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

    #[test]
    fn get_returns_stored_value() {
        test_host::reset();
        test_host::with_mock(|m| {
            m.plasma_store.insert("k".into(), b"value".to_vec());
        });
        assert_eq!(get("k"), Some(b"value".to_vec()));
    }

    #[test]
    fn get_none_for_missing() {
        test_host::reset();
        assert!(get("missing").is_none());
    }

    #[test]
    fn get_string_decodes_utf8() {
        test_host::reset();
        test_host::with_mock(|m| {
            m.plasma_store
                .insert("k".into(), "héllo".as_bytes().to_vec());
        });
        assert_eq!(get_string("k").as_deref(), Some("héllo"));
    }

    #[test]
    fn set_stores_value() {
        test_host::reset();
        set("k", b"v").expect("set should succeed");
        assert_eq!(
            test_host::read_mock(|m| m.plasma_store.get("k").cloned()),
            Some(b"v".to_vec())
        );
    }

    #[test]
    fn set_maps_error_codes() {
        for (code, expected_disc) in [
            (1, PlasmaError::NotAvailable),
            (2, PlasmaError::WriteLimit),
            (3, PlasmaError::TooLarge),
            (4, PlasmaError::BadKey),
            (5, PlasmaError::NoCapability),
            (6, PlasmaError::Internal),
        ] {
            test_host::reset();
            test_host::with_mock(|m| m.plasma_set_error = code);
            let err = set("k", b"v").unwrap_err();
            assert!(
                std::mem::discriminant(&err) == std::mem::discriminant(&expected_disc),
                "code {} should map to {:?}, got {:?}",
                code,
                expected_disc,
                err,
            );
        }
    }

    #[test]
    fn set_unknown_error_code() {
        test_host::reset();
        test_host::with_mock(|m| m.plasma_set_error = 42);
        match set("k", b"v").unwrap_err() {
            PlasmaError::Unknown(42) => {}
            other => panic!("expected Unknown(42), got {:?}", other),
        }
    }

    #[test]
    fn delete_removes_value() {
        test_host::reset();
        test_host::with_mock(|m| {
            m.plasma_store.insert("k".into(), b"v".to_vec());
        });
        delete("k").unwrap();
        assert!(test_host::read_mock(|m| m.plasma_store.is_empty()));
    }

    #[test]
    fn increment_returns_new_counter_value() {
        test_host::reset();
        let v1 = increment("c", 5).expect("first increment");
        assert_eq!(v1, 5);
        let v2 = increment("c", 3).expect("second increment");
        assert_eq!(v2, 8);
        let v3 = increment("c", -2).expect("negative delta");
        assert_eq!(v3, 6);
    }

    #[test]
    fn decrement_returns_new_counter_value() {
        test_host::reset();
        increment("c", 10).unwrap();
        let v = decrement("c", 4).expect("decrement");
        assert_eq!(v, 6);
    }

    #[test]
    fn increment_captures_args() {
        test_host::reset();
        increment("counter", 7).unwrap();
        let captured = test_host::read_mock(|m| m.last_plasma_increment.clone());
        assert_eq!(captured, Some(("counter".into(), 7)));
    }

    #[test]
    fn increment_returns_none_on_error() {
        test_host::reset();
        test_host::with_mock(|m| m.plasma_increment_error = true);
        assert!(increment("c", 1).is_none());
    }

    #[test]
    fn list_returns_keys() {
        test_host::reset();
        test_host::with_mock(|m| {
            m.plasma_store.insert("a".into(), b"1".to_vec());
            m.plasma_store.insert("b".into(), b"2".to_vec());
        });
        let mut keys = list();
        keys.sort();
        assert_eq!(keys, vec!["a".to_string(), "b".to_string()]);
    }
}