1use crate::state::{Durable, Kind, now_ms};
9use crate::store::StoreError;
10use serde::{Deserialize, Serialize};
11use serde_json::{Value, json};
12use std::collections::BTreeMap;
13
14pub const INDEX_ID: &str = "_index";
16
17#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
19pub struct Record {
20 pub value: Value,
21 pub ts: u64,
22 #[serde(default, skip_serializing_if = "Option::is_none")]
23 pub ttl_ms: Option<u64>,
24 #[serde(default, skip_serializing_if = "Option::is_none")]
25 pub by: Option<String>,
26}
27
28impl Record {
29 pub fn expired(&self, now: u64) -> bool {
30 self.ttl_ms
31 .is_some_and(|ttl| now >= self.ts.saturating_add(ttl))
32 }
33 pub fn meta(&self) -> Value {
34 json!({"ts": self.ts, "ttl_ms": self.ttl_ms, "by": self.by})
35 }
36}
37
38pub struct Memory {
40 max_value_bytes: usize,
41 list_default_limit: usize,
42 index: Option<BTreeMap<String, u64>>,
44 probed: bool,
45}
46
47impl Memory {
48 pub fn new(max_value_bytes: usize, list_default_limit: usize) -> Memory {
49 Memory {
50 max_value_bytes,
51 list_default_limit,
52 index: None,
53 probed: false,
54 }
55 }
56
57 pub fn check_key(key: &str) -> Result<(), String> {
59 if key.is_empty() || key.len() > 256 {
60 return Err("memory key must be 1..=256 chars".into());
61 }
62 if key.starts_with('_') {
63 return Err("memory keys starting with '_' are reserved".into());
64 }
65 if key.chars().any(char::is_whitespace) {
66 return Err("memory key must not contain whitespace".into());
67 }
68 Ok(())
69 }
70
71 fn probe(&mut self, d: &Durable) {
72 if self.probed {
73 return;
74 }
75 self.probed = true;
76 match d.list(Kind::Memory) {
77 Ok(_) => self.index = None,
78 Err(StoreError::Unsupported(_)) => {
79 let mut idx = BTreeMap::new();
80 if let Ok(Some(env)) = d.get(Kind::Memory, INDEX_ID)
81 && let Some(m) = env.state.get("keys").and_then(Value::as_object)
82 {
83 for (k, v) in m {
84 idx.insert(k.clone(), v.as_u64().unwrap_or(0));
85 }
86 }
87 self.index = Some(idx);
88 }
89 Err(_) => {}
90 }
91 }
92
93 fn write_index(&self, d: &Durable) -> Result<(), StoreError> {
94 if let Some(idx) = &self.index {
95 d.put(Kind::Memory, INDEX_ID, json!({"keys": idx}), None)?;
96 }
97 Ok(())
98 }
99
100 pub fn set(
102 &mut self,
103 d: &Durable,
104 key: &str,
105 value: Value,
106 ttl_ms: Option<u64>,
107 by: Option<&str>,
108 ) -> Result<Value, String> {
109 Self::check_key(key)?;
110 let bytes = value.to_string().len();
111 if bytes > self.max_value_bytes {
112 return Err(format!(
113 "memory value is {bytes} bytes; memory.max_value_bytes is {}",
114 self.max_value_bytes
115 ));
116 }
117 self.probe(d);
118 let rec = Record {
119 value,
120 ts: now_ms(),
121 ttl_ms,
122 by: by.map(str::to_string),
123 };
124 d.put(
125 Kind::Memory,
126 key,
127 serde_json::to_value(&rec).unwrap_or(Value::Null),
128 None,
129 )
130 .map_err(|e| e.to_string())?;
131 if let Some(idx) = &mut self.index {
132 idx.insert(key.to_string(), rec.ts);
133 self.write_index(d).map_err(|e| e.to_string())?;
134 }
135 Ok(json!({"ok": true, "key": key, "meta": rec.meta()}))
136 }
137
138 pub fn get(&mut self, d: &Durable, key: &str) -> Result<Value, String> {
140 Self::check_key(key)?;
141 match d.get(Kind::Memory, key).map_err(|e| e.to_string())? {
142 None => Ok(json!({"found": false, "key": key})),
143 Some(env) => {
144 let rec: Record = serde_json::from_value(env.state)
145 .map_err(|e| format!("memory record {key}: {e}"))?;
146 if rec.expired(now_ms()) {
147 let _ = self.delete(d, key);
148 return Ok(json!({"found": false, "key": key, "expired": true}));
149 }
150 Ok(json!({"found": true, "key": key, "value": rec.value, "meta": rec.meta()}))
151 }
152 }
153 }
154
155 pub fn delete(&mut self, d: &Durable, key: &str) -> Result<Value, String> {
157 Self::check_key(key)?;
158 self.probe(d);
159 d.delete(Kind::Memory, key).map_err(|e| e.to_string())?;
160 if let Some(idx) = &mut self.index {
161 idx.remove(key);
162 self.write_index(d).map_err(|e| e.to_string())?;
163 }
164 Ok(json!({"ok": true, "key": key}))
165 }
166
167 pub fn list(
169 &mut self,
170 d: &Durable,
171 prefix: Option<&str>,
172 limit: Option<usize>,
173 ) -> Result<Value, String> {
174 self.probe(d);
175 let limit = limit.unwrap_or(self.list_default_limit).max(1);
176 let prefix = prefix.unwrap_or("");
177 let mut keys: Vec<Value> = match &self.index {
178 Some(idx) => idx
179 .iter()
180 .filter(|(k, _)| k.starts_with(prefix))
181 .map(|(k, ts)| json!({"key": k, "ts": ts}))
182 .collect(),
183 None => {
184 let listed = d.list(Kind::Memory).map_err(|e| e.to_string())?;
185 let mut out = Vec::new();
186 for ks in listed {
187 let Some((_, id)) = crate::store::parse_key(d.prefix(), d.instance(), &ks.key)
188 else {
189 continue;
190 };
191 if id == INDEX_ID || !id.starts_with(prefix) {
192 continue;
193 }
194 out.push(json!({"key": id, "seq": ks.seq}));
195 }
196 out
197 }
198 };
199 keys.sort_by(|a, b| a["key"].as_str().cmp(&b["key"].as_str()));
200 let truncated = keys.len() > limit;
201 keys.truncate(limit);
202 Ok(json!({"keys": keys, "truncated": truncated}))
203 }
204}
205
206#[cfg(test)]
207mod tests {
208 use super::*;
209 use crate::state::Policy;
210 use crate::store::memory::MemoryStore;
211 use std::sync::Arc;
212
213 #[test]
214 fn set_get_list_delete_ttl_and_caps() {
215 let mem = Arc::new(MemoryStore::new());
216 let d = Durable::new(mem, "agentd", "i", Policy::default(), None);
217 let mut m = Memory::new(64, 10);
218 m.set(&d, "user/name", json!("andrii"), None, Some("root"))
219 .unwrap();
220 m.set(&d, "user/tz", json!("Europe/Kyiv"), Some(1), None)
221 .unwrap();
222 m.set(&d, "other", json!({"a": 1}), None, None).unwrap();
223 let g = m.get(&d, "user/name").unwrap();
224 assert_eq!(g["found"], json!(true));
225 assert_eq!(g["value"], json!("andrii"));
226 assert_eq!(g["meta"]["by"], json!("root"));
227 std::thread::sleep(std::time::Duration::from_millis(3));
228 let g = m.get(&d, "user/tz").unwrap();
229 assert_eq!(g["found"], json!(false), "expired: {g}");
230 assert_eq!(g["expired"], json!(true));
231 let l = m.list(&d, Some("user/"), None).unwrap();
232 assert_eq!(l["keys"].as_array().unwrap().len(), 1, "{l}");
233 assert_eq!(l["keys"][0]["key"], json!("user/name"));
234 let all = m.list(&d, None, Some(1)).unwrap();
235 assert_eq!(all["truncated"], json!(true));
236 m.delete(&d, "user/name").unwrap();
237 assert_eq!(m.get(&d, "user/name").unwrap()["found"], json!(false));
238 assert!(
240 m.set(&d, "big", json!("x".repeat(100)), None, None)
241 .is_err()
242 );
243 assert!(m.set(&d, "_reserved", json!(1), None, None).is_err());
244 assert!(m.set(&d, "has space", json!(1), None, None).is_err());
245 assert!(m.get(&d, "").is_err());
246 }
247
248 #[test]
249 fn index_record_is_kept_when_the_store_cannot_list() {
250 struct NoList(MemoryStore);
252 impl crate::store::Store for NoList {
253 fn put(
254 &self,
255 k: &str,
256 s: u64,
257 e: &Value,
258 ) -> Result<crate::store::PutOutcome, StoreError> {
259 self.0.put(k, s, e)
260 }
261 fn get(&self, k: &str, s: Option<u64>) -> Result<Option<Value>, StoreError> {
262 self.0.get(k, s)
263 }
264 fn list(&self, _p: &str) -> Result<Vec<crate::store::KeySeq>, StoreError> {
265 Err(StoreError::Unsupported("list"))
266 }
267 fn delete(&self, k: &str) -> Result<(), StoreError> {
268 self.0.delete(k)
269 }
270 fn kind(&self) -> &'static str {
271 "nolist"
272 }
273 }
274 let d = Durable::new(
275 Arc::new(NoList(MemoryStore::new())),
276 "agentd",
277 "i",
278 Policy::default(),
279 None,
280 );
281 let mut m = Memory::new(1024, 10);
282 m.set(&d, "a", json!(1), None, None).unwrap();
283 m.set(&d, "b", json!(2), None, None).unwrap();
284 let idx = d
285 .get(Kind::Memory, INDEX_ID)
286 .unwrap()
287 .expect("index record");
288 assert!(idx.state["keys"].get("a").is_some());
289 assert_eq!(
290 m.list(&d, None, None).unwrap()["keys"]
291 .as_array()
292 .unwrap()
293 .len(),
294 2
295 );
296 m.delete(&d, "a").unwrap();
297 let mut m2 = Memory::new(1024, 10);
299 let l = m2.list(&d, None, None).unwrap();
300 assert_eq!(l["keys"].as_array().unwrap().len(), 1);
301 assert_eq!(l["keys"][0]["key"], json!("b"));
302 }
303}