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

laterite_core/parameters/
mod.rs
