laterite_core/parameters/
mod.rs

1//! The parameters store: state a module keeps for itself, keyed by name.
2//!
3//! A parameter is a JSON value under a `namespace::group.item` key: the
4//! moment setup completed, the cursor of the last sync, a counter. It is
5//! internal state, written by code, distinct from a settings model an
6//! operator edits on a screen. The functions read and write the table
7//! directly; [`Parameters`] wraps them with a per-process cache that a write
8//! through it refreshes and a lifetime bounds.
9//!
10//! ```ignore
11//! use laterite_core::parameters;
12//!
13//! parameters::set(&db, "acme.blog::sync.cursor", &cursor).await?;
14//! let cursor: Option<String> = parameters::get(&db, "acme.blog::sync.cursor").await?;
15//! ```
16
17use std::collections::HashMap;
18use std::sync::{Arc, Mutex};
19use std::time::{Duration, Instant};
20
21use chrono::{SecondsFormat, Utc};
22use serde::de::DeserializeOwned;
23use serde::Serialize;
24
25use crate::error::{CoreError, CoreResult};
26use crate::query::{bind_values, build, AnyRowExt};
27use crate::{Db, Module, ModuleId};
28use sea_query::{Expr, Iden, OnConflict, Query};
29
30pub mod migrations;
31
32/// The parameters table.
33#[derive(Iden)]
34pub(crate) enum ParametersTable {
35    #[iden = "parameters"]
36    Table,
37    Namespace,
38    Group,
39    Item,
40    Value,
41    UpdatedAt,
42}
43
44/// The module that owns the table, listed among an application's built-ins.
45pub struct ParametersModule;
46
47impl Module for ParametersModule {
48    fn id(&self) -> ModuleId {
49        ModuleId::new(migrations::MODULE_ID)
50    }
51    fn migrations(&self) -> crate::MigrationSet {
52        migrations::migrations()
53    }
54}
55
56/// A key's three parts: `namespace::group.item`. The namespace is the owning
57/// module's id; the item may itself contain dots.
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct Key<'a> {
60    pub namespace: &'a str,
61    pub group: &'a str,
62    pub item: &'a str,
63}
64
65impl<'a> Key<'a> {
66    /// Splits `namespace::group.item`; every part must be present and
67    /// non-empty.
68    pub fn parse(key: &'a str) -> CoreResult<Self> {
69        let malformed = || CoreError::Validation(format!("malformed parameter key `{key}`"));
70        let (namespace, rest) = key.split_once("::").ok_or_else(malformed)?;
71        let (group, item) = rest.split_once('.').ok_or_else(malformed)?;
72        if namespace.is_empty() || group.is_empty() || item.is_empty() {
73            return Err(malformed());
74        }
75        Ok(Self {
76            namespace,
77            group,
78            item,
79        })
80    }
81}
82
83/// Reads a parameter as `T`, or `None` when unset.
84pub async fn get<T: DeserializeOwned>(db: &Db, key: &str) -> CoreResult<Option<T>> {
85    match get_value(db, key).await? {
86        Some(value) => Ok(Some(serde_json::from_value(value).map_err(|e| {
87            CoreError::Internal(format!(
88                "parameter `{key}` holds a value of another shape: {e}"
89            ))
90        })?)),
91        None => Ok(None),
92    }
93}
94
95/// Reads a parameter's raw JSON value, or `None` when unset.
96pub async fn get_value(db: &Db, key: &str) -> CoreResult<Option<serde_json::Value>> {
97    let parts = Key::parse(key)?;
98    let (sql, values) = build(
99        db.backend,
100        Query::select()
101            .column(ParametersTable::Value)
102            .from(ParametersTable::Table)
103            .and_where(Expr::col(ParametersTable::Namespace).eq(parts.namespace))
104            .and_where(Expr::col(ParametersTable::Group).eq(parts.group))
105            .and_where(Expr::col(ParametersTable::Item).eq(parts.item))
106            .to_owned(),
107    );
108    let row = bind_values(sqlx::query(&sql), values)
109        .fetch_optional(&db.pool)
110        .await?;
111    match row {
112        Some(r) => {
113            let text = r.get_text("value")?;
114            Ok(Some(serde_json::from_str(&text).map_err(|e| {
115                CoreError::Internal(format!("parameter `{key}` holds invalid JSON: {e}"))
116            })?))
117        }
118        None => Ok(None),
119    }
120}
121
122/// Writes a parameter, replacing any value it held.
123pub async fn set<T: Serialize + ?Sized>(db: &Db, key: &str, value: &T) -> CoreResult<()> {
124    let parts = Key::parse(key)?;
125    let json = serde_json::to_string(value)
126        .map_err(|e| CoreError::Internal(format!("parameter `{key}` cannot serialize: {e}")))?;
127    let now = Utc::now().to_rfc3339_opts(SecondsFormat::Micros, true);
128    let (sql, values) = build(
129        db.backend,
130        Query::insert()
131            .into_table(ParametersTable::Table)
132            .columns([
133                ParametersTable::Namespace,
134                ParametersTable::Group,
135                ParametersTable::Item,
136                ParametersTable::Value,
137                ParametersTable::UpdatedAt,
138            ])
139            .values_panic([
140                parts.namespace.into(),
141                parts.group.into(),
142                parts.item.into(),
143                json.into(),
144                now.into(),
145            ])
146            .on_conflict(
147                OnConflict::columns([
148                    ParametersTable::Namespace,
149                    ParametersTable::Group,
150                    ParametersTable::Item,
151                ])
152                .update_columns([ParametersTable::Value, ParametersTable::UpdatedAt])
153                .to_owned(),
154            )
155            .to_owned(),
156    );
157    bind_values(sqlx::query(&sql), values)
158        .execute(&db.pool)
159        .await?;
160    Ok(())
161}
162
163/// Removes a parameter. `true` when one was set.
164pub async fn reset(db: &Db, key: &str) -> CoreResult<bool> {
165    let parts = Key::parse(key)?;
166    let (sql, values) = build(
167        db.backend,
168        Query::delete()
169            .from_table(ParametersTable::Table)
170            .and_where(Expr::col(ParametersTable::Namespace).eq(parts.namespace))
171            .and_where(Expr::col(ParametersTable::Group).eq(parts.group))
172            .and_where(Expr::col(ParametersTable::Item).eq(parts.item))
173            .to_owned(),
174    );
175    let done = bind_values(sqlx::query(&sql), values)
176        .execute(&db.pool)
177        .await?;
178    Ok(done.rows_affected() > 0)
179}
180
181/// What the cache holds for a key: the value (or its absence) and when it
182/// was read.
183#[derive(Debug, Clone)]
184struct Cached {
185    value: Option<serde_json::Value>,
186    read_at: Instant,
187}
188
189/// The store behind a per-process cache. A read within `ttl` of the last
190/// read or write of the same key returns the remembered value; a write or
191/// reset through this handle refreshes it; another process's write is seen
192/// once the entry ages out. Clones share the cache.
193#[derive(Clone)]
194pub struct Parameters {
195    db: Db,
196    cache: Arc<Mutex<HashMap<String, Cached>>>,
197    ttl: Duration,
198}
199
200impl std::fmt::Debug for Parameters {
201    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
202        f.debug_struct("Parameters")
203            .field("backend", &self.db.backend)
204            .field("ttl", &self.ttl)
205            .finish_non_exhaustive()
206    }
207}
208
209/// How long a cached read stays good: five minutes.
210pub const DEFAULT_TTL: Duration = Duration::from_secs(300);
211
212impl Parameters {
213    pub fn new(db: Db) -> Self {
214        Self {
215            db,
216            cache: Arc::new(Mutex::new(HashMap::new())),
217            ttl: DEFAULT_TTL,
218        }
219    }
220
221    /// The same store with another lifetime for cached reads.
222    pub fn with_ttl(mut self, ttl: Duration) -> Self {
223        self.ttl = ttl;
224        self
225    }
226
227    /// The connection the store writes through.
228    pub fn db(&self) -> &Db {
229        &self.db
230    }
231
232    fn remembered(&self, key: &str) -> Option<Option<serde_json::Value>> {
233        let cache = self.cache.lock().unwrap_or_else(|e| e.into_inner());
234        cache
235            .get(key)
236            .filter(|c| c.read_at.elapsed() < self.ttl)
237            .map(|c| c.value.clone())
238    }
239
240    fn remember(&self, key: &str, value: Option<serde_json::Value>) {
241        let mut cache = self.cache.lock().unwrap_or_else(|e| e.into_inner());
242        cache.insert(
243            key.to_string(),
244            Cached {
245                value,
246                read_at: Instant::now(),
247            },
248        );
249    }
250
251    /// Reads a parameter as `T`, from the cache when fresh.
252    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> CoreResult<Option<T>> {
253        let value = match self.remembered(key) {
254            Some(value) => value,
255            None => {
256                let value = get_value(&self.db, key).await?;
257                self.remember(key, value.clone());
258                value
259            }
260        };
261        match value {
262            Some(value) => Ok(Some(serde_json::from_value(value).map_err(|e| {
263                CoreError::Internal(format!(
264                    "parameter `{key}` holds a value of another shape: {e}"
265                ))
266            })?)),
267            None => Ok(None),
268        }
269    }
270
271    /// Reads a parameter, or `default` when unset.
272    pub async fn get_or<T: DeserializeOwned>(&self, key: &str, default: T) -> CoreResult<T> {
273        Ok(self.get(key).await?.unwrap_or(default))
274    }
275
276    /// Writes a parameter and remembers it.
277    pub async fn set<T: Serialize + ?Sized>(&self, key: &str, value: &T) -> CoreResult<()> {
278        set(&self.db, key, value).await?;
279        let json = serde_json::to_value(value)
280            .map_err(|e| CoreError::Internal(format!("parameter `{key}` cannot serialize: {e}")))?;
281        self.remember(key, Some(json));
282        Ok(())
283    }
284
285    /// Removes a parameter and remembers its absence.
286    pub async fn reset(&self, key: &str) -> CoreResult<bool> {
287        let done = reset(&self.db, key).await?;
288        self.remember(key, None);
289        Ok(done)
290    }
291
292    /// Drops one key from the cache, so the next read hits the table.
293    pub fn forget(&self, key: &str) {
294        let mut cache = self.cache.lock().unwrap_or_else(|e| e.into_inner());
295        cache.remove(key);
296    }
297
298    /// Drops every cached entry.
299    pub fn clear(&self) {
300        let mut cache = self.cache.lock().unwrap_or_else(|e| e.into_inner());
301        cache.clear();
302    }
303}
304
305#[cfg(test)]
306mod tests {
307    use super::*;
308
309    #[test]
310    fn a_key_splits_into_namespace_group_and_item() {
311        let key = Key::parse("acme.blog::sync.cursor").unwrap();
312        assert_eq!(key.namespace, "acme.blog");
313        assert_eq!(key.group, "sync");
314        assert_eq!(key.item, "cursor");
315        let dotted = Key::parse("laterite.backend::setup.completed.at").unwrap();
316        assert_eq!(dotted.group, "setup");
317        assert_eq!(dotted.item, "completed.at");
318    }
319
320    #[test]
321    fn a_malformed_key_is_refused() {
322        for bad in [
323            "",
324            "sync.cursor",
325            "acme::cursor",
326            "::sync.cursor",
327            "acme::.cursor",
328            "acme::sync.",
329        ] {
330            assert!(
331                matches!(Key::parse(bad), Err(CoreError::Validation(_))),
332                "{bad}"
333            );
334        }
335    }
336}
laterite_core/parameters/mod.rs

laterite_core/parameters/
mod.rs
