a3s_effect/
log.rs

1use std::collections::HashMap;
2use std::fs::{self, OpenOptions};
3use std::io::Write;
4use std::path::{Path, PathBuf};
5use std::sync::Mutex;
6
7use crate::error::ActorError;
8use crate::fact::{parse_fact_json, Fact, NewFact};
9
10fn valid_thread(thread_id: &str) -> Result<(), ActorError> {
11    let ok = !thread_id.is_empty()
12        && thread_id.len() <= 128
13        && thread_id
14            .chars()
15            .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '.' | ':' | '-'));
16    if ok {
17        Ok(())
18    } else {
19        Err(ActorError::BadThreadId {
20            thread_id: thread_id.to_string(),
21        })
22    }
23}
24
25fn same_payload(stored: &Fact, incoming: &NewFact) -> bool {
26    stored.kind == incoming.kind && stored.payload == incoming.payload
27}
28
29fn commit(
30    current: &[Fact],
31    incoming: &[NewFact],
32    cause: Option<&str>,
33) -> Result<Vec<Fact>, ActorError> {
34    if incoming.is_empty() {
35        return Ok(current.to_vec());
36    }
37    let already = incoming.iter().all(|item| {
38        current
39            .iter()
40            .any(|stored| stored.key == item.key && same_payload(stored, item))
41    });
42    if already {
43        return Ok(current.to_vec());
44    }
45    if let Some(conflict) = incoming
46        .iter()
47        .find(|item| current.iter().any(|stored| stored.key == item.key))
48    {
49        return Err(ActorError::DuplicateFact {
50            key: conflict.key.clone(),
51        });
52    }
53    let mut seq = current.last().map(|fact| fact.seq).unwrap_or(0);
54    let mut next = current.to_vec();
55    for item in incoming {
56        seq += 1;
57        next.push(Fact {
58            seq,
59            kind: item.kind.clone(),
60            key: item.key.clone(),
61            cause: cause.map(str::to_string),
62            payload: item.payload.clone(),
63        });
64    }
65    Ok(next)
66}
67
68pub trait LogStore: Send + Sync {
69    fn read(&self, thread_id: &str) -> Result<Vec<Fact>, ActorError>;
70    fn append(
71        &self,
72        thread_id: &str,
73        facts: &[NewFact],
74        cause: Option<&str>,
75    ) -> Result<Vec<Fact>, ActorError>;
76}
77
78#[derive(Default)]
79pub struct MemoryLog {
80    threads: Mutex<HashMap<String, Vec<Fact>>>,
81}
82
83impl MemoryLog {
84    pub fn new() -> Self {
85        Self::default()
86    }
87}
88
89impl LogStore for MemoryLog {
90    fn read(&self, thread_id: &str) -> Result<Vec<Fact>, ActorError> {
91        valid_thread(thread_id)?;
92        let threads = self
93            .threads
94            .lock()
95            .unwrap_or_else(|poison| poison.into_inner());
96        Ok(threads.get(thread_id).cloned().unwrap_or_default())
97    }
98
99    fn append(
100        &self,
101        thread_id: &str,
102        facts: &[NewFact],
103        cause: Option<&str>,
104    ) -> Result<Vec<Fact>, ActorError> {
105        valid_thread(thread_id)?;
106        let mut threads = self
107            .threads
108            .lock()
109            .unwrap_or_else(|poison| poison.into_inner());
110        let current = threads.get(thread_id).cloned().unwrap_or_default();
111        let next = commit(&current, facts, cause)?;
112        threads.insert(thread_id.to_string(), next.clone());
113        Ok(next)
114    }
115}
116
117pub struct FileLog {
118    dir: PathBuf,
119    lock: Mutex<()>,
120}
121
122impl FileLog {
123    pub fn open(dir: impl AsRef<Path>) -> Result<Self, ActorError> {
124        let dir = dir.as_ref().to_path_buf();
125        fs::create_dir_all(&dir).map_err(|error| ActorError::Schema(error.to_string()))?;
126        Ok(Self {
127            dir,
128            lock: Mutex::new(()),
129        })
130    }
131
132    fn path(&self, thread_id: &str) -> PathBuf {
133        self.dir.join(format!("{thread_id}.jsonl"))
134    }
135}
136
137impl LogStore for FileLog {
138    fn read(&self, thread_id: &str) -> Result<Vec<Fact>, ActorError> {
139        valid_thread(thread_id)?;
140        let _guard = self
141            .lock
142            .lock()
143            .unwrap_or_else(|poison| poison.into_inner());
144        let path = self.path(thread_id);
145        if !path.exists() {
146            return Ok(Vec::new());
147        }
148        let raw =
149            fs::read_to_string(&path).map_err(|error| ActorError::Schema(error.to_string()))?;
150        let mut facts = Vec::new();
151        for line in raw.lines() {
152            if line.is_empty() {
153                continue;
154            }
155            facts.push(parse_fact_json(line)?);
156        }
157        Ok(facts)
158    }
159
160    fn append(
161        &self,
162        thread_id: &str,
163        facts: &[NewFact],
164        cause: Option<&str>,
165    ) -> Result<Vec<Fact>, ActorError> {
166        valid_thread(thread_id)?;
167        let _guard = self
168            .lock
169            .lock()
170            .unwrap_or_else(|poison| poison.into_inner());
171        let path = self.path(thread_id);
172        let current = if path.exists() {
173            let raw =
174                fs::read_to_string(&path).map_err(|error| ActorError::Schema(error.to_string()))?;
175            let mut parsed = Vec::new();
176            for line in raw.lines() {
177                if !line.is_empty() {
178                    parsed.push(parse_fact_json(line)?);
179                }
180            }
181            parsed
182        } else {
183            Vec::new()
184        };
185        let next = commit(&current, facts, cause)?;
186        if next.len() == current.len() {
187            return Ok(next);
188        }
189        let mut file = OpenOptions::new()
190            .create(true)
191            .append(true)
192            .open(&path)
193            .map_err(|error| ActorError::Schema(error.to_string()))?;
194        for fact in next.iter().skip(current.len()) {
195            let line = serde_json::to_string(fact)
196                .map_err(|error| ActorError::Schema(error.to_string()))?;
197            writeln!(file, "{line}").map_err(|error| ActorError::Schema(error.to_string()))?;
198        }
199        file.sync_all()
200            .map_err(|error| ActorError::Schema(error.to_string()))?;
201        Ok(next)
202    }
203}
a3s_effect/log.rs

a3s_effect/
log.rs
