comindly_core/
docstore.rs

1//! One settings document in a LanceDB table: each save is a new table version, recording who
2//! changed it and when, and a save based on an older version is refused.
3
4use std::sync::Arc;
5
6use anyhow::{bail, Result};
7use futures::TryStreamExt;
8use lancedb::arrow::arrow_array::{ArrayRef, RecordBatch, StringArray};
9use lancedb::arrow::arrow_schema::{DataType, Field, Schema};
10use lancedb::query::ExecutableQuery;
11use serde::{de::DeserializeOwned, Serialize};
12
13/// The document with the table version it came from and who saved it.
14#[derive(Debug, Clone, Default)]
15pub struct Stored<T> {
16    pub doc: T,
17    pub version: Option<u64>,
18    pub changed_by: Option<String>,
19    pub changed_at: Option<String>,
20}
21
22/// One saved version, for the history.
23#[derive(Debug, Clone, serde::Serialize)]
24pub struct Version {
25    pub version: u64,
26    pub changed_by: String,
27    pub changed_at: String,
28    pub doc: serde_json::Value,
29}
30
31/// The document in table `table` at `uri`; the default when nothing was saved.
32pub async fn load<T: DeserializeOwned + Default>(uri: &str, table: &str) -> Result<Stored<T>> {
33    let db = crate::storage::connect(uri).execute().await?;
34    let Ok(tbl) = db.open_table(table).execute().await else {
35        return Ok(Stored {
36            doc: T::default(),
37            version: None,
38            changed_by: None,
39            changed_at: None,
40        });
41    };
42    let version = tbl.version().await.ok();
43    let (doc, changed_by, changed_at) = read(&tbl).await?;
44    Ok(Stored {
45        doc: match doc {
46            Some(d) => serde_json::from_str(&d)?,
47            None => T::default(),
48        },
49        version,
50        changed_by,
51        changed_at,
52    })
53}
54
55async fn read(tbl: &lancedb::Table) -> Result<(Option<String>, Option<String>, Option<String>)> {
56    let mut st = tbl.query().execute().await?;
57    let mut out = (None, None, None);
58    while let Some(b) = st.try_next().await? {
59        let col = |n: &str| crate::index::str_col(&b, n);
60        if b.num_rows() > 0 {
61            out = (
62                Some(col("settings")?.value(0).to_string()),
63                Some(col("changed_by")?.value(0).to_string()),
64                Some(col("changed_at")?.value(0).to_string()),
65            );
66        }
67    }
68    Ok(out)
69}
70
71/// Save `doc` as `by` if nobody saved since `expected` (the version it was loaded at).
72pub async fn save<T: Serialize>(
73    uri: &str,
74    table: &str,
75    doc: &T,
76    expected: Option<u64>,
77    by: &str,
78) -> Result<u64> {
79    let db = crate::storage::connect(uri).execute().await?;
80    let now = match db.open_table(table).execute().await {
81        Ok(t) => Some(t.version().await?),
82        Err(_) => None,
83    };
84    if now != expected {
85        bail!("the settings were changed meanwhile; reload and try again");
86    }
87    let schema = Arc::new(Schema::new(vec![
88        Field::new("settings", DataType::Utf8, false),
89        Field::new("changed_by", DataType::Utf8, false),
90        Field::new("changed_at", DataType::Utf8, false),
91    ]));
92    let col = |v: String| Arc::new(StringArray::from(vec![v])) as ArrayRef;
93    let batch = RecordBatch::try_new(
94        schema,
95        vec![
96            col(serde_json::to_string(doc)?),
97            col(by.to_string()),
98            col(chrono::Utc::now().to_rfc3339()),
99        ],
100    )?;
101    crate::index::write_versioned(&db, table, batch).await
102}
103
104/// Every saved version, newest first.
105pub async fn history(uri: &str, table: &str) -> Result<Vec<Version>> {
106    let db = crate::storage::connect(uri).execute().await?;
107    let Ok(tbl) = db.open_table(table).execute().await else {
108        return Ok(Vec::new());
109    };
110    let mut out = Vec::new();
111    for v in tbl.list_versions().await? {
112        tbl.checkout(v.version).await?;
113        if let (Some(doc), Some(by), Some(at)) = read(&tbl).await? {
114            out.push(Version {
115                version: v.version,
116                changed_by: by,
117                changed_at: at,
118                doc: serde_json::from_str(&doc).unwrap_or_default(),
119            });
120        }
121    }
122    tbl.checkout_latest().await?;
123    out.reverse();
124    Ok(out)
125}
comindly_core/docstore.rs

comindly_core/
docstore.rs
