datumesh-store-cli 0.3.2

Command-line tool for datumesh-store: migrate, import, export, bench.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//! `export`: everything as JSON lines, in an order `import` can replay: namespaces (the
//! receipts are verified against them), descriptors, receipts, manifests (by entry; since
//! 0.3.1), blobs, versions (per cell in tx order), computed versions (per cell, by tx and
//! asof; since 0.2.0), name bindings (per name in tx order). The slice-entry lineage and the
//! typed-argument index are not lines of their own: `put_receipt` derives them from the
//! receipts. One consistent snapshot (a read-only REPEATABLE READ transaction), streamed.
//! `import` reads 0.1.x, 0.2.x and 0.3.0 exports too (they have no manifests, and 0.1.x no
//! computed versions).
//!
//! ```text
//! {"namespace":{"host":"…","keydoc":"<key document JSON text>","validated_at":"…"}}
//! {"descriptor":"<descriptor JSON text>"}
//! {"receipt":"<receipt JSON text>"}
//! {"manifest":"<manifest JSON text>"}
//! {"blob":"<content, lowercase hex>"}
//! {"version":{"cell_id":"…","receipt_hash":"…","tx":"…"}}
//! {"computed":{"asof":"…","cell_id":"…","receipt_hash":"…","tx":"…"}}
//! {"name":{"fn_hash":"…","fn_name":"…","tx":"…"}}
//! ```

use std::io::{BufWriter, Write};

use datumesh_core::json::Value;
use datumesh_core::lexical::to_hex;
use postgres::fallible_iterator::FallibleIterator;
use postgres::types::ToSql;
use postgres::Transaction;

use crate::{connect, Fail};

fn s(x: impl Into<String>) -> Value {
    Value::String(x.into())
}

fn obj(members: Vec<(&str, Value)>) -> Value {
    Value::Object(members.into_iter().map(|(k, v)| (k.to_string(), v)).collect())
}

fn line(out: &mut impl Write, kind: &str, v: Value) -> Result<(), Fail> {
    out.write_all(&obj(vec![(kind, v)]).to_jcs())?;
    out.write_all(b"\n")?;
    Ok(())
}

fn each(t: &mut Transaction<'_>, sql: &str, mut f: impl FnMut(&postgres::Row) -> Result<(), Fail>) -> Result<(), Fail> {
    let params: [&(dyn ToSql + Sync); 0] = [];
    let mut rows = t.query_raw(sql, params)?;
    while let Some(r) = rows.next()? {
        f(&r)?;
    }
    Ok(())
}

fn text(b: Vec<u8>) -> Result<String, Fail> {
    String::from_utf8(b).map_err(|e| Fail(1, format!("stored JSON is not UTF-8: {e}")))
}

pub(crate) fn run(url: &str) -> Result<u8, Fail> {
    let mut c = connect(url)?;
    let mut t =
        c.build_transaction().isolation_level(postgres::IsolationLevel::RepeatableRead).read_only(true).start()?;
    let stdout = std::io::stdout();
    let mut out = BufWriter::new(stdout.lock());
    each(&mut t, "SELECT host, keydoc, validated_at FROM namespaces ORDER BY host", |r| {
        let (host, doc, at): (String, Vec<u8>, String) = (r.try_get(0)?, r.try_get(1)?, r.try_get(2)?);
        line(&mut out, "namespace", obj(vec![("host", s(host)), ("keydoc", s(text(doc)?)), ("validated_at", s(at))]))
    })?;
    each(&mut t, "SELECT bytes FROM descriptors ORDER BY fn_hash", |r| {
        line(&mut out, "descriptor", s(text(r.try_get(0)?)?))
    })?;
    each(&mut t, "SELECT bytes FROM receipts ORDER BY hash", |r| line(&mut out, "receipt", s(text(r.try_get(0)?)?)))?;
    each(&mut t, "SELECT bytes FROM manifests ORDER BY entry", |r| {
        line(&mut out, "manifest", s(text(r.try_get(0)?)?))
    })?;
    each(&mut t, "SELECT bytes FROM blobs ORDER BY sha256", |r| {
        let b: Vec<u8> = r.try_get(0)?;
        line(&mut out, "blob", s(to_hex(&b)))
    })?;
    each(&mut t, "SELECT cell_id, tx, receipt_hash FROM versions ORDER BY cell_id, tx", |r| {
        let (cell, tx, h): (String, String, String) = (r.try_get(0)?, r.try_get(1)?, r.try_get(2)?);
        line(&mut out, "version", obj(vec![("cell_id", s(cell)), ("tx", s(tx)), ("receipt_hash", s(h))]))
    })?;
    each(&mut t, "SELECT cell_id, tx, asof, receipt_hash FROM computed_versions ORDER BY cell_id, tx, asof", |r| {
        let (cell, tx, asof, h): (String, String, String, String) =
            (r.try_get(0)?, r.try_get(1)?, r.try_get(2)?, r.try_get(3)?);
        line(
            &mut out,
            "computed",
            obj(vec![("cell_id", s(cell)), ("tx", s(tx)), ("asof", s(asof)), ("receipt_hash", s(h))]),
        )
    })?;
    each(&mut t, "SELECT fn_name, tx, fn_hash FROM name_bindings ORDER BY fn_name, tx", |r| {
        let (name, tx, h): (String, String, String) = (r.try_get(0)?, r.try_get(1)?, r.try_get(2)?);
        line(&mut out, "name", obj(vec![("fn_name", s(name)), ("tx", s(tx)), ("fn_hash", s(h))]))
    })?;
    out.flush()?;
    t.commit()?;
    Ok(0)
}