awaken_stores/memory/
mod.rs

1//! In-memory storage backend for testing and local development.
2use std::collections::HashMap;
3use std::sync::atomic::{AtomicU64, Ordering};
4
5use async_trait::async_trait;
6use awaken_server_contract::contract::config_store::{
7    ConfigChangeEvent, ConfigChangeKind, ConfigChangeNotifier, ConfigChangeSubscriber, ConfigStore,
8    extract_meta_revision,
9};
10use awaken_server_contract::contract::message::{
11    Message, PendingMessageRecord, strip_unpaired_tool_calls_from_view,
12};
13use awaken_server_contract::contract::profile_store::{ProfileEntry, ProfileOwner, ProfileStore};
14use awaken_server_contract::contract::storage::{
15    MessagePage, MessageQuery, RunPage, RunQuery, RunRecord, RunStore, StorageError, ThreadPage,
16    ThreadQuery, ThreadRunStore, ThreadStore, checkpoint_parent_thread_id, message_append,
17    paginate_message_records, paginate_threads,
18};
19use awaken_server_contract::thread::{Thread, normalize_lineage_id};
20use serde_json::Value;
21use tokio::sync::RwLock;
22
23use crate::message_validation::{validate_committed_message_records, validate_committed_messages};
24
25mod pending;
26
27/// In-memory storage implementing all four store traits.
28///
29/// Uses `tokio::sync::RwLock` for async-safe concurrent access.
30/// Data lives only in memory and is lost when the store is dropped.
31#[derive(Debug)]
32pub struct InMemoryStore {
33    pub(crate) threads: RwLock<HashMap<String, Thread>>,
34    pub(crate) runs: RwLock<HashMap<String, RunRecord>>,
35    /// Monotonic per-run insertion sequence used as a `latest_run` tie-break.
36    /// `created_at`/`updated_at` are second-precision, so multiple runs in
37    /// the same wall-clock second otherwise yield non-deterministic
38    /// ordering and silently break agent-binding inference.
39    pub(crate) run_insertion: RwLock<HashMap<String, u64>>,
40    pub(crate) run_seq: AtomicU64,
41    /// Thread ID -> ordered messages (single source of truth).
42    pub(crate) messages: RwLock<HashMap<String, Vec<Message>>>,
43    /// Thread ID -> ordered pending messages not yet consumed by a run.
44    pub(crate) pending_messages: RwLock<HashMap<String, Vec<PendingMessageRecord>>>,
45    /// Profile entries keyed by (owner, key).
46    profiles: RwLock<HashMap<ProfileOwner, HashMap<String, ProfileEntry>>>,
47    /// Config entries keyed by namespace then ID.
48    configs: RwLock<HashMap<String, HashMap<String, Value>>>,
49    /// Thread ID -> thread-scoped persisted state (cross-run).
50    pub(crate) thread_states: RwLock<HashMap<String, awaken_server_contract::PersistedState>>,
51    /// Broadcast sender for config change notifications.
52    config_change_tx: tokio::sync::broadcast::Sender<ConfigChangeEvent>,
53}
54
55impl InMemoryStore {
56    /// Create a new empty in-memory store.
57    pub fn new() -> Self {
58        let (config_change_tx, _) = tokio::sync::broadcast::channel(256);
59        Self {
60            threads: RwLock::new(HashMap::new()),
61            runs: RwLock::new(HashMap::new()),
62            run_insertion: RwLock::new(HashMap::new()),
63            run_seq: AtomicU64::new(0),
64            messages: RwLock::new(HashMap::new()),
65            pending_messages: RwLock::new(HashMap::new()),
66            profiles: RwLock::new(HashMap::new()),
67            configs: RwLock::new(HashMap::new()),
68            thread_states: RwLock::new(HashMap::new()),
69            config_change_tx,
70        }
71    }
72
73    fn next_run_seq(&self) -> u64 {
74        self.run_seq.fetch_add(1, Ordering::Relaxed)
75    }
76}
77
78impl Default for InMemoryStore {
79    fn default() -> Self {
80        Self::new()
81    }
82}
83
84fn validate_thread_hierarchy_map(
85    threads: &HashMap<String, Thread>,
86    thread_id: &str,
87    parent_thread_id: Option<&str>,
88) -> Result<(), StorageError> {
89    let Some(parent_thread_id) = normalize_lineage_id(parent_thread_id) else {
90        return Ok(());
91    };
92    if parent_thread_id == thread_id {
93        return Err(StorageError::Validation(format!(
94            "thread '{thread_id}' cannot parent itself"
95        )));
96    }
97
98    let root_parent_thread_id = parent_thread_id.clone();
99    let mut current_thread_id = parent_thread_id;
100    let mut visited = std::collections::HashSet::from([thread_id.to_owned()]);
101
102    loop {
103        if !visited.insert(current_thread_id.clone()) {
104            return Err(StorageError::Validation(format!(
105                "thread hierarchy cycle detected at '{current_thread_id}'"
106            )));
107        }
108
109        let Some(thread) = threads.get(&current_thread_id) else {
110            let message = if current_thread_id == root_parent_thread_id {
111                format!("parent thread not found: {root_parent_thread_id}")
112            } else {
113                format!("thread hierarchy references missing ancestor '{current_thread_id}'")
114            };
115            return Err(StorageError::Validation(message));
116        };
117
118        let Some(next_parent_thread_id) = normalize_lineage_id(thread.parent_thread_id.as_deref())
119        else {
120            return Ok(());
121        };
122        current_thread_id = next_parent_thread_id;
123    }
124}
125
126fn collect_child_ids(threads: &HashMap<String, Thread>, parent_thread_id: &str) -> Vec<String> {
127    let mut child_ids: Vec<String> = threads
128        .values()
129        .filter(|thread| thread.parent_thread_id.as_deref() == Some(parent_thread_id))
130        .map(|thread| thread.id.clone())
131        .collect();
132    child_ids.sort();
133    child_ids
134}
135
136// ── ThreadStore ─────────────────────────────────────────────────────
137
138#[async_trait]
139impl ThreadStore for InMemoryStore {
140    async fn load_thread(&self, thread_id: &str) -> Result<Option<Thread>, StorageError> {
141        let guard = self.threads.read().await;
142        Ok(guard.get(thread_id).cloned())
143    }
144
145    async fn save_thread(&self, thread: &Thread) -> Result<(), StorageError> {
146        let mut normalized = thread.clone();
147        normalized.normalize_lineage();
148        normalized.validate_for_persist()?;
149        let mut guard = self.threads.write().await;
150        guard.insert(normalized.id.clone(), normalized);
151        Ok(())
152    }
153
154    async fn save_thread_validated(&self, thread: &Thread) -> Result<(), StorageError> {
155        let mut normalized = thread.clone();
156        normalized.normalize_lineage();
157        normalized.validate_for_persist()?;
158        let mut guard = self.threads.write().await;
159        validate_thread_hierarchy_map(
160            &guard,
161            &normalized.id,
162            normalized.parent_thread_id.as_deref(),
163        )?;
164        guard.insert(normalized.id.clone(), normalized);
165        Ok(())
166    }
167
168    async fn delete_thread(&self, thread_id: &str) -> Result<(), StorageError> {
169        let mut threads = self.threads.write().await;
170        let mut messages = self.messages.write().await;
171        let mut pending_messages = self.pending_messages.write().await;
172        threads.remove(thread_id);
173        messages.remove(thread_id);
174        pending_messages.remove(thread_id);
175        self.thread_states.write().await.remove(thread_id);
176        Ok(())
177    }
178
179    async fn save_thread_state(
180        &self,
181        thread_id: &str,
182        state: &awaken_server_contract::PersistedState,
183    ) -> Result<(), StorageError> {
184        self.thread_states
185            .write()
186            .await
187            .insert(thread_id.to_string(), state.clone());
188        Ok(())
189    }
190
191    async fn load_thread_state(
192        &self,
193        thread_id: &str,
194    ) -> Result<Option<awaken_server_contract::PersistedState>, StorageError> {
195        Ok(self.thread_states.read().await.get(thread_id).cloned())
196    }
197
198    async fn delete_thread_with_strategy(
199        &self,
200        thread_id: &str,
201        strategy: awaken_server_contract::contract::storage::ChildThreadDeleteStrategy,
202    ) -> Result<(), StorageError> {
203        let mut threads = self.threads.write().await;
204        let mut messages = self.messages.write().await;
205        let mut pending_messages = self.pending_messages.write().await;
206        if !threads.contains_key(thread_id) {
207            return Err(StorageError::NotFound(thread_id.to_owned()));
208        }
209
210        match strategy {
211            awaken_server_contract::contract::storage::ChildThreadDeleteStrategy::Reject => {
212                if !collect_child_ids(&threads, thread_id).is_empty() {
213                    return Err(StorageError::Validation(format!(
214                        "thread '{thread_id}' has child threads; choose 'detach' or 'cascade'"
215                    )));
216                }
217                threads.remove(thread_id);
218                messages.remove(thread_id);
219                pending_messages.remove(thread_id);
220            }
221            awaken_server_contract::contract::storage::ChildThreadDeleteStrategy::Detach => {
222                let updated_at = current_millis();
223                for child_id in collect_child_ids(&threads, thread_id) {
224                    if let Some(child) = threads.get_mut(&child_id) {
225                        child.parent_thread_id = None;
226                        child.normalize_lineage();
227                        child.metadata.updated_at = Some(updated_at);
228                    }
229                }
230                threads.remove(thread_id);
231                messages.remove(thread_id);
232                pending_messages.remove(thread_id);
233            }
234            awaken_server_contract::contract::storage::ChildThreadDeleteStrategy::Cascade => {
235                let mut visited = std::collections::HashSet::new();
236                let mut stack = vec![(thread_id.to_owned(), false)];
237                let mut delete_order = Vec::new();
238
239                while let Some((current_thread_id, expanded)) = stack.pop() {
240                    if expanded {
241                        delete_order.push(current_thread_id);
242                        continue;
243                    }
244
245                    if !visited.insert(current_thread_id.clone()) {
246                        return Err(StorageError::Validation(format!(
247                            "thread hierarchy cycle detected while deleting '{thread_id}'"
248                        )));
249                    }
250
251                    stack.push((current_thread_id.clone(), true));
252                    for child_id in collect_child_ids(&threads, &current_thread_id)
253                        .into_iter()
254                        .rev()
255                    {
256                        stack.push((child_id, false));
257                    }
258                }
259
260                for id in delete_order {
261                    threads.remove(&id);
262                    messages.remove(&id);
263                    pending_messages.remove(&id);
264                }
265            }
266        }
267
268        Ok(())
269    }
270
271    async fn list_threads(&self, offset: usize, limit: usize) -> Result<Vec<String>, StorageError> {
272        let guard = self.threads.read().await;
273        let mut threads: Vec<Thread> = guard.values().cloned().collect();
274        awaken_server_contract::contract::storage::sort_threads_by_recent_activity(&mut threads);
275        Ok(threads
276            .into_iter()
277            .skip(offset)
278            .take(limit)
279            .map(|thread| thread.id)
280            .collect())
281    }
282
283    async fn list_threads_query(&self, query: &ThreadQuery) -> Result<ThreadPage, StorageError> {
284        let guard = self.threads.read().await;
285        let threads: Vec<Thread> = guard.values().cloned().collect();
286        Ok(paginate_threads(threads, query))
287    }
288
289    async fn load_messages(&self, thread_id: &str) -> Result<Option<Vec<Message>>, StorageError> {
290        let guard = self.messages.read().await;
291        let Some(mut messages) = guard.get(thread_id).cloned() else {
292            return Ok(None);
293        };
294        validate_committed_messages(&messages)?;
295        strip_unpaired_tool_calls_from_view(&mut messages);
296        Ok(Some(messages))
297    }
298
299    async fn load_committed_messages(
300        &self,
301        thread_id: &str,
302    ) -> Result<Option<Vec<Message>>, StorageError> {
303        let guard = self.messages.read().await;
304        let Some(messages) = guard.get(thread_id).cloned() else {
305            return Ok(None);
306        };
307        validate_committed_messages(&messages)?;
308        Ok(Some(messages))
309    }
310
311    async fn list_message_records(
312        &self,
313        thread_id: &str,
314        query: &MessageQuery,
315    ) -> Result<MessagePage, StorageError> {
316        let guard = self.messages.read().await;
317        let Some(messages) = guard.get(thread_id) else {
318            return Ok(MessagePage::empty());
319        };
320        validate_committed_messages(messages)?;
321        let mut messages = messages.clone();
322        strip_unpaired_tool_calls_from_view(&mut messages);
323        let records = messages
324            .iter()
325            .cloned()
326            .enumerate()
327            .map(|(index, message)| {
328                awaken_server_contract::contract::message::MessageRecord::from_message(
329                    thread_id.to_owned(),
330                    index as u64 + 1,
331                    message,
332                )
333            })
334            .collect::<Vec<_>>();
335        validate_committed_message_records(thread_id, &records)?;
336        Ok(paginate_message_records(records, query))
337    }
338
339    async fn save_messages(
340        &self,
341        thread_id: &str,
342        messages: &[Message],
343    ) -> Result<(), StorageError> {
344        validate_committed_messages(messages)?;
345        let mut guard = self.messages.write().await;
346        guard.insert(thread_id.to_owned(), messages.to_vec());
347        Ok(())
348    }
349
350    /// Atomic append: holds the messages write lock across the whole
351    /// read-modify-write, so concurrent writers (including separate `Mailbox`
352    /// instances sharing this store) never lose an append. Overrides the
353    /// non-atomic default `load → extend → save` (ADR-0042 D4/D5).
354    async fn append_message_records(
355        &self,
356        thread_id: &str,
357        messages: &[Message],
358    ) -> Result<Vec<awaken_server_contract::contract::message::MessageRecord>, StorageError> {
359        let mut guard = self.messages.write().await;
360        let existing = guard.entry(thread_id.to_owned()).or_default();
361        message_append::validate_append_only_delta(existing, messages)?;
362        let start_seq = existing.len() as u64 + 1;
363        existing.extend(messages.iter().cloned());
364        Ok(messages
365            .iter()
366            .cloned()
367            .enumerate()
368            .map(|(index, message)| {
369                awaken_server_contract::contract::message::MessageRecord::from_message(
370                    thread_id.to_owned(),
371                    start_seq + index as u64,
372                    message,
373                )
374            })
375            .collect())
376    }
377
378    async fn delete_messages(&self, thread_id: &str) -> Result<(), StorageError> {
379        let threads = self.threads.read().await;
380        if !threads.contains_key(thread_id) {
381            return Err(StorageError::NotFound(thread_id.to_owned()));
382        }
383        drop(threads);
384        let mut guard = self.messages.write().await;
385        guard.remove(thread_id);
386        Ok(())
387    }
388
389    async fn update_thread_metadata(
390        &self,
391        id: &str,
392        metadata: awaken_server_contract::thread::ThreadMetadata,
393    ) -> Result<(), StorageError> {
394        let mut guard = self.threads.write().await;
395        let thread = guard
396            .get_mut(id)
397            .ok_or_else(|| StorageError::NotFound(id.to_owned()))?;
398        thread.metadata = metadata;
399        Ok(())
400    }
401}
402
403// ── RunStore ────────────────────────────────────────────────────────
404
405#[async_trait]
406impl RunStore for InMemoryStore {
407    async fn create_run(&self, record: &RunRecord) -> Result<(), StorageError> {
408        record.validate_for_persist()?;
409        let mut guard = self.runs.write().await;
410        if guard.contains_key(&record.run_id) {
411            return Err(StorageError::AlreadyExists(record.run_id.clone()));
412        }
413        guard.insert(record.run_id.clone(), record.clone());
414        self.run_insertion
415            .write()
416            .await
417            .insert(record.run_id.clone(), self.next_run_seq());
418        Ok(())
419    }
420
421    async fn load_run(&self, run_id: &str) -> Result<Option<RunRecord>, StorageError> {
422        let guard = self.runs.read().await;
423        Ok(guard.get(run_id).cloned())
424    }
425
426    async fn latest_run(&self, thread_id: &str) -> Result<Option<RunRecord>, StorageError> {
427        let runs = self.runs.read().await;
428        let insertion = self.run_insertion.read().await;
429        // Tie-break on insertion sequence so that two runs that landed in
430        // the same wall-clock second still have a deterministic order.
431        Ok(runs
432            .values()
433            .filter(|r| r.thread_id == thread_id)
434            .max_by_key(|r| (r.updated_at, insertion.get(&r.run_id).copied().unwrap_or(0)))
435            .cloned())
436    }
437
438    async fn list_runs(&self, query: &RunQuery) -> Result<RunPage, StorageError> {
439        let guard = self.runs.read().await;
440        let mut filtered: Vec<RunRecord> = guard
441            .values()
442            .filter(|r| query.thread_id.as_deref().is_none_or(|t| r.thread_id == t))
443            .filter(|r| query.status.is_none_or(|s| r.status == s))
444            .filter(|r| query.matches_id_prefix(&r.thread_id))
445            .cloned()
446            .collect();
447        filtered.sort_by_key(|r| r.created_at);
448        let total = filtered.len();
449        let offset = query.offset.min(total);
450        let limit = query.limit.clamp(1, 200);
451        let items: Vec<RunRecord> = filtered.into_iter().skip(offset).take(limit).collect();
452        let has_more = offset + items.len() < total;
453        Ok(RunPage {
454            items,
455            total,
456            has_more,
457        })
458    }
459}
460
461// ── ThreadRunStore ──────────────────────────────────────────────────
462
463#[async_trait]
464impl ThreadRunStore for InMemoryStore {
465    fn thread_run_storage_identity(&self) -> Option<String> {
466        Some(format!("memory-thread-run::{:p}", self))
467    }
468
469    async fn checkpoint(
470        &self,
471        thread_id: &str,
472        messages: &[Message],
473        run: &RunRecord,
474    ) -> Result<(), StorageError> {
475        run.validate_for_persist()?;
476        validate_committed_messages(messages)?;
477        let now = current_millis();
478        let mut thread_guard = self.threads.write().await;
479        let existing_thread = thread_guard.get(thread_id).cloned();
480        validate_thread_hierarchy_map(
481            &thread_guard,
482            thread_id,
483            checkpoint_parent_thread_id(existing_thread.as_ref(), run),
484        )?;
485        let mut msg_guard = self.messages.write().await;
486        let mut run_guard = self.runs.write().await;
487        let mut thread = existing_thread.unwrap_or_else(|| Thread::with_id(thread_id));
488        thread.touch(now);
489        thread.apply_run_projection(run);
490        thread.normalize_lineage();
491        thread_guard.insert(thread_id.to_owned(), thread);
492        msg_guard.insert(thread_id.to_owned(), messages.to_vec());
493        run_guard.insert(run.run_id.clone(), run.clone());
494        // Refresh the insertion order on every checkpoint so the run that
495        // most recently advanced becomes the new `latest_run` even when
496        // updated_at ties with an earlier run on the same thread.
497        self.run_insertion
498            .write()
499            .await
500            .insert(run.run_id.clone(), self.next_run_seq());
501        Ok(())
502    }
503
504    /// Atomic, version-guarded committed append: holds the thread/message/run
505    /// write locks across the read-check-append-write so concurrent writers
506    /// (including separate `Mailbox` instances sharing this store) never lose
507    /// an append (ADR-0042 D5). A stale `expected_version` leaves all state
508    /// untouched.
509    async fn checkpoint_append(
510        &self,
511        thread_id: &str,
512        messages: &[Message],
513        expected_version: Option<u64>,
514        run: &RunRecord,
515    ) -> Result<u64, StorageError> {
516        run.validate_for_persist()?;
517        let now = current_millis();
518        let mut thread_guard = self.threads.write().await;
519        let existing_thread = thread_guard.get(thread_id).cloned();
520        validate_thread_hierarchy_map(
521            &thread_guard,
522            thread_id,
523            checkpoint_parent_thread_id(existing_thread.as_ref(), run),
524        )?;
525        let mut msg_guard = self.messages.write().await;
526        let mut run_guard = self.runs.write().await;
527        let actual = msg_guard
528            .get(thread_id)
529            .map(|messages| messages.len() as u64)
530            .unwrap_or(0);
531        if let Some(expected) = expected_version
532            && expected != actual
533        {
534            return Err(StorageError::VersionConflict { expected, actual });
535        }
536        let committed = msg_guard.entry(thread_id.to_owned()).or_default();
537        message_append::merge_checkpoint_append_messages(committed, messages)?;
538        let new_version = committed.len() as u64;
539        let mut thread = existing_thread.unwrap_or_else(|| Thread::with_id(thread_id));
540        thread.touch(now);
541        thread.apply_run_projection(run);
542        thread.normalize_lineage();
543        thread_guard.insert(thread_id.to_owned(), thread);
544        run_guard.insert(run.run_id.clone(), run.clone());
545        self.run_insertion
546            .write()
547            .await
548            .insert(run.run_id.clone(), self.next_run_seq());
549        Ok(new_version)
550    }
551}
552
553// ── ProfileStore ────────────────────────────────────────────────────
554
555use crate::current_millis;
556
557#[async_trait]
558impl ProfileStore for InMemoryStore {
559    async fn get(
560        &self,
561        owner: &ProfileOwner,
562        key: &str,
563    ) -> Result<Option<ProfileEntry>, StorageError> {
564        let guard = self.profiles.read().await;
565        Ok(guard.get(owner).and_then(|inner| inner.get(key)).cloned())
566    }
567
568    async fn set(&self, owner: &ProfileOwner, key: &str, value: Value) -> Result<(), StorageError> {
569        let mut guard = self.profiles.write().await;
570        let inner = guard.entry(owner.clone()).or_default();
571        inner.insert(
572            key.to_owned(),
573            ProfileEntry {
574                key: key.to_owned(),
575                value,
576                updated_at: current_millis(),
577            },
578        );
579        Ok(())
580    }
581
582    async fn delete(&self, owner: &ProfileOwner, key: &str) -> Result<(), StorageError> {
583        let mut guard = self.profiles.write().await;
584        if let Some(inner) = guard.get_mut(owner) {
585            inner.remove(key);
586        }
587        Ok(())
588    }
589
590    async fn list(&self, owner: &ProfileOwner) -> Result<Vec<ProfileEntry>, StorageError> {
591        let guard = self.profiles.read().await;
592        let mut entries: Vec<ProfileEntry> = guard
593            .get(owner)
594            .map(|inner| inner.values().cloned().collect())
595            .unwrap_or_default();
596        entries.sort_by(|a, b| a.key.cmp(&b.key));
597        Ok(entries)
598    }
599
600    async fn clear_owner(&self, owner: &ProfileOwner) -> Result<(), StorageError> {
601        let mut guard = self.profiles.write().await;
602        guard.remove(owner);
603        Ok(())
604    }
605}
606
607// ── ConfigStore ─────────────────────────────────────────────────────
608
609#[async_trait]
610impl ConfigStore for InMemoryStore {
611    async fn get(&self, namespace: &str, id: &str) -> Result<Option<Value>, StorageError> {
612        let guard = self.configs.read().await;
613        Ok(guard
614            .get(namespace)
615            .and_then(|entries| entries.get(id))
616            .cloned())
617    }
618
619    async fn list(
620        &self,
621        namespace: &str,
622        offset: usize,
623        limit: usize,
624    ) -> Result<Vec<(String, Value)>, StorageError> {
625        let guard = self.configs.read().await;
626        let Some(entries) = guard.get(namespace) else {
627            return Ok(Vec::new());
628        };
629        let mut items: Vec<_> = entries
630            .iter()
631            .map(|(id, value)| (id.clone(), value.clone()))
632            .collect();
633        items.sort_by(|left, right| left.0.cmp(&right.0));
634        Ok(items.into_iter().skip(offset).take(limit).collect())
635    }
636
637    async fn put(&self, namespace: &str, id: &str, value: &Value) -> Result<(), StorageError> {
638        let mut guard = self.configs.write().await;
639        guard
640            .entry(namespace.to_string())
641            .or_default()
642            .insert(id.to_string(), value.clone());
643        drop(guard);
644        let _ = self.config_change_tx.send(ConfigChangeEvent {
645            namespace: namespace.to_string(),
646            id: id.to_string(),
647            kind: ConfigChangeKind::Put,
648        });
649        Ok(())
650    }
651
652    async fn put_if_absent(
653        &self,
654        namespace: &str,
655        id: &str,
656        value: &Value,
657    ) -> Result<(), StorageError> {
658        let mut guard = self.configs.write().await;
659        let entries = guard.entry(namespace.to_string()).or_default();
660        if entries.contains_key(id) {
661            return Err(StorageError::AlreadyExists(format!("{namespace}/{id}")));
662        }
663        entries.insert(id.to_string(), value.clone());
664        drop(guard);
665        let _ = self.config_change_tx.send(ConfigChangeEvent {
666            namespace: namespace.to_string(),
667            id: id.to_string(),
668            kind: ConfigChangeKind::Put,
669        });
670        Ok(())
671    }
672
673    async fn delete(&self, namespace: &str, id: &str) -> Result<(), StorageError> {
674        let mut guard = self.configs.write().await;
675        if let Some(entries) = guard.get_mut(namespace) {
676            entries.remove(id);
677        }
678        drop(guard);
679        let _ = self.config_change_tx.send(ConfigChangeEvent {
680            namespace: namespace.to_string(),
681            id: id.to_string(),
682            kind: ConfigChangeKind::Delete,
683        });
684        Ok(())
685    }
686
687    async fn put_if_revision(
688        &self,
689        namespace: &str,
690        id: &str,
691        value: &Value,
692        expected_revision: u64,
693    ) -> Result<(), StorageError> {
694        let mut guard = self.configs.write().await;
695        let actual = guard
696            .get(namespace)
697            .and_then(|entries| entries.get(id))
698            .and_then(extract_meta_revision)
699            .unwrap_or(0);
700        if actual != expected_revision {
701            return Err(StorageError::VersionConflict {
702                expected: expected_revision,
703                actual,
704            });
705        }
706        guard
707            .entry(namespace.to_string())
708            .or_default()
709            .insert(id.to_string(), value.clone());
710        drop(guard);
711        let _ = self.config_change_tx.send(ConfigChangeEvent {
712            namespace: namespace.to_string(),
713            id: id.to_string(),
714            kind: ConfigChangeKind::Put,
715        });
716        Ok(())
717    }
718
719    async fn delete_if_revision(
720        &self,
721        namespace: &str,
722        id: &str,
723        expected_revision: u64,
724    ) -> Result<(), StorageError> {
725        let mut guard = self.configs.write().await;
726        let actual = guard
727            .get(namespace)
728            .and_then(|entries| entries.get(id))
729            .and_then(extract_meta_revision)
730            .unwrap_or(0);
731        if actual != expected_revision {
732            return Err(StorageError::VersionConflict {
733                expected: expected_revision,
734                actual,
735            });
736        }
737        if let Some(entries) = guard.get_mut(namespace) {
738            entries.remove(id);
739        }
740        drop(guard);
741        let _ = self.config_change_tx.send(ConfigChangeEvent {
742            namespace: namespace.to_string(),
743            id: id.to_string(),
744            kind: ConfigChangeKind::Delete,
745        });
746        Ok(())
747    }
748}
749
750// ── ConfigChangeNotifier ────────────────────────────────────────────
751
752#[async_trait]
753impl ConfigChangeNotifier for InMemoryStore {
754    async fn subscribe(&self) -> Result<Box<dyn ConfigChangeSubscriber>, StorageError> {
755        Ok(Box::new(InMemoryConfigChangeSubscriber {
756            rx: self.config_change_tx.subscribe(),
757        }))
758    }
759}
760
761struct InMemoryConfigChangeSubscriber {
762    rx: tokio::sync::broadcast::Receiver<ConfigChangeEvent>,
763}
764
765#[async_trait]
766impl ConfigChangeSubscriber for InMemoryConfigChangeSubscriber {
767    async fn next(&mut self) -> Result<ConfigChangeEvent, StorageError> {
768        match self.rx.recv().await {
769            Ok(event) => Ok(event),
770            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
771                tracing::warn!(skipped, "in-memory config notifier lagged");
772                Ok(ConfigChangeEvent {
773                    namespace: String::new(),
774                    id: String::new(),
775                    kind: ConfigChangeKind::Put,
776                })
777            }
778            Err(tokio::sync::broadcast::error::RecvError::Closed) => {
779                Err(StorageError::Io("config change channel closed".into()))
780            }
781        }
782    }
783}
784
785#[cfg(test)]
786mod tests;
awaken_stores/memory/mod.rs

awaken_stores/memory/
mod.rs
