Atomically increment an i64 value by `delta`. Returns the new value. - If key doesn't exist: initializes to 0, adds delta, returns delta. - If value is not a MessagePack integer: returns `TypeMismatch`. - On i64 overflow: returns `Overflow` (never wraps silently). - TTL behavior: if `ttl_ms > 0` and key is new, sets TTL. If key exists and `ttl_ms > 0`, resets TTL. If `ttl_ms == 0`, preserves.
(
&mut self,
tenant_id: u64,
collection: &str,
key: &[u8],
delta: i64,
ttl_ms: u64,
now_ms: u64,
)
| 38 | /// - TTL behavior: if `ttl_ms > 0` and key is new, sets TTL. |
| 39 | /// If key exists and `ttl_ms > 0`, resets TTL. If `ttl_ms == 0`, preserves. |
| 40 | pub fn incr( |
| 41 | &mut self, |
| 42 | tenant_id: u64, |
| 43 | collection: &str, |
| 44 | key: &[u8], |
| 45 | delta: i64, |
| 46 | ttl_ms: u64, |
| 47 | now_ms: u64, |
| 48 | ) -> Result<i64, AtomicError> { |
| 49 | let tkey = table_key(tenant_id, collection); |
| 50 | let table = self.ensure_table(tkey, tenant_id, collection); |
| 51 | |
| 52 | let current = table.get(key, now_ms).map(|v| v.to_vec()); |
| 53 | let old_i64 = match ¤t { |
| 54 | None => 0i64, |
| 55 | Some(bytes) => decode_msgpack_i64(bytes)?, |
| 56 | }; |
| 57 | |
| 58 | let new_i64 = old_i64.checked_add(delta).ok_or(AtomicError::Overflow)?; |
| 59 | |
| 60 | // If value is a map (typed KV entry), update the numeric field in-place. |
| 61 | let new_bytes = if let Some(ref cur) = current |
| 62 | && let Ok(nodedb_types::Value::Object(mut map)) = nodedb_types::value_from_msgpack(cur) |
| 63 | && map.len() > 1 |
| 64 | { |
| 65 | // Find and update the first numeric field. |
| 66 | let mut updated = false; |
| 67 | for (k, v) in map.iter_mut() { |
| 68 | if k == "key" { |
| 69 | continue; |
| 70 | } |
| 71 | if matches!( |
| 72 | v, |
| 73 | nodedb_types::Value::Integer(_) | nodedb_types::Value::Float(_) |
| 74 | ) { |
| 75 | *v = nodedb_types::Value::Integer(new_i64); |
| 76 | updated = true; |
| 77 | break; |
| 78 | } |
| 79 | } |
| 80 | if updated { |
| 81 | nodedb_types::value_to_msgpack(&nodedb_types::Value::Object(map)) |
| 82 | .unwrap_or_else(|_| zerompk::to_msgpack_vec(&new_i64).expect("i64 serializes")) |
| 83 | } else { |
| 84 | zerompk::to_msgpack_vec(&new_i64).expect("i64 always serializes") |
| 85 | } |
| 86 | } else { |
| 87 | zerompk::to_msgpack_vec(&new_i64).expect("i64 always serializes") |
| 88 | }; |
| 89 | self.atomic_put( |
| 90 | tenant_id, |
| 91 | collection, |
| 92 | tkey, |
| 93 | key, |
| 94 | &new_bytes, |
| 95 | ttl_ms, |
| 96 | now_ms, |
| 97 | current.is_none(), |