70ecd6caa0
Transaction Science open standards monorepo. Each subdirectory is a self-contained reference protocol with its own README and licence. - openpay/ Apache-2.0 payment-acceptance stack (Rust, 26 crates) - smart-byte/ CC-BY-4.0 content-addressed value carrier (spec) - eoc/ CC-BY-4.0 energy-optimised compute (spec) The wire format and the right to fork are public. The optional hosted services are the offer; the protocols are not.
801 lines
30 KiB
Rust
801 lines
30 KiB
Rust
//! Typed facade over Minigraf.
|
|
//!
|
|
//! Minigraf is an EAV (Entity-Attribute-Value) fact store with a
|
|
//! Datalog query language. The OpenPay graph is property-graph
|
|
//! shaped — typed vertices, typed directed edges, JSON properties on
|
|
//! both — so this facade maps the property-graph model onto EAV
|
|
//! facts:
|
|
//!
|
|
//! ```text
|
|
//! Vertex(id, type) → [id :_type type]
|
|
//! VertexProperty(id, name, value) → [id name value]
|
|
//!
|
|
//! Edge(from, type, to) → mint edge_id; assert
|
|
//! [edge_id :_edge/from from]
|
|
//! [edge_id :_edge/to to]
|
|
//! [edge_id :_edge/type type]
|
|
//! EdgeProperty(edge_id, name, val) → [edge_id name value]
|
|
//! ```
|
|
//!
|
|
//! Internal attributes are prefixed with `_` so user property names
|
|
//! can be any non-underscore-leading keyword without collision.
|
|
//!
|
|
//! ## Persistence
|
|
//!
|
|
//! [`new_in_memory`](GraphHandle::new_in_memory) gives an ephemeral
|
|
//! store (tests, single-process kiosks). [`new_persistent`](GraphHandle::new_persistent)
|
|
//! opens a single `.graph` file — the "SQLite of graphs" model
|
|
//! Minigraf is shaped around. Closing the handle flushes; reopening
|
|
//! the same path recovers every fact and lets the operator pick up
|
|
//! exactly where they left off.
|
|
|
|
use std::sync::Arc;
|
|
|
|
use minigraf::{BindValue, Minigraf, Value as MgValue};
|
|
use serde_json::Value;
|
|
use uuid::Uuid;
|
|
|
|
use crate::error::{Error, Result};
|
|
|
|
// ============================================================
|
|
// Vertex / edge type constants
|
|
// ============================================================
|
|
|
|
/// Ledger vertex types.
|
|
pub mod vtypes {
|
|
/// A `ledger_account`.
|
|
pub const LEDGER_ACCOUNT: &str = "ledger_account";
|
|
/// A `ledger_tx`.
|
|
pub const LEDGER_TX: &str = "ledger_tx";
|
|
/// A `ledger_ledger`.
|
|
pub const LEDGER_LEDGER: &str = "ledger_ledger";
|
|
/// A `webhook_event`.
|
|
pub const WEBHOOK_EVENT: &str = "webhook_event";
|
|
/// A `webhook_endpoint`.
|
|
pub const WEBHOOK_ENDPOINT: &str = "webhook_endpoint";
|
|
/// A `webhook_attempt`.
|
|
pub const WEBHOOK_ATTEMPT: &str = "webhook_attempt";
|
|
/// A `statement_line` ingested from a reconciliation source.
|
|
pub const STATEMENT_LINE: &str = "statement_line";
|
|
/// A `reconciliation_task` — one open discrepancy for an operator.
|
|
pub const RECONCILIATION_TASK: &str = "reconciliation_task";
|
|
/// A `rail_attempt` — one rail-driver attempt recorded for
|
|
/// routing-signal purposes. See `crate::rail_telemetry`.
|
|
pub const RAIL_ATTEMPT: &str = "rail_attempt";
|
|
/// A `ledger_checkpoint` — operator-named bookmark into the
|
|
/// bi-temporal history. Holds `name` (String) and `tx_count`
|
|
/// (u64) properties.
|
|
pub const LEDGER_CHECKPOINT: &str = "ledger_checkpoint";
|
|
/// A `refund` — one refund workflow record. Vertex id matches
|
|
/// `RefundId`'s UUID. Properties carry the full `Refund` JSON
|
|
/// in `state`, plus indexed fields `external_id`,
|
|
/// `original_tx_id`, `status_code`.
|
|
pub const REFUND: &str = "refund";
|
|
/// A `dispute` — one chargeback / inquiry / representment
|
|
/// record. Mirror shape of `REFUND`.
|
|
pub const DISPUTE: &str = "dispute";
|
|
/// A `settlement_batch` — one payout-batch lifecycle.
|
|
pub const SETTLEMENT_BATCH: &str = "settlement_batch";
|
|
/// An `idempotency_record` — orchestrator idempotency-key
|
|
/// cache entry. Vertex id is the UUIDv5 over the key string;
|
|
/// properties carry `key`, `body_signature`, `state` (the
|
|
/// cached outcome).
|
|
pub const IDEMPOTENCY_RECORD: &str = "idempotency_record";
|
|
/// A `subscription` — one customer's recurring-billing record.
|
|
/// Vertex id matches `SubscriptionId`'s UUID. Properties carry
|
|
/// full `Subscription` JSON in `state`, plus indexed
|
|
/// `external_id`, `customer_ref`, `status_code`,
|
|
/// `current_period_end`.
|
|
pub const SUBSCRIPTION: &str = "subscription";
|
|
}
|
|
|
|
/// Edge type constants.
|
|
pub mod etypes {
|
|
/// `ledger_tx` --debit--> `ledger_account`.
|
|
pub const LEDGER_DEBIT: &str = "ledger_debit";
|
|
/// `ledger_tx` --credit--> `ledger_account`.
|
|
pub const LEDGER_CREDIT: &str = "ledger_credit";
|
|
/// `ledger_tx` --in--> `ledger_ledger`, or
|
|
/// `ledger_account` --in--> `ledger_ledger`.
|
|
pub const LEDGER_IN_LEDGER: &str = "ledger_in_ledger";
|
|
/// `ledger_tx` --reverses--> `ledger_tx` (the original).
|
|
pub const LEDGER_REVERSES: &str = "ledger_reverses";
|
|
/// `webhook_event` --delivers--> `webhook_attempt`.
|
|
pub const WEBHOOK_DELIVERS: &str = "webhook_delivers";
|
|
/// `webhook_attempt` --to--> `webhook_endpoint`.
|
|
pub const WEBHOOK_TO: &str = "webhook_to";
|
|
/// `statement_line` --reconciles--> `ledger_tx` (matched line).
|
|
pub const RECONCILES: &str = "reconciles";
|
|
/// `reconciliation_task` --about--> `statement_line` | `ledger_tx`.
|
|
pub const TASK_ABOUT: &str = "task_about";
|
|
/// `refund` --refunds--> `ledger_tx` (the original transaction).
|
|
pub const REFUND_REFUNDS: &str = "refund_refunds";
|
|
/// `dispute` --disputes--> `ledger_tx`.
|
|
pub const DISPUTE_DISPUTES: &str = "dispute_disputes";
|
|
/// `settlement_batch` --includes--> `ledger_tx` (each posted tx
|
|
/// that's part of the batch).
|
|
pub const BATCH_INCLUDES: &str = "batch_includes";
|
|
}
|
|
|
|
// ============================================================
|
|
// Local Vertex / Edge value types
|
|
// ============================================================
|
|
|
|
/// A typed vertex as returned by [`GraphHandle::get_vertex`] and
|
|
/// related queries.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub struct Vertex {
|
|
/// Vertex id (matches the caller-supplied id at creation).
|
|
pub id: Uuid,
|
|
/// Vertex type, e.g. [`vtypes::LEDGER_TX`].
|
|
pub t: String,
|
|
}
|
|
|
|
/// A typed directed edge. Edges have their own opaque id so they can
|
|
/// carry properties.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub struct Edge {
|
|
/// Internal edge id (used to attach properties).
|
|
pub id: Uuid,
|
|
/// Source vertex id.
|
|
pub from: Uuid,
|
|
/// Edge type, e.g. [`etypes::LEDGER_DEBIT`].
|
|
pub t: String,
|
|
/// Destination vertex id.
|
|
pub to: Uuid,
|
|
}
|
|
|
|
// ============================================================
|
|
// GraphHandle
|
|
// ============================================================
|
|
|
|
/// Thread-safe handle to a graph store.
|
|
///
|
|
/// Internally wraps an `Arc<Minigraf>` so it can be cheaply cloned for
|
|
/// multi-threaded callers. The backend is chosen at construction:
|
|
/// [`new_in_memory`](Self::new_in_memory) for an ephemeral store, or
|
|
/// [`new_persistent`](Self::new_persistent) for a single-file
|
|
/// `.graph` database that survives restarts.
|
|
#[derive(Clone)]
|
|
pub struct GraphHandle {
|
|
inner: Arc<Minigraf>,
|
|
}
|
|
|
|
impl std::fmt::Debug for GraphHandle {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
// The Minigraf instance itself doesn't surface a useful Debug;
|
|
// logging the tx counter is the most informative thing we can
|
|
// produce without forcing a query.
|
|
f.debug_struct("GraphHandle")
|
|
.field("tx_count", &self.inner.current_tx_count())
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl Default for GraphHandle {
|
|
fn default() -> Self {
|
|
Self::new_in_memory()
|
|
}
|
|
}
|
|
|
|
impl GraphHandle {
|
|
/// Construct backed by an in-memory Minigraf instance.
|
|
///
|
|
/// Drop the handle and the data is gone — appropriate for tests
|
|
/// and for single-process embedded deployments that rebuild graph
|
|
/// state on every restart from a system-of-record store.
|
|
///
|
|
/// # Panics
|
|
/// Panics only if Minigraf itself fails to initialize an
|
|
/// in-memory database, which it does not under any documented
|
|
/// condition.
|
|
#[must_use]
|
|
pub fn new_in_memory() -> Self {
|
|
Self {
|
|
inner: Arc::new(
|
|
Minigraf::in_memory()
|
|
.expect("Minigraf::in_memory never fails per documented contract"),
|
|
),
|
|
}
|
|
}
|
|
|
|
/// Construct backed by a single-file `.graph` database at `path`.
|
|
///
|
|
/// Opens (or creates) the file. Subsequent constructions pointed
|
|
/// at the same path see the previously-written facts — the graph
|
|
/// is the system of record across restarts.
|
|
///
|
|
/// # Errors
|
|
/// [`Error::Backend`] if Minigraf can't open the path (permission
|
|
/// denied, corrupt file, ...).
|
|
pub fn new_persistent(path: impl AsRef<std::path::Path>) -> Result<Self> {
|
|
let db = Minigraf::open(path).map_err(|e| Error::Backend(e.to_string()))?;
|
|
Ok(Self {
|
|
inner: Arc::new(db),
|
|
})
|
|
}
|
|
|
|
/// Borrow the underlying Minigraf instance. **Stability note:**
|
|
/// Minigraf's API is not part of OpenPay's stability surface.
|
|
#[must_use]
|
|
pub fn raw(&self) -> &Minigraf {
|
|
&self.inner
|
|
}
|
|
|
|
/// Flush in-memory facts to the durable file and truncate the
|
|
/// WAL sidecar.
|
|
///
|
|
/// **Scope.** This is Minigraf's `checkpoint` operation —
|
|
/// consolidates the write-ahead log into the main `.graph`
|
|
/// file so reopen is faster and the operator's disk footprint
|
|
/// is what they expect. It does **not** purge retracted facts
|
|
/// from history; Minigraf preserves the full bi-temporal log
|
|
/// by design (time-travel queries depend on it). Operators who
|
|
/// genuinely need historical pruning ship a custom store that
|
|
/// implements the trait surface against a substrate that
|
|
/// supports it.
|
|
///
|
|
/// In-memory databases: no-op (returns `Ok(())`).
|
|
///
|
|
/// # Errors
|
|
/// [`Error::Backend`] on I/O failure during checkpoint.
|
|
pub fn compact(&self) -> Result<()> {
|
|
self.inner
|
|
.checkpoint()
|
|
.map_err(|e| Error::Backend(format!("minigraf checkpoint: {e}")))
|
|
}
|
|
|
|
/// The current monotonic transaction counter.
|
|
///
|
|
/// Each successful `transact` / `retract` advances this by
|
|
/// exactly 1, regardless of how many facts the batch carries.
|
|
/// Snapshot it right after a write (`let snap = h.tx_count();`)
|
|
/// to get a stable reference point for later time-travel
|
|
/// queries (`:as-of snap`).
|
|
///
|
|
/// This is the value [`LedgerHistory`](op_ledger::LedgerHistory)
|
|
/// queries against.
|
|
#[must_use]
|
|
pub fn tx_count(&self) -> u64 {
|
|
self.inner.current_tx_count()
|
|
}
|
|
|
|
// --------------------------------------------------------
|
|
// Vertex create / read
|
|
// --------------------------------------------------------
|
|
|
|
/// Create a typed vertex with a caller-supplied id.
|
|
pub fn create_vertex(&self, vertex_type: &str, id: Uuid) -> Result<()> {
|
|
let script = format!(
|
|
r#"(transact [[#uuid "{id}" :_type "{vt}"]])"#,
|
|
id = id,
|
|
vt = escape_str(vertex_type),
|
|
);
|
|
self.execute(&script).map(|_| ())
|
|
}
|
|
|
|
/// Fetch a vertex by id. Returns `None` if no vertex exists.
|
|
pub fn get_vertex(&self, id: Uuid) -> Result<Option<Vertex>> {
|
|
let script = format!(
|
|
r#"(query [:find ?t :where [#uuid "{id}" :_type ?t]])"#,
|
|
id = id
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let row = result.rows.into_iter().next();
|
|
Ok(row.map(|r| Vertex {
|
|
id,
|
|
t: match r.into_iter().next() {
|
|
Some(MgValue::String(s)) => s,
|
|
_ => String::new(),
|
|
},
|
|
}))
|
|
}
|
|
|
|
/// Fetch a vertex and verify its type.
|
|
///
|
|
/// # Errors
|
|
/// [`Error::VertexNotFound`] / [`Error::VertexTypeMismatch`].
|
|
pub fn get_typed_vertex(&self, id: Uuid, expected_type: &str) -> Result<Vertex> {
|
|
let v = self.get_vertex(id)?.ok_or_else(|| Error::VertexNotFound {
|
|
vertex_type: expected_type.to_owned(),
|
|
id: id.to_string(),
|
|
})?;
|
|
if v.t != expected_type {
|
|
return Err(Error::VertexTypeMismatch {
|
|
id: id.to_string(),
|
|
expected: expected_type.to_owned(),
|
|
actual: v.t,
|
|
});
|
|
}
|
|
Ok(v)
|
|
}
|
|
|
|
/// Does a vertex with this id exist?
|
|
pub fn vertex_exists(&self, id: Uuid) -> Result<bool> {
|
|
Ok(self.get_vertex(id)?.is_some())
|
|
}
|
|
|
|
// --------------------------------------------------------
|
|
// Edge create / list
|
|
// --------------------------------------------------------
|
|
|
|
/// Create a directed, typed edge. The returned [`Edge`] carries
|
|
/// the minted edge id so callers can attach properties.
|
|
pub fn create_edge(&self, from: Uuid, edge_type: &str, to: Uuid) -> Result<Edge> {
|
|
let edge_id = Uuid::new_v4();
|
|
let script = format!(
|
|
r#"(transact [[#uuid "{eid}" :_edge/from #uuid "{from}"]
|
|
[#uuid "{eid}" :_edge/to #uuid "{to}"]
|
|
[#uuid "{eid}" :_edge/type "{et}"]])"#,
|
|
eid = edge_id,
|
|
from = from,
|
|
to = to,
|
|
et = escape_str(edge_type),
|
|
);
|
|
self.execute(&script)?;
|
|
Ok(Edge {
|
|
id: edge_id,
|
|
from,
|
|
t: edge_type.to_owned(),
|
|
to,
|
|
})
|
|
}
|
|
|
|
/// List outbound edges of a given type from a vertex.
|
|
pub fn out_edges(&self, from: Uuid, edge_type: &str) -> Result<Vec<Edge>> {
|
|
let script = format!(
|
|
r#"(query [:find ?eid ?to
|
|
:where [?eid :_edge/from #uuid "{from}"]
|
|
[?eid :_edge/type "{et}"]
|
|
[?eid :_edge/to ?to]])"#,
|
|
from = from,
|
|
et = escape_str(edge_type),
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let mut edges = Vec::with_capacity(result.rows.len());
|
|
for row in result.rows {
|
|
let mut it = row.into_iter();
|
|
let eid = expect_ref(it.next())?;
|
|
let to = expect_ref(it.next())?;
|
|
edges.push(Edge {
|
|
id: eid,
|
|
from,
|
|
t: edge_type.to_owned(),
|
|
to,
|
|
});
|
|
}
|
|
Ok(edges)
|
|
}
|
|
|
|
/// List inbound edges of a given type to a vertex.
|
|
pub fn in_edges(&self, to: Uuid, edge_type: &str) -> Result<Vec<Edge>> {
|
|
let script = format!(
|
|
r#"(query [:find ?eid ?from
|
|
:where [?eid :_edge/to #uuid "{to}"]
|
|
[?eid :_edge/type "{et}"]
|
|
[?eid :_edge/from ?from]])"#,
|
|
to = to,
|
|
et = escape_str(edge_type),
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let mut edges = Vec::with_capacity(result.rows.len());
|
|
for row in result.rows {
|
|
let mut it = row.into_iter();
|
|
let eid = expect_ref(it.next())?;
|
|
let from = expect_ref(it.next())?;
|
|
edges.push(Edge {
|
|
id: eid,
|
|
from,
|
|
t: edge_type.to_owned(),
|
|
to,
|
|
});
|
|
}
|
|
Ok(edges)
|
|
}
|
|
|
|
// --------------------------------------------------------
|
|
// Properties
|
|
// --------------------------------------------------------
|
|
|
|
/// Set a property on a vertex.
|
|
///
|
|
/// Property names must be valid Datalog attribute keywords
|
|
/// (alphanumeric, dash, underscore, dot, slash) and must not
|
|
/// start with an underscore (`_` is reserved for internal
|
|
/// schema attributes like `_type`, `_edge/from`).
|
|
///
|
|
/// Minigraf is bi-temporal: every assert is appended to history.
|
|
/// To give scalar properties "current value" semantics, we
|
|
/// retract the old `(entity, attr)` fact before asserting the
|
|
/// new one. The historical trail is preserved (queryable via
|
|
/// `:as-of`); only the *current* view reflects the latest set.
|
|
pub fn set_vertex_property(&self, vertex_id: Uuid, name: &str, value: Value) -> Result<()> {
|
|
self.set_entity_property(vertex_id, name, value)
|
|
}
|
|
|
|
/// Set a property on a directed edge. The edge must already
|
|
/// exist (its id is carried in the `Edge` struct returned by
|
|
/// [`create_edge`](Self::create_edge)). Same retract-then-assert
|
|
/// semantics as [`set_vertex_property`](Self::set_vertex_property).
|
|
pub fn set_edge_property(&self, edge: &Edge, name: &str, value: Value) -> Result<()> {
|
|
self.set_entity_property(edge.id, name, value)
|
|
}
|
|
|
|
/// Shared retract-then-assert implementation for scalar
|
|
/// properties on any entity (vertex or edge).
|
|
fn set_entity_property(&self, entity_id: Uuid, name: &str, value: Value) -> Result<()> {
|
|
validate_user_attr(name)?;
|
|
// 1. Find the current value(s) for this (entity, attr) so we
|
|
// can retract them. There should be at most one for a
|
|
// scalar, but retracting all defends against any earlier
|
|
// multi-assert.
|
|
let lookup = format!(
|
|
r#"(query [:find ?v :where [#uuid "{id}" :{attr} ?v]])"#,
|
|
id = entity_id,
|
|
attr = name,
|
|
);
|
|
let current = self.execute(&lookup)?;
|
|
for row in current.rows {
|
|
if let Some(old) = row.into_iter().next() {
|
|
let lit = mg_value_to_literal(&old);
|
|
let script = format!(
|
|
r#"(retract [[#uuid "{id}" :{attr} {val}]])"#,
|
|
id = entity_id,
|
|
attr = name,
|
|
val = lit,
|
|
);
|
|
self.execute(&script)?;
|
|
}
|
|
}
|
|
// 2. Assert the new value.
|
|
let v = json_to_mg_literal(&value);
|
|
let script = format!(
|
|
r#"(transact [[#uuid "{id}" :{attr} {val}]])"#,
|
|
id = entity_id,
|
|
attr = name,
|
|
val = v,
|
|
);
|
|
self.execute(&script).map(|_| ())
|
|
}
|
|
|
|
/// Fetch all user-visible properties for a vertex as a
|
|
/// `serde_json::Map`. Internal `_*` attributes (`_type`,
|
|
/// `_edge/*`) are filtered out.
|
|
pub fn get_vertex_properties(&self, vertex_id: Uuid) -> Result<serde_json::Map<String, Value>> {
|
|
let script = format!(
|
|
r#"(query [:find ?attr ?val :where [#uuid "{id}" ?attr ?val]])"#,
|
|
id = vertex_id
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let mut map = serde_json::Map::new();
|
|
for row in result.rows {
|
|
let mut it = row.into_iter();
|
|
// Minigraf returns keywords with their leading colon
|
|
// (e.g. `Keyword(":name")`); strip it so callers use bare
|
|
// attribute names. Internal `_*` attrs (`:_type`,
|
|
// `:_edge/*`) are filtered out as they're schema, not
|
|
// user-visible properties.
|
|
let attr = match it.next() {
|
|
Some(MgValue::Keyword(k)) => k.trim_start_matches(':').to_owned(),
|
|
Some(MgValue::String(s)) => s,
|
|
_ => continue,
|
|
};
|
|
if attr.starts_with('_') {
|
|
continue;
|
|
}
|
|
if let Some(val) = it.next() {
|
|
map.insert(attr, mg_to_json(val));
|
|
}
|
|
}
|
|
Ok(map)
|
|
}
|
|
|
|
/// Fetch all user-visible properties for a specific edge.
|
|
pub fn get_edge_properties(&self, edge: &Edge) -> Result<serde_json::Map<String, Value>> {
|
|
self.get_vertex_properties(edge.id)
|
|
}
|
|
|
|
// --------------------------------------------------------
|
|
// Diagnostics
|
|
// --------------------------------------------------------
|
|
|
|
/// Number of vertices total (diagnostic only). A vertex is
|
|
/// anything that has a `:_type` fact.
|
|
pub fn vertex_count(&self) -> Result<u64> {
|
|
let result = self.execute("(query [:find ?e :where [?e :_type _]])")?;
|
|
Ok(result.rows.len() as u64)
|
|
}
|
|
|
|
/// Number of edges total (diagnostic only). An edge is anything
|
|
/// that has a `:_edge/type` fact.
|
|
pub fn edge_count(&self) -> Result<u64> {
|
|
let result = self.execute("(query [:find ?e :where [?e :_edge/type _]])")?;
|
|
Ok(result.rows.len() as u64)
|
|
}
|
|
|
|
// --------------------------------------------------------
|
|
// Time-travel reads (`:as-of N`)
|
|
//
|
|
// Each `_at` method parallels its present-time sibling but
|
|
// appends `:as-of <tx_count>` to the underlying Datalog query.
|
|
// Returns the slice of the graph that was visible at that
|
|
// transaction counter — facts retracted after that point are
|
|
// resurrected; facts asserted after that point are hidden.
|
|
//
|
|
// The Datalog query model: `:as-of N` is a transaction-time
|
|
// filter applied before retraction logic, so the result is the
|
|
// same property-graph view a present-time query would have
|
|
// produced at tx_count = N.
|
|
// --------------------------------------------------------
|
|
|
|
/// `out_edges`, scoped to the graph state at `tx_count`.
|
|
pub fn out_edges_at(&self, from: Uuid, edge_type: &str, tx_count: u64) -> Result<Vec<Edge>> {
|
|
let script = format!(
|
|
r#"(query [:find ?eid ?to
|
|
:where [?eid :_edge/from #uuid "{from}"]
|
|
[?eid :_edge/type "{et}"]
|
|
[?eid :_edge/to ?to]
|
|
:as-of {tx_count}])"#,
|
|
from = from,
|
|
et = escape_str(edge_type),
|
|
tx_count = tx_count,
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let mut edges = Vec::with_capacity(result.rows.len());
|
|
for row in result.rows {
|
|
let mut it = row.into_iter();
|
|
let eid = expect_ref(it.next())?;
|
|
let to = expect_ref(it.next())?;
|
|
edges.push(Edge {
|
|
id: eid,
|
|
from,
|
|
t: edge_type.to_owned(),
|
|
to,
|
|
});
|
|
}
|
|
Ok(edges)
|
|
}
|
|
|
|
/// `in_edges`, scoped to the graph state at `tx_count`.
|
|
pub fn in_edges_at(&self, to: Uuid, edge_type: &str, tx_count: u64) -> Result<Vec<Edge>> {
|
|
let script = format!(
|
|
r#"(query [:find ?eid ?from
|
|
:where [?eid :_edge/to #uuid "{to}"]
|
|
[?eid :_edge/type "{et}"]
|
|
[?eid :_edge/from ?from]
|
|
:as-of {tx_count}])"#,
|
|
to = to,
|
|
et = escape_str(edge_type),
|
|
tx_count = tx_count,
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let mut edges = Vec::with_capacity(result.rows.len());
|
|
for row in result.rows {
|
|
let mut it = row.into_iter();
|
|
let eid = expect_ref(it.next())?;
|
|
let from = expect_ref(it.next())?;
|
|
edges.push(Edge {
|
|
id: eid,
|
|
from,
|
|
t: edge_type.to_owned(),
|
|
to,
|
|
});
|
|
}
|
|
Ok(edges)
|
|
}
|
|
|
|
/// `get_vertex_properties`, scoped to `tx_count`. The returned
|
|
/// map reflects the property values in force at that point —
|
|
/// scalars set later are absent, retracted-since values are
|
|
/// resurrected.
|
|
pub fn get_vertex_properties_at(
|
|
&self,
|
|
vertex_id: Uuid,
|
|
tx_count: u64,
|
|
) -> Result<serde_json::Map<String, Value>> {
|
|
let script = format!(
|
|
r#"(query [:find ?attr ?val
|
|
:where [#uuid "{id}" ?attr ?val]
|
|
:as-of {tx_count}])"#,
|
|
id = vertex_id,
|
|
tx_count = tx_count,
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let mut map = serde_json::Map::new();
|
|
for row in result.rows {
|
|
let mut it = row.into_iter();
|
|
let attr = match it.next() {
|
|
Some(MgValue::Keyword(k)) => k.trim_start_matches(':').to_owned(),
|
|
Some(MgValue::String(s)) => s,
|
|
_ => continue,
|
|
};
|
|
if attr.starts_with('_') {
|
|
continue;
|
|
}
|
|
if let Some(val) = it.next() {
|
|
map.insert(attr, mg_to_json(val));
|
|
}
|
|
}
|
|
Ok(map)
|
|
}
|
|
|
|
/// `get_edge_properties`, scoped to `tx_count`.
|
|
pub fn get_edge_properties_at(
|
|
&self,
|
|
edge: &Edge,
|
|
tx_count: u64,
|
|
) -> Result<serde_json::Map<String, Value>> {
|
|
self.get_vertex_properties_at(edge.id, tx_count)
|
|
}
|
|
|
|
/// All vertices of a given type.
|
|
pub fn vertices_of_type(&self, vertex_type: &str) -> Result<Vec<Vertex>> {
|
|
let script = format!(
|
|
r#"(query [:find ?e :where [?e :_type "{vt}"]])"#,
|
|
vt = escape_str(vertex_type),
|
|
);
|
|
let result = self.execute(&script)?;
|
|
let mut out = Vec::with_capacity(result.rows.len());
|
|
for row in result.rows {
|
|
let mut it = row.into_iter();
|
|
let id = expect_ref(it.next())?;
|
|
out.push(Vertex {
|
|
id,
|
|
t: vertex_type.to_owned(),
|
|
});
|
|
}
|
|
Ok(out)
|
|
}
|
|
|
|
// --------------------------------------------------------
|
|
// Internal helpers
|
|
// --------------------------------------------------------
|
|
|
|
fn execute(&self, script: &str) -> Result<QueryRows> {
|
|
let result = self
|
|
.inner
|
|
.execute(script)
|
|
.map_err(|e| Error::Backend(format!("minigraf: {e}")))?;
|
|
Ok(QueryRows::from(result))
|
|
}
|
|
}
|
|
|
|
// ============================================================
|
|
// Result adapter + helpers
|
|
// ============================================================
|
|
|
|
/// A minimal projection of `minigraf::QueryResult` that exposes the
|
|
/// `rows` as `Vec<Vec<Value>>` regardless of the underlying query
|
|
/// shape. Keeps the rest of this file from having to handle the
|
|
/// distinction between `Query` and `Transact` result variants.
|
|
struct QueryRows {
|
|
rows: Vec<Vec<MgValue>>,
|
|
}
|
|
|
|
impl From<minigraf::QueryResult> for QueryRows {
|
|
fn from(result: minigraf::QueryResult) -> Self {
|
|
match result {
|
|
minigraf::QueryResult::QueryResults { results, .. } => Self { rows: results },
|
|
minigraf::QueryResult::Transacted(_)
|
|
| minigraf::QueryResult::Retracted(_)
|
|
| minigraf::QueryResult::Ok => Self { rows: Vec::new() },
|
|
}
|
|
}
|
|
}
|
|
|
|
fn expect_ref(v: Option<MgValue>) -> Result<Uuid> {
|
|
match v {
|
|
// EntityId is `pub type EntityId = Uuid`, so a `Ref` already
|
|
// carries the Uuid directly.
|
|
Some(MgValue::Ref(eid)) => Ok(eid),
|
|
Some(other) => Err(Error::Backend(format!(
|
|
"expected entity ref, got {other:?}"
|
|
))),
|
|
None => Err(Error::Backend("query row missing expected column".into())),
|
|
}
|
|
}
|
|
|
|
fn escape_str(s: &str) -> String {
|
|
s.replace('\\', "\\\\").replace('"', "\\\"")
|
|
}
|
|
|
|
/// Allow only attribute names that won't break our minimal Datalog
|
|
/// formatting (no whitespace, no quotes), and reject leading `_`
|
|
/// which is reserved for schema attributes.
|
|
fn validate_user_attr(name: &str) -> Result<()> {
|
|
if name.is_empty()
|
|
|| name.starts_with('_')
|
|
|| name
|
|
.chars()
|
|
.any(|c| !(c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | '/')))
|
|
{
|
|
return Err(Error::InvalidInput(format!(
|
|
"property name {name:?} is not a valid attribute keyword"
|
|
)));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Render a `serde_json::Value` as a Datalog literal embedded in a
|
|
/// transact form. Nested arrays/objects are stored as their JSON
|
|
/// string representation, prefixed with `json:` so reads can detect
|
|
/// and re-parse them.
|
|
fn json_to_mg_literal(v: &Value) -> String {
|
|
match v {
|
|
Value::Null => "nil".to_owned(),
|
|
Value::Bool(true) => "true".to_owned(),
|
|
Value::Bool(false) => "false".to_owned(),
|
|
Value::Number(n) => {
|
|
if let Some(i) = n.as_i64() {
|
|
i.to_string()
|
|
} else if let Some(f) = n.as_f64() {
|
|
// Datalog floats need a decimal point.
|
|
if f.fract() == 0.0 {
|
|
format!("{f:.1}")
|
|
} else {
|
|
f.to_string()
|
|
}
|
|
} else {
|
|
"nil".to_owned()
|
|
}
|
|
}
|
|
Value::String(s) => format!("\"{}\"", escape_str(s)),
|
|
Value::Array(_) | Value::Object(_) => {
|
|
let blob = serde_json::to_string(v).unwrap_or_else(|_| "null".to_owned());
|
|
format!("\"json:{}\"", escape_str(&blob))
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Render a minigraf `Value` as a Datalog literal, for embedding in
|
|
/// a `(retract [[entity attr LITERAL]])` form. The inverse of
|
|
/// [`json_to_mg_literal`] when crossed through the assert/retract
|
|
/// round trip.
|
|
fn mg_value_to_literal(v: &MgValue) -> String {
|
|
match v {
|
|
MgValue::Null => "nil".to_owned(),
|
|
MgValue::Boolean(true) => "true".to_owned(),
|
|
MgValue::Boolean(false) => "false".to_owned(),
|
|
MgValue::Integer(i) => i.to_string(),
|
|
MgValue::Float(f) => {
|
|
if f.fract() == 0.0 {
|
|
format!("{f:.1}")
|
|
} else {
|
|
f.to_string()
|
|
}
|
|
}
|
|
MgValue::String(s) => format!("\"{}\"", escape_str(s)),
|
|
// Keywords minigraf returns include the leading `:`; emit
|
|
// them verbatim. Refs use the `#uuid "..."` form.
|
|
MgValue::Keyword(k) => k.clone(),
|
|
MgValue::Ref(u) => format!(r#"#uuid "{u}""#),
|
|
}
|
|
}
|
|
|
|
/// Convert a minigraf `Value` back to a `serde_json::Value`. Strings
|
|
/// prefixed with `json:` are re-parsed as their original JSON shape
|
|
/// (the inverse of [`json_to_mg_literal`]).
|
|
fn mg_to_json(v: MgValue) -> Value {
|
|
match v {
|
|
MgValue::Null => Value::Null,
|
|
MgValue::Boolean(b) => Value::Bool(b),
|
|
MgValue::Integer(i) => Value::Number(i.into()),
|
|
MgValue::Float(f) => serde_json::Number::from_f64(f)
|
|
.map(Value::Number)
|
|
.unwrap_or(Value::Null),
|
|
MgValue::String(s) => {
|
|
if let Some(rest) = s.strip_prefix("json:") {
|
|
serde_json::from_str(rest).unwrap_or(Value::String(s))
|
|
} else {
|
|
Value::String(s)
|
|
}
|
|
}
|
|
MgValue::Keyword(k) => Value::String(k),
|
|
MgValue::Ref(eid) => Value::String(eid.to_string()),
|
|
}
|
|
}
|
|
|
|
// `BindValue` import kept so future parameterized-query helpers can
|
|
// be added without touching the import list.
|
|
#[allow(dead_code)]
|
|
fn _ensure_bindvalue_imported(_: BindValue) {}
|