mod db;
#[cfg(feature = "postgres")]
pub mod pg;
#[cfg(feature = "postgres")]
mod pgtls;
use std::collections::{HashMap, HashSet, VecDeque};
use std::time::{SystemTime, UNIX_EPOCH};
use areev_core::authz;
use areev_core::error::{Hash, AreevError, Result};
use areev_core::format::deserialize::{deserialize_blob, DeserializedGrain};
use areev_core::format::serialize::serialize_grain;
use areev_core::ns::{require_exact_ns, require_writable_ns, NsScope};
use areev_core::types::Grain;
use areev_core::types::{
step_action_node, step_action_relation, Observation, RelatedTo, STEP_ACTION_PREFIX,
};
use turso::Value;
use zeroize::Zeroize;
use db::{db_err, with_txn, Db, TursoDb};
pub const OP_ADD: i64 = 1;
pub const OP_SUPERSEDE: i64 = 2;
pub const OP_FORGET: i64 = 3;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Axis {
World,
Knowledge,
}
#[derive(Debug, Clone)]
pub struct OpRecord {
pub op_seq: i64,
pub hlc: i64,
pub op: i64,
pub hash: Hash,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Direction {
Out,
In,
Both,
}
impl Direction {
pub fn parse(s: &str) -> Option<Self> {
match s.trim().to_ascii_lowercase().as_str() {
"out" | "" => Some(Self::Out),
"in" => Some(Self::In),
"both" => Some(Self::Both),
_ => None,
}
}
}
impl Axis {
pub fn parse(s: &str) -> Option<Self> {
match s.trim().to_ascii_lowercase().as_str() {
"world" | "" => Some(Self::World),
"knowledge" => Some(Self::Knowledge),
_ => None,
}
}
}
pub fn parse_relations(csv: &str) -> Vec<String> {
csv.split(',')
.map(str::trim)
.filter(|r| !r.is_empty())
.map(str::to_string)
.collect()
}
#[derive(Debug, Clone)]
pub struct BundleStats {
pub ops: usize,
pub bytes: u64,
pub last_op_seq: i64,
}
#[derive(Debug, Clone, Default)]
pub struct ImportStats {
pub applied: usize,
pub skipped: usize,
pub meta_applied: usize,
pub meta_skipped: usize,
}
pub(crate) fn like_escape(s: &str) -> String {
s.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_")
}
fn cas_hexes(view: &DeserializedGrain) -> Vec<String> {
let mut out = Vec::new();
if let Some(refs) = view.fields.get("content_refs").and_then(|v| v.as_array()) {
for r in refs {
let uri = r
.get("uri")
.and_then(|u| u.as_str())
.or_else(|| r.get("u").and_then(|u| u.as_str()));
if let Some(hex) = uri.and_then(|u| u.strip_prefix("cas://sha256:")) {
out.push(hex.to_string());
}
}
}
out
}
#[derive(Default)]
struct IdentitySelection {
identity_ids: Vec<i64>,
identity_names: Vec<String>,
seqs: Vec<i64>,
}
enum ErasureSelector {
Identity { ns_id: i64, subject: String, text_mentions: bool },
OlderThan {
ns_id: Option<i64>,
cutoff_ms: i64,
gtype: Option<i64>,
limit: Option<usize>,
},
}
#[derive(Debug, Clone, Copy, Default)]
pub struct ErasureOptions {
pub text_mentions: bool,
}
#[derive(Debug, Clone)]
pub struct RetentionSweep {
pub ns: String,
pub days: f64,
pub outcome: RetentionOutcome,
}
#[derive(Debug, Clone)]
pub enum RetentionOutcome {
Swept(ErasureReport),
Skipped { why: String },
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct RetentionPolicy {
pub days: f64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub grain_type: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub because: Option<String>,
}
#[derive(Debug, Clone, Default)]
pub struct ErasureReport {
pub grains_erased: usize,
pub terms_removed: usize,
pub vocab_removed: usize,
pub blobs_reclaimed: usize,
pub erased_hashes: Vec<String>,
}
#[derive(Debug, Clone, serde::Serialize)]
pub struct CorpusExportRegistry {
pub manifest_hash: String,
pub export_id: String,
pub destination: String,
pub recipient: Option<String>,
pub exported_at_ms: i64,
pub subject_fingerprints: Vec<String>,
pub source_hashes: Vec<String>,
}
pub fn exports_touching_hashes(
exports: &[CorpusExportRegistry],
hashes: &[String],
) -> Vec<CorpusExportRegistry> {
let hashes: std::collections::HashSet<&str> =
hashes.iter().map(String::as_str).collect();
exports
.iter()
.filter(|export| {
export
.source_hashes
.iter()
.any(|hash| hashes.contains(hash.as_str()))
})
.cloned()
.collect()
}
#[derive(Debug, Default)]
pub struct SubjectReport {
pub identity_names: Vec<String>,
pub grains: Vec<DeserializedGrain>,
}
pub trait EmbedBackend: Send + Sync {
fn dim(&self) -> usize;
fn embed(&self, text: &str) -> Result<Vec<f32>>;
fn model(&self) -> &str {
"unspecified"
}
}
pub struct CommandAnonymize {
argv: Vec<String>,
id: String,
kind: String,
}
impl CommandAnonymize {
pub fn new(cmd: &str) -> Result<Self> {
let argv: Vec<String> = cmd.split_whitespace().map(str::to_string).collect();
if argv.is_empty() {
return Err(AreevError::Validation("empty anonymizer command".into()));
}
let mut backend = CommandAnonymize {
argv,
id: cmd.split_whitespace().next().unwrap_or("anonymizer").to_string(),
kind: "ner".to_string(),
};
let probe = backend.run(&serde_json::json!({"areev_anonymize": 1, "op": "probe"}))?;
let v: serde_json::Value = serde_json::from_str(&probe).map_err(|e| {
AreevError::Validation(format!(
"anonymizer probe returned invalid JSON ({e}) — refusing the backend \
at setup rather than failing closed on every covered read"
))
})?;
if let Some(id) = v["id"].as_str() {
backend.id = id.to_string();
}
if let Some(kind) = v["kind"].as_str() {
if !matches!(kind, "ner" | "llm") {
return Err(AreevError::Validation(format!(
"anonymizer probe declared unknown kind {kind:?} (expected ner or llm)"
)));
}
backend.kind = kind.to_string();
}
Ok(backend)
}
fn run(&self, payload: &serde_json::Value) -> Result<String> {
use areev_core::proc::{self, SpawnPolicy, StderrMode};
let mut cmd = std::process::Command::new(&self.argv[0]);
cmd.args(&self.argv[1..]);
let policy = SpawnPolicy::default().stderr(StderrMode::Inherit);
let out = proc::run(cmd, Some(payload.to_string().as_bytes()), &[], &policy)
.map_err(|e| AreevError::Storage(format!("spawn anonymizer {}: {e}", self.argv[0])))?;
if let Some(why) = out.failure(&format!("anonymizer {}", self.argv[0])) {
return Err(AreevError::Storage(why));
}
String::from_utf8(out.stdout)
.map_err(|e| AreevError::Storage(format!("anonymizer stdout not UTF-8: {e}")))
}
}
impl areev_core::anon::DetectorBackend for CommandAnonymize {
fn kind(&self) -> &str {
&self.kind
}
fn id(&self) -> &str {
&self.id
}
fn detect(&self, text: &str) -> Result<Vec<areev_core::anon::Detection>> {
let out = self.run(&serde_json::json!({
"areev_anonymize": 1, "op": "detect", "text": text
}))?;
let v: serde_json::Value = serde_json::from_str(&out).map_err(|e| {
AreevError::Validation(format!("anonymizer {} returned invalid JSON: {e}", self.id))
})?;
let mut detections: Vec<areev_core::anon::Detection> =
serde_json::from_value(v["detections"].clone()).map_err(|e| {
AreevError::Validation(format!(
"anonymizer {} detections malformed: {e}",
self.id
))
})?;
for d in &mut detections {
if d.detector.is_empty() {
d.detector = self.id.clone();
}
}
Ok(detections)
}
}
pub struct CommandEmbed {
argv: Vec<String>,
dim: usize,
model: String,
}
impl CommandEmbed {
pub fn new(cmd: &str, model: Option<&str>) -> Result<Self> {
let argv: Vec<String> = cmd.split_whitespace().map(str::to_string).collect();
if argv.is_empty() {
return Err(AreevError::Validation("embed command is empty".into()));
}
let mut ce = CommandEmbed {
argv,
dim: 0,
model: model.unwrap_or("command").to_string(),
};
let probe = ce.run("dimension probe")?;
if probe.is_empty() {
return Err(AreevError::Validation(
"embed command returned an empty vector".into(),
));
}
ce.dim = probe.len();
Ok(ce)
}
fn run(&self, text: &str) -> Result<Vec<f32>> {
use areev_core::proc::{self, SpawnPolicy, StderrMode};
let mut cmd = std::process::Command::new(&self.argv[0]);
cmd.args(&self.argv[1..]);
let policy = SpawnPolicy::default().stderr(StderrMode::Inherit);
let out = proc::run(cmd, Some(text.as_bytes()), &[], &policy)
.map_err(|e| AreevError::Storage(format!("embed command '{}': {e}", self.argv[0])))?;
if let Some(why) = out.failure(&format!("embed command '{}'", self.argv[0])) {
return Err(AreevError::Storage(why));
}
serde_json::from_slice::<Vec<f32>>(&out.stdout).map_err(|e| {
AreevError::Validation(format!(
"embed command output must be a JSON array of numbers: {e}"
))
})
}
}
impl EmbedBackend for CommandEmbed {
fn dim(&self) -> usize {
self.dim
}
fn embed(&self, text: &str) -> Result<Vec<f32>> {
let v = self.run(text)?;
if v.len() != self.dim {
return Err(AreevError::Validation(format!(
"embed command returned {} dims, expected {}",
v.len(),
self.dim
)));
}
Ok(v)
}
fn model(&self) -> &str {
&self.model
}
}
pub trait RerankBackend: Send + Sync {
fn rerank(&self, query: &str, docs: &[&str]) -> Result<Vec<f32>>;
fn model(&self) -> &str {
"unspecified"
}
}
pub trait QueryExpander: Send + Sync {
fn expand(&self, query: &str) -> Vec<String>;
}
pub struct EnglishExpander {
max_variants: usize,
}
impl Default for EnglishExpander {
fn default() -> Self {
Self { max_variants: 4 }
}
}
impl EnglishExpander {
pub fn new(max_variants: usize) -> Self {
Self { max_variants: max_variants.clamp(1, 16) }
}
fn synonyms(token: &str) -> &'static [&'static str] {
match token {
"cell" | "cellphone" => &["mobile", "phone"],
"mobile" => &["cell", "phone"],
"phone" => &["cell", "mobile", "telephone"],
"email" | "e-mail" => &["mail"],
"buy" | "bought" | "purchased" => &["purchase"],
"purchase" => &["buy"],
"car" | "automobile" => &["vehicle"],
"vehicle" => &["car"],
"doctor" | "physician" => &["doctor", "physician"],
"kid" | "kids" | "child" => &["child", "children"],
"spouse" => &["wife", "husband", "partner"],
"job" => &["work", "employer"],
"home" | "house" => &["residence", "address"],
"birthday" => &["birthdate", "born"],
"big" => &["large"],
"small" => &["little"],
_ => &[],
}
}
fn stem(token: &str) -> Option<String> {
let lower = token;
for suf in ["ing", "ed", "es", "s"] {
if lower.len() > suf.len() + 2 && lower.ends_with(suf) {
let stem = &lower[..lower.len() - suf.len()];
if stem.len() >= 3 {
return Some(stem.to_string());
}
}
}
None
}
}
impl QueryExpander for EnglishExpander {
fn expand(&self, query: &str) -> Vec<String> {
let tokens: Vec<String> = query
.split_whitespace()
.map(|t| t.trim_matches(|c: char| !c.is_alphanumeric()).to_lowercase())
.filter(|t| !t.is_empty())
.collect();
if tokens.is_empty() {
return Vec::new();
}
let mut variants: Vec<String> = Vec::new();
let mut seen: HashSet<String> = HashSet::new();
let original = tokens.join(" ");
seen.insert(original.clone());
for (i, tok) in tokens.iter().enumerate() {
for syn in Self::synonyms(tok) {
let mut v = tokens.clone();
v[i] = (*syn).to_string();
let s = v.join(" ");
if seen.insert(s.clone()) {
variants.push(s);
if variants.len() >= self.max_variants {
return variants;
}
}
}
}
let stemmed: Vec<String> = tokens
.iter()
.map(|t| Self::stem(t).unwrap_or_else(|| t.clone()))
.collect();
let s = stemmed.join(" ");
if seen.insert(s.clone()) {
variants.push(s);
}
variants.truncate(self.max_variants);
variants
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum RecentOrder {
#[default]
Seq,
CreatedAtDesc,
CreatedAtAsc,
}
#[derive(Debug, Clone, Copy, Default)]
pub struct RecallTuning {
pub query_expansion: bool,
pub rerank: bool,
pub diversity_lambda: Option<f32>,
pub include_superseded: bool,
}
#[derive(Debug, Clone)]
pub struct FactDraft {
pub subject: String,
pub relation: String,
pub object: String,
pub confidence: f64,
}
pub const DRAFT_DEFAULT_CONFIDENCE: f64 = 0.8;
impl FactDraft {
pub fn from_json_array(json: &str) -> Result<Vec<FactDraft>> {
let arr: Vec<serde_json::Value> = serde_json::from_str(json).map_err(|e| {
AreevError::Validation(format!(
"facts must be a JSON array of {{subject, relation, object, confidence}}: {e}"
))
})?;
arr.iter()
.enumerate()
.map(|(i, v)| {
let field = |k: &str| {
v.get(k)
.and_then(|x| x.as_str())
.unwrap_or("")
.trim()
.to_string()
};
let (subject, relation, object) = (field("subject"), field("relation"), field("object"));
if subject.is_empty() || relation.is_empty() || object.is_empty() {
return Err(AreevError::Validation(format!(
"facts[{i}] needs a non-empty subject, relation, and object"
)));
}
Ok(FactDraft {
subject,
relation,
object,
confidence: v
.get("confidence")
.and_then(|c| c.as_f64())
.unwrap_or(DRAFT_DEFAULT_CONFIDENCE)
.clamp(0.0, 1.0),
})
})
.collect()
}
}
#[derive(Debug, Clone, Default)]
pub struct FactAttribution<'a> {
pub verification_status: Option<&'a str>,
pub extractor_model: Option<&'a str>,
}
#[derive(Debug, Clone, Default)]
pub struct Capture<'a> {
pub observer: Option<&'a str>,
pub session_id: Option<&'a str>,
pub role: Option<&'a str>,
pub run_id: Option<&'a str>,
}
#[derive(Debug, Clone)]
pub struct RememberResult {
pub event: Hash,
pub facts: Vec<Hash>,
}
#[derive(Debug, Clone)]
pub struct VerifyReport {
pub integrity: String,
pub fts_notes: Vec<String>,
pub grains: usize,
pub hash_mismatches: usize,
pub undecodable: usize,
}
#[derive(Debug, Clone)]
pub struct StoreStats {
pub grains: usize,
pub current: usize,
pub triples: usize,
pub terms: usize,
pub ops: usize,
pub events_indexed: usize,
}
#[derive(Debug, Clone)]
pub struct HistoryEntry {
pub hash: Hash,
pub object: String,
pub created_at: i64,
pub confidence: f64,
pub superseded_by: Option<Hash>,
}
#[derive(Debug, Clone)]
pub struct ForkGroup {
pub namespace: String,
pub subject: String,
pub relation: String,
pub heads: Vec<Hash>,
}
const BUNDLE_MAGIC: &[u8; 4] = b"MGB1";
const BUNDLE_MAGIC_V2: &[u8; 4] = b"MGB2";
pub type AnonMappingRow = (String, String, std::collections::BTreeMap<String, String>);
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct TriggerState {
#[serde(skip_serializing_if = "Option::is_none")]
pub next_due_at: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_fired_at: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cursor: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub op_cursor: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub claimed_by: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub lease_until: Option<i64>,
pub fence: u64,
pub paused: bool,
pub exhausted: bool,
pub consecutive_failures: u32,
#[serde(skip_serializing_if = "Option::is_none")]
pub last_error: Option<String>,
#[serde(skip_serializing_if = "std::collections::BTreeMap::is_empty")]
pub partials: std::collections::BTreeMap<String, PartialMatch>,
}
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct PartialMatch {
pub members: std::collections::BTreeMap<String, i64>,
pub started_ms: i64,
}
impl TriggerState {
pub fn leased(&self, now_ms: i64) -> bool {
self.lease_until.is_some_and(|until| until > now_ms)
}
pub fn permits_firing(&self, now_ms: i64) -> bool {
!self.paused && !self.exhausted && self.next_due_at.is_none_or(|due| due <= now_ms)
}
}
const VAULT_PREFIX: &str = "vault:";
const REPLICABLE_META_PREFIXES: [&str; 5] =
["qry:", "tpl:", "retention:", "retention_floor:", "anon:"];
pub const RRF_K0: f64 = 60.0;
const REFINE_POOL: usize = 64;
fn version_lt(a: &str, b: &str) -> bool {
let parts = |v: &str| -> Vec<u32> {
v.split(['.', '-', '+'])
.map(|p| p.parse::<u32>().unwrap_or(0))
.collect()
};
let (pa, pb) = (parts(a), parts(b));
for i in 0..pa.len().max(pb.len()) {
let (x, y) = (pa.get(i).copied().unwrap_or(0), pb.get(i).copied().unwrap_or(0));
if x != y {
return x < y;
}
}
false
}
const LEGACY_MAX_GRAIN_BYTE: u8 = 0x0B;
const RETENTION_PREFIX: &str = "retention:";
const RETENTION_FLOOR_PREFIX: &str = "retention_floor:";
const HOLD_PREFIX: &str = "hold:";
const TRG_PREFIX: &str = "trg:";
pub const MAX_SUPERSESSION_CHAIN_HOPS: usize = 64;
const MIN_READER_VERSION_KEY: &str = "min_reader_version";
const LINK_INDEX_KEY: &str = "link_index";
const LINK_INDEX_VERSION: &str = "3";
const NS_REGISTRY_KEY: &str = "ns_registry";
const NS_REGISTRY_VERSION: &str = "1";
pub struct AreevOptions {
pub entity_relations: HashSet<String>,
pub index_text: bool,
pub encryption_key: Option<[u8; 32]>,
pub anon_key: Option<[u8; 32]>,
pub telemetry: TelemetryMode,
pub read_only: bool,
}
impl Default for AreevOptions {
fn default() -> Self {
let ents = [
"mg:delegates_to",
"mg:owned_by",
"mg:assigned_to",
"mg:depends_on",
"mg:handed_off_to",
"mg:capable_of",
"delegates_to",
"reports_to",
"part_of",
"knows",
];
AreevOptions {
entity_relations: ents.iter().map(|s| s.to_string()).collect(),
index_text: true,
encryption_key: None,
anon_key: None,
telemetry: TelemetryMode::Off,
read_only: false,
}
}
}
const SCHEMA: &[&str] = &[
"CREATE TABLE IF NOT EXISTS meta(k TEXT PRIMARY KEY, v TEXT)",
"CREATE TABLE IF NOT EXISTS terms(id INTEGER PRIMARY KEY, term TEXT UNIQUE)",
"CREATE TABLE IF NOT EXISTS grains(
seq INTEGER PRIMARY KEY,
hash BLOB,
ns INTEGER, gtype INTEGER, created_at INTEGER,
s INTEGER, p INTEGER, o INTEGER,
vf INTEGER, vt INTEGER,
svf INTEGER, svt INTEGER,
superseded_by BLOB, supersedes BLOB,
text TEXT,
blob BLOB NOT NULL)",
"CREATE UNIQUE INDEX IF NOT EXISTS idx_grains_hash ON grains(hash)",
"CREATE INDEX IF NOT EXISTS idx_grains_ns_s ON grains(ns, s, p)",
"CREATE TABLE IF NOT EXISTS embeddings(seq INTEGER PRIMARY KEY, vec BLOB)",
"CREATE TABLE IF NOT EXISTS fts_vocab(id INTEGER PRIMARY KEY, term TEXT UNIQUE)",
"CREATE TABLE IF NOT EXISTS fts_post(term INTEGER, seq INTEGER, ns INTEGER, tf INTEGER)",
"CREATE INDEX IF NOT EXISTS idx_fts_post ON fts_post(term, ns)",
"CREATE INDEX IF NOT EXISTS idx_fts_post_seq ON fts_post(seq)",
"CREATE TABLE IF NOT EXISTS fts_doc(seq INTEGER PRIMARY KEY, len INTEGER)",
"CREATE TABLE IF NOT EXISTS triples(ns INTEGER, s INTEGER, p INTEGER, o INTEGER, seq INTEGER, cur INTEGER)",
"CREATE INDEX IF NOT EXISTS idx_spo ON triples(ns,s,p,o,seq)",
"CREATE INDEX IF NOT EXISTS idx_pos ON triples(ns,p,o,s,seq)",
"CREATE INDEX IF NOT EXISTS idx_triples_seq ON triples(seq)",
"CREATE TABLE IF NOT EXISTS osp(ns INTEGER, o INTEGER, s INTEGER, p INTEGER, seq INTEGER, cur INTEGER)",
"CREATE INDEX IF NOT EXISTS idx_osp ON osp(ns,o,s)",
"CREATE INDEX IF NOT EXISTS idx_osp_seq ON osp(seq)",
"CREATE TABLE IF NOT EXISTS entity_latest(ns INTEGER, s INTEGER, p INTEGER, o INTEGER, seq INTEGER, hash BLOB, PRIMARY KEY(ns,s,p))",
"CREATE TABLE IF NOT EXISTS heads(ns INTEGER, s INTEGER, p INTEGER, seq INTEGER, hash BLOB, created_at INTEGER, PRIMARY KEY(ns,s,p,seq))",
"CREATE TABLE IF NOT EXISTS oplog(op_seq INTEGER PRIMARY KEY, hlc INTEGER, op INTEGER, hash BLOB)",
"CREATE TABLE IF NOT EXISTS thread_idx(ns INTEGER, session INTEGER, seq INTEGER)",
"CREATE INDEX IF NOT EXISTS idx_thread ON thread_idx(ns, session, seq)",
"CREATE TABLE IF NOT EXISTS prov_idx(ns INTEGER, parent BLOB, seq INTEGER)",
"CREATE INDEX IF NOT EXISTS idx_prov ON prov_idx(ns, parent, seq)",
"CREATE TABLE IF NOT EXISTS run_idx(ns INTEGER, run INTEGER, seq INTEGER)",
"CREATE INDEX IF NOT EXISTS idx_run ON run_idx(ns, run, seq)",
"CREATE TABLE IF NOT EXISTS corpus_idx(seq INTEGER PRIMARY KEY)",
"CREATE TABLE IF NOT EXISTS ns_reg(ns INTEGER PRIMARY KEY, n INTEGER NOT NULL)",
];
fn pi(x: i64) -> Value {
Value::Integer(x)
}
fn pb(b: Vec<u8>) -> Value {
Value::Blob(b)
}
fn pt(s: &str) -> Value {
Value::Text(s.to_string())
}
fn opt_i(v: Option<i64>) -> Value {
match v {
Some(x) => Value::Integer(x),
None => Value::Null,
}
}
#[cfg(test)]
fn hex32(k: &[u8; 32]) -> String {
use std::fmt::Write;
let mut s = String::with_capacity(64);
for b in k {
let _ = write!(s, "{b:02x}");
}
s
}
const KDF_M_COST: u32 = 19_456; const KDF_T_COST: u32 = 2; const KDF_P_COST: u32 = 1; const KDF_SALT_LEN: usize = 16;
fn kdf_err<E: std::fmt::Display>(e: E) -> AreevError {
AreevError::CryptoError(e.to_string())
}
fn load_or_create_kdf_sidecar(sidecar: &str) -> Result<([u8; KDF_SALT_LEN], u32, u32, u32)> {
match std::fs::read_to_string(sidecar) {
Ok(text) => parse_kdf_sidecar(&text, sidecar),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
let mut salt = [0u8; KDF_SALT_LEN];
getrandom::getrandom(&mut salt).map_err(kdf_err)?;
let line = format!(
"v1 argon2id {} {KDF_M_COST} {KDF_T_COST} {KDF_P_COST}\n",
hex::encode(salt)
);
match std::fs::OpenOptions::new().write(true).create_new(true).open(sidecar) {
Ok(mut f) => {
use std::io::Write;
f.write_all(line.as_bytes()).map_err(kdf_err)?;
Ok((salt, KDF_M_COST, KDF_T_COST, KDF_P_COST))
}
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
let text = std::fs::read_to_string(sidecar).map_err(kdf_err)?;
parse_kdf_sidecar(&text, sidecar)
}
Err(e) => Err(kdf_err(e)),
}
}
Err(e) => Err(kdf_err(e)),
}
}
fn parse_kdf_sidecar(text: &str, sidecar: &str) -> Result<([u8; KDF_SALT_LEN], u32, u32, u32)> {
let toks: Vec<&str> = text.split_whitespace().collect();
if toks.len() != 6 || toks[0] != "v1" || toks[1] != "argon2id" {
return Err(kdf_err(format!("malformed KDF sidecar: {sidecar}")));
}
let salt_bytes = hex::decode(toks[2]).map_err(kdf_err)?;
if salt_bytes.len() != KDF_SALT_LEN {
return Err(kdf_err("KDF salt has wrong length"));
}
let mut salt = [0u8; KDF_SALT_LEN];
salt.copy_from_slice(&salt_bytes);
let m = toks[3].parse::<u32>().map_err(kdf_err)?;
let t = toks[4].parse::<u32>().map_err(kdf_err)?;
let p = toks[5].parse::<u32>().map_err(kdf_err)?;
if !(8..=1_048_576).contains(&m) || !(1..=16).contains(&t) || !(1..=16).contains(&p) {
return Err(kdf_err("KDF parameters out of range"));
}
Ok((salt, m, t, p))
}
pub fn read_only_requires_existing(path: &str, read_only: bool) -> Result<()> {
if read_only && !std::path::Path::new(path).exists() {
return Err(AreevError::ReadOnlyOpenFailed(format!(
"memory file {path:?} does not exist — a read-only open never creates one. \
Open it read-write once (drop --read-only / read_only: true) to create the \
memory, then retry read-only"
)));
}
Ok(())
}
impl Areev {
pub fn derive_key_for(path: &str, passphrase: &str) -> Result<zeroize::Zeroizing<[u8; 32]>> {
if passphrase.trim().is_empty() {
return Err(kdf_err("passphrase must not be empty or whitespace-only"));
}
let sidecar = format!("{path}.kdf");
let (salt, m, t, p) = load_or_create_kdf_sidecar(&sidecar)?;
let params = argon2::Params::new(m, t, p, Some(32)).map_err(kdf_err)?;
let argon = argon2::Argon2::new(argon2::Algorithm::Argon2id, argon2::Version::V0x13, params);
let mut key = zeroize::Zeroizing::new([0u8; 32]);
argon
.hash_password_into(passphrase.as_bytes(), &salt, &mut key[..])
.map_err(kdf_err)?;
Ok(key)
}
pub fn open_with_passphrase(path: &str, passphrase: &str) -> Result<Self> {
let key = Self::derive_key_for(path, passphrase)?;
Self::open_with(
path,
AreevOptions { encryption_key: Some(*key), ..AreevOptions::default() },
)
}
pub fn open_with_passphrase_telemetry(
path: &str,
passphrase: &str,
telemetry: TelemetryMode,
) -> Result<Self> {
let key = Self::derive_key_for(path, passphrase)?;
Self::open_with(
path,
AreevOptions {
encryption_key: Some(*key),
telemetry,
..AreevOptions::default()
},
)
}
}
pub const EMBEDDING_BATCH_CHUNK: usize = 200;
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub struct VectorRecallReport {
pub index: Option<String>,
pub ef_search: Option<usize>,
pub k: usize,
pub queries: usize,
pub recall: Option<f32>,
}
#[derive(serde::Deserialize)]
struct EmbeddingItemRaw {
hash: String,
vector: Vec<f32>,
}
pub fn parse_embedding_items(json: &str) -> Result<Vec<(Hash, Vec<f32>)>> {
let rows: Vec<EmbeddingItemRaw> = serde_json::from_str(json).map_err(|e| {
AreevError::Validation(format!(
"items must be a JSON array of {{\"hash\", \"vector\"}} objects: {e}"
))
})?;
rows.into_iter()
.enumerate()
.map(|(i, r)| {
let hash = Hash::from_hex(&r.hash)
.map_err(|e| AreevError::Validation(format!("item {i}: {e}")))?;
Ok((hash, r.vector))
})
.collect()
}
fn vec_to_json(v: &[f32]) -> String {
let mut s = String::with_capacity(v.len() * 8);
s.push('[');
for (i, x) in v.iter().enumerate() {
if i > 0 { s.push(','); }
s.push_str(&format!("{x}"));
}
s.push(']');
s
}
fn now_ms() -> i64 {
if let Ok(value) = std::env::var("AREEV_LOOP_NOW_MS") {
return value.trim().parse().unwrap_or_else(|_| {
panic!("AREEV_LOOP_NOW_MS is set but not epoch milliseconds: {value:?}")
});
}
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
#[cfg(test)]
fn v_i64(v: &Value) -> Option<i64> {
match v {
Value::Integer(i) => Some(*i),
_ => None,
}
}
#[cfg(test)]
fn v_blob(v: &Value) -> Option<Vec<u8>> {
match v {
Value::Blob(b) => Some(b.clone()),
_ => None,
}
}
#[cfg(test)]
fn v_f64(v: &Value) -> Option<f64> {
match v {
Value::Real(r) => Some(*r),
Value::Integer(i) => Some(*i as f64),
_ => None,
}
}
fn seq_csv(seqs: &[i64]) -> String {
let mut s = String::with_capacity(seqs.len() * 6);
for (i, x) in seqs.iter().enumerate() {
if i > 0 {
s.push(',');
}
s.push_str(&x.to_string());
}
s
}
struct GrainPrep {
blob: Vec<u8>,
hash: Hash,
ns_id: i64,
s: Option<i64>,
p: Option<i64>,
o: Option<i64>,
osp: bool,
session: Option<i64>,
vf: Option<i64>,
vt: Option<i64>,
created: i64,
gtype: i64,
text: Option<String>,
embedding: Option<Vec<f32>>,
tokens: Vec<(i64, i64)>,
doc_len: i64,
links: Vec<(i64, i64, i64)>,
subject_only: Option<i64>,
run: Option<i64>,
parent: Option<Vec<u8>>,
corpus_export: bool,
}
struct GrainView {
ns: String,
subject: Option<String>,
relation: Option<String>,
object: Option<String>,
session: Option<String>,
vf: Option<i64>,
vt: Option<i64>,
created_at: i64,
gtype: u8,
links: Vec<(String, String)>,
run: Option<String>,
parent: Option<String>,
corpus_export: bool,
}
fn extract_view(view: &DeserializedGrain) -> GrainView {
GrainView {
ns: view.get_str("namespace").unwrap_or("shared").to_string(),
subject: view.get_str("subject").map(str::to_string),
relation: view.get_str("relation").map(str::to_string),
object: view.get_str("object").map(str::to_string),
session: view.get_str("session_id").map(str::to_string),
vf: view.get_i64("valid_from"),
vt: view.get_i64("valid_to"),
created_at: view.get_i64("created_at").unwrap_or_else(now_ms),
gtype: view.grain_type as u8,
run: view.get_str("run_id").map(str::to_string),
parent: view.get_str("derived_from").map(str::to_string),
corpus_export: view.grain_type == areev_core::types::GrainType::Observation
&& view
.fields
.get("context")
.and_then(|value| value.get("corpus_export"))
.and_then(|value| value.as_bool())
== Some(true),
links: view
.fields
.get("related_to")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|l| {
Some((
l.get("relation_type")?.as_str()?.to_string(),
l.get("hash")?.as_str()?.to_string(),
))
})
.collect()
})
.unwrap_or_default(),
}
}
const BM25_K1: f64 = 1.2;
const BM25_B: f64 = 0.75;
const MAX_TOKEN_LEN: usize = 64;
fn tokenize(text: &str) -> Vec<String> {
text.split(|c: char| !c.is_alphanumeric())
.filter(|t| !t.is_empty() && t.chars().count() <= MAX_TOKEN_LEN)
.map(|t| t.to_lowercase())
.collect()
}
fn token_freqs(text: &str) -> (Vec<(String, i64)>, i64) {
let tokens = tokenize(text);
let len = tokens.len() as i64;
let mut freqs: HashMap<String, i64> = HashMap::new();
for t in tokens {
*freqs.entry(t).or_insert(0) += 1;
}
(freqs.into_iter().collect(), len)
}
fn projected_text(view: &DeserializedGrain) -> Option<String> {
if let Some(et) = view.get_str("embedding_text") {
if !et.trim().is_empty() {
return Some(et.to_string());
}
}
let mut parts: Vec<String> = Vec::new();
if let (Some(s), Some(r), Some(o)) = (
view.get_str("subject"),
view.get_str("relation"),
view.get_str("object"),
) {
parts.push(format!("{s} {r} {o}"));
}
if let Some(c) = view.get_str("content") {
parts.push(c.to_string());
}
if parts.is_empty() {
None
} else {
Some(parts.join(" "))
}
}
pub fn read_blob_offline(db_path: &str, uri: &str) -> Result<Option<Vec<u8>>> {
use sha2::{Digest, Sha256};
let hex = Areev::cas_hex(uri)?;
let raw = if is_pg_dsn(db_path) {
read_blob_pg(db_path, hex, uri)?
} else {
let dir = std::path::PathBuf::from(format!("{db_path}.blobs"));
std::fs::read(fs_blob_path(&dir, hex))
.map_err(|_| AreevError::Storage(format!("blob missing: {uri}")))?
};
if blobcrypt::is_sealed(&raw) {
return Ok(None);
}
if hex::encode(Sha256::digest(&raw)) != hex {
return Err(AreevError::Storage(format!("blob corrupt: {uri}")));
}
Ok(Some(raw))
}
pub fn is_pg_dsn(locator: &str) -> bool {
locator.starts_with("postgres://") || locator.starts_with("postgresql://")
}
#[cfg(feature = "postgres")]
fn read_blob_pg(dsn: &str, hex: &str, uri: &str) -> Result<Vec<u8>> {
let (url, schema) = pg::split_schema_url(dsn)?;
let raw = hex::decode(hex).map_err(|e| AreevError::Storage(e.to_string()))?;
let (pool, _) = pg::PgPool::for_url(&url)?;
let mut co = pool.checkout()?;
let sql = pg::qualify_tables("SELECT body FROM blobs WHERE hash = $1", &schema);
let rows = pool.block_on(co.client().query(sql.as_str(), &[&raw])).map_err(pg::pg_err);
let rows = match rows {
Ok(rows) => rows,
Err(e) => {
co.note(&e);
return Err(e);
}
};
let row = rows
.first()
.ok_or_else(|| AreevError::Storage(format!("blob missing: {uri}")))?;
row.try_get::<_, Vec<u8>>(0).map_err(pg::pg_err)
}
#[cfg(not(feature = "postgres"))]
fn read_blob_pg(_dsn: &str, _hex: &str, _uri: &str) -> Result<Vec<u8>> {
Err(AreevError::Storage(
"this build lacks the postgres backend — rebuild with --features postgres-tls".into(),
))
}
enum BlobStore {
Fs(std::path::PathBuf),
#[cfg_attr(not(feature = "postgres"), allow(dead_code))]
Table,
}
fn fs_blob_path(dir: &std::path::Path, hex: &str) -> std::path::PathBuf {
dir.join(&hex[..2]).join(&hex[2..])
}
static OPEN_FILES: std::sync::Mutex<Option<HashSet<std::path::PathBuf>>> =
std::sync::Mutex::new(None);
struct OpenFileGuard(std::path::PathBuf);
impl OpenFileGuard {
fn claim(path: &str) -> Result<Self> {
let key = std::path::absolute(path).unwrap_or_else(|_| std::path::PathBuf::from(path));
let mut held = OPEN_FILES.lock().unwrap_or_else(|e| e.into_inner());
let set = held.get_or_insert_with(HashSet::new);
if !set.insert(key.clone()) {
return Err(AreevError::StoreBusy(format!(
"{} is already open in this process. One memory is one writer: \
share the handle (it is safe across threads) instead of opening \
a second one — a second handle keeps its own sequence and \
dictionary allocators, and their first collision fails the \
OTHER handle's write",
key.display()
)));
}
Ok(Self(key))
}
}
impl Drop for OpenFileGuard {
fn drop(&mut self) {
if let Ok(mut held) = OPEN_FILES.lock() {
if let Some(set) = held.as_mut() {
set.remove(&self.0);
}
}
}
}
pub struct Areev {
db: Box<dyn Db>,
dict: HashMap<String, i64>,
next_term: i64,
next_seq: i64,
next_op: i64,
hlc_last: i64,
entity_rels: HashSet<String>,
index_text: bool,
read_only: bool,
fts_vocab: HashMap<String, i64>,
fts_docs: i64,
fts_total_len: i64,
fts_deferred: bool,
embedder: Option<Box<dyn EmbedBackend>>,
reranker: Option<Box<dyn RerankBackend>>,
expander: Option<Box<dyn QueryExpander>>,
meta_embed: Option<(String, usize)>,
warnings: Vec<String>,
needs_min_reader_stamp: bool,
blob_store: BlobStore,
blob_key: Option<zeroize::Zeroizing<[u8; 32]>>,
telemetry: Option<Telemetry>,
anon: anon_gate::AnonGate,
ambient_run_id: Option<String>,
_file_guard: Option<OpenFileGuard>,
}
impl Areev {
pub fn open(path: &str) -> Result<Self> {
Self::open_internal(path, None, TelemetryMode::Off)
}
pub fn open_with(path: &str, opts: AreevOptions) -> Result<Self> {
let telemetry = opts.telemetry;
Self::open_internal(path, Some(opts), telemetry)
}
pub fn open_with_telemetry(path: &str, telemetry: TelemetryMode) -> Result<Self> {
Self::open_internal(path, None, telemetry)
}
pub fn open_encrypted(path: &str, key: [u8; 32]) -> Result<Self> {
Self::open_with(
path,
AreevOptions { encryption_key: Some(key), ..AreevOptions::default() },
)
}
fn open_internal(
path: &str,
explicit: Option<AreevOptions>,
telemetry_mode: TelemetryMode,
) -> Result<Self> {
let read_only = explicit.as_ref().is_some_and(|o| o.read_only);
read_only_requires_existing(path, read_only)?;
let telemetry_overridden = read_only && telemetry_mode != TelemetryMode::Off;
let telemetry_mode = if telemetry_overridden { TelemetryMode::Off } else { telemetry_mode };
let enc_key = explicit
.as_ref()
.and_then(|o| o.encryption_key)
.map(zeroize::Zeroizing::new);
let guard = OpenFileGuard::claim(path)?;
let dbh: Box<dyn Db> = Box::new(TursoDb::open(path, enc_key.as_deref())?);
for sql in SCHEMA {
dbh.execute(sql, vec![])?;
}
let mut warnings: Vec<String> = Vec::new();
if telemetry_overridden {
warnings.push(
"telemetry disabled: this open asked for a telemetry sidecar, but a \
read-only handle attaches none (its flush is itself a write) — the \
request was downgraded to off rather than refused"
.into(),
);
}
if enc_key.is_some() {
warnings.push(
"encryption-at-rest ON (AES-256-GCM): the memory database and the .blobs CAS \
sidecar are both encrypted (blobs under an HKDF-derived key). Attachments \
written by a build before blob encryption landed stay plaintext until \
migrated — run `areev blobs encrypt` on such a file"
.into(),
);
}
let blob_dir = std::path::PathBuf::from(format!("{}.blobs", path));
std::fs::create_dir_all(&blob_dir).map_err(db_err)?;
let telemetry = match telemetry_mode {
TelemetryMode::Off => None,
mode => Some(Telemetry::open(path, enc_key.as_deref(), mode)?),
};
Self::finish_open(dbh, explicit, BlobStore::Fs(blob_dir), telemetry, warnings, Some(guard))
}
#[cfg(feature = "postgres")]
pub fn open_postgres(url: &str, schema: &str) -> Result<Self> {
Self::open_postgres_internal(url, schema, None, TelemetryMode::Off)
}
#[cfg(feature = "postgres")]
pub fn open_postgres_with(url: &str, schema: &str, opts: AreevOptions) -> Result<Self> {
let telemetry = opts.telemetry;
Self::open_postgres_internal(url, schema, Some(opts), telemetry)
}
#[cfg(feature = "postgres")]
pub fn open_postgres_with_telemetry(
url: &str,
schema: &str,
telemetry: TelemetryMode,
) -> Result<Self> {
Self::open_postgres_internal(url, schema, None, telemetry)
}
#[cfg(feature = "postgres")]
fn open_postgres_internal(
url: &str,
schema: &str,
explicit: Option<AreevOptions>,
telemetry_mode: TelemetryMode,
) -> Result<Self> {
if let Some(o) = &explicit {
if o.encryption_key.is_some() {
return Err(AreevError::Validation(
"encryption_key is a file-backend capability (page cipher); on the postgres \
backend use TDE/pgcrypto at the deployment layer. If you wanted it for \
ANONYMIZATION — deterministic value-derived pseudonyms and the mapping \
vault — use AreevOptions::anon_key instead: a host-supplied 32-byte key \
(from your KMS) that works on every backend and is independent of \
encryption at rest"
.into(),
));
}
}
let read_only = explicit.as_ref().is_some_and(|o| o.read_only);
let telemetry_overridden = read_only && telemetry_mode != TelemetryMode::Off;
let telemetry_mode = if telemetry_overridden { TelemetryMode::Off } else { telemetry_mode };
let mut warnings: Vec<String> = Vec::new();
if telemetry_overridden {
warnings.push(
"telemetry disabled: this open asked for a telemetry sidecar, but a \
read-only handle attaches none (its flush is itself a write) — the \
request was downgraded to off rather than refused"
.into(),
);
}
let mut bootstrap: Vec<&str> = Vec::with_capacity(pg::PG_SCHEMA.len() + pg::PG_SEED.len());
bootstrap.extend_from_slice(pg::PG_SCHEMA);
bootstrap.extend_from_slice(pg::PG_SEED);
let pg = pg::PgDb::open(url, schema, &bootstrap, read_only, Some(pg::STORE_STAMP))?;
if let Some(w) = pg.pool_warning() {
warnings.push(w.to_string());
}
let dbh: Box<dyn Db> = Box::new(pg);
let telemetry = match telemetry_mode {
TelemetryMode::Off => None,
mode => Some(Telemetry::open_pg(url, schema, mode)?),
};
Self::finish_open(dbh, explicit, BlobStore::Table, telemetry, warnings, None)
}
fn finish_open(
dbh: Box<dyn Db>,
explicit: Option<AreevOptions>,
blob_store: BlobStore,
telemetry: Option<Telemetry>,
mut warnings: Vec<String>,
file_guard: Option<OpenFileGuard>,
) -> Result<Self> {
let blob_key = explicit
.as_ref()
.and_then(|o| o.encryption_key.as_ref())
.map(blobcrypt::derive_blob_key);
let page_key = explicit.as_ref().and_then(|o| o.encryption_key.as_ref());
let anon_root = explicit
.as_ref()
.and_then(|o| o.anon_key.as_ref())
.or(page_key);
let anon_memory_key: Option<[u8; 32]> = anon_root.map(|pk| {
let hk = hkdf::Hkdf::<sha2::Sha256>::new(None, pk);
let mut out = [0u8; 32];
hk.expand(b"areev.anon.memory.v1", &mut out).expect("32-byte HKDF output");
out
});
let anon_vault_key: Option<[u8; 32]> = anon_root.map(|pk| {
let hk = hkdf::Hkdf::<sha2::Sha256>::new(None, pk);
let mut out = [0u8; 32];
hk.expand(b"areev.vault.v1", &mut out).expect("32-byte HKDF output");
out
});
let anon_session_key: [u8; 32] = match anon_root {
Some(pk) => {
let hk = hkdf::Hkdf::<sha2::Sha256>::new(None, pk);
let mut out = [0u8; 32];
hk.expand(b"areev.anon.session.v1", &mut out)
.expect("32-byte HKDF output");
out
}
None => {
let mut out = [0u8; 32];
getrandom::getrandom(&mut out).map_err(|e| {
AreevError::CryptoError(format!("anon session key: {e}"))
})?;
out
}
};
let meta: HashMap<String, String> = {
let mut m = HashMap::new();
for row in dbh.query("SELECT k, v FROM meta", vec![])? {
if let (Some(k), Some(v)) = (row.text(0), row.text(1)) {
m.insert(k.to_string(), v.to_string());
}
}
m
};
let declared_text = meta.get("text_index").map(|v| v == "1");
let declared_rels: Option<HashSet<String>> = meta
.get("entity_relations")
.and_then(|v| serde_json::from_str::<Vec<String>>(v).ok())
.map(|v| v.into_iter().collect());
let meta_embed = match (
meta.get("embedding_model"),
meta.get("embedding_dim").and_then(|d| d.parse::<usize>().ok()),
) {
(Some(m), Some(d)) => Some((m.clone(), d)),
_ => None,
};
let mut opts = match explicit {
Some(o) if o.read_only => {
if let Some(d) = declared_text {
if d != o.index_text {
return Err(AreevError::ReadOnly(format!(
"requested index_text={} differs from the file's declared \
index_text={} — a read-only open cannot re-stamp it. Open \
read-write once to change the declaration, or drop the explicit \
index_text setting",
o.index_text, d
)));
}
}
if let Some(ref d) = declared_rels {
if *d != o.entity_relations {
return Err(AreevError::ReadOnly(
"requested entity_relations differ from the file's declared set — \
a read-only open cannot re-stamp it"
.into(),
));
}
}
o
}
Some(o) => {
if let Some(d) = declared_text {
if d != o.index_text {
warnings.push(format!(
"file declared text_index={}; explicit open changed it to {} (re-stamped) — \
grains written under the old setting keep their old indexing",
if d { "on" } else { "off" },
if o.index_text { "on" } else { "off" },
));
}
}
if let Some(ref d) = declared_rels {
if *d != o.entity_relations {
warnings.push(
"file-declared entity_relations differ from explicit options (re-stamped) — \
OSP rows indexed under the old set are unchanged"
.into(),
);
}
}
o
}
None => AreevOptions {
index_text: declared_text.unwrap_or(true),
entity_relations: declared_rels
.unwrap_or_else(|| AreevOptions::default().entity_relations),
encryption_key: None,
anon_key: None,
telemetry: TelemetryMode::Off,
read_only: false,
},
};
opts.encryption_key.zeroize();
if !opts.read_only {
let want_text = if opts.index_text { "1" } else { "0" };
if meta.get("text_index").map(String::as_str) != Some(want_text) {
dbh.execute(
"INSERT OR REPLACE INTO meta(k, v) VALUES ('text_index', ?1)",
vec![pt(want_text)],
)?;
}
let mut rels: Vec<&String> = opts.entity_relations.iter().collect();
rels.sort();
let rels = serde_json::to_string(&rels).unwrap_or_else(|_| "[]".into());
if meta.get("entity_relations") != Some(&rels) {
dbh.execute(
"INSERT OR REPLACE INTO meta(k, v) VALUES ('entity_relations', ?1)",
vec![pt(&rels)],
)?;
}
if dbh.may_have_legacy_fts_index() {
let _ = dbh.execute("DROP INDEX idx_fts", vec![]);
}
}
let seeds = dbh.seeds_state_at_open();
let mut dict = HashMap::new();
let mut next_term = 1i64;
let one = |sql: &'static str| -> Result<i64> {
Ok(dbh.query(sql, vec![])?.first().and_then(|r| r.i64(0)).unwrap_or(0))
};
let (next_seq, next_op, hlc_last, fts_docs, fts_total_len, indexed_text, grain_count) =
if seeds {
for row in dbh.query("SELECT id, term FROM terms", vec![])? {
let id = row.i64(0).unwrap_or(0);
if let Some(t) = row.text(1) {
dict.insert(t.to_string(), id);
}
next_term = next_term.max(id + 1);
}
(
one("SELECT COALESCE(MAX(seq),0) FROM grains")? + 1,
one("SELECT COALESCE(MAX(op_seq),0) FROM oplog")? + 1,
one("SELECT COALESCE(MAX(hlc),0) FROM oplog")?,
one("SELECT COUNT(*) FROM fts_doc")?,
one("SELECT COALESCE(SUM(len),0) FROM fts_doc")?,
one("SELECT COUNT(*) FROM grains WHERE text IS NOT NULL")?,
one("SELECT COUNT(*) FROM grains")?,
)
} else {
(1, 1, 0, 0, 0, 0, 0)
};
let mut store = Areev {
db: dbh,
dict,
next_term,
next_seq,
next_op,
hlc_last,
entity_rels: opts.entity_relations,
index_text: opts.index_text,
read_only: opts.read_only,
fts_vocab: HashMap::new(),
fts_docs,
fts_total_len,
fts_deferred: false,
embedder: None,
reranker: None,
expander: None,
meta_embed,
warnings,
needs_min_reader_stamp: !meta.contains_key(MIN_READER_VERSION_KEY),
blob_store,
blob_key,
telemetry,
anon: anon_gate::AnonGate::new(anon_session_key, anon_memory_key, anon_vault_key),
ambient_run_id: None,
_file_guard: file_guard,
};
store.reload_anon_policies()?;
if let Some(w) = store.anon.poison_warning() {
store.warnings.push(w.to_string());
}
if let Some(req) = meta.get(MIN_READER_VERSION_KEY) {
if version_lt(env!("CARGO_PKG_VERSION"), req) {
store.warnings.push(format!(
"this memory declares min_reader_version {req} but this build is {} — it \
contains grain types this version cannot decode, and reads that touch them \
will fail. Upgrade areev to {req} or later",
env!("CARGO_PKG_VERSION")
));
}
}
let needs_text_rebuild = store.index_text
&& if seeds {
indexed_text > 0 && store.fts_docs == 0
} else {
store.db.query("SELECT seq FROM fts_doc LIMIT 1", vec![])?.is_empty()
&& !store
.db
.query("SELECT seq FROM grains WHERE text IS NOT NULL LIMIT 1", vec![])?
.is_empty()
};
if needs_text_rebuild {
if store.read_only {
store.warnings.push(
"text index needs a one-time rebuild (this file predates the current BM25 \
leg) but this is a read-only open, which never writes — free-text recall \
will answer empty until an operator opens the memory read-write once"
.into(),
);
} else {
store.rebuild_text_index()?;
let indexed = store.fts_docs;
store.warnings.push(format!(
"text index rebuilt on open ({indexed} grains): this file was written when the \
BM25 leg used Turso's experimental FTS index, which is no longer used. \
One-time; later opens skip it"
));
}
}
if meta.get(LINK_INDEX_KEY).map(String::as_str) != Some(LINK_INDEX_VERSION) {
let grain_count = if seeds {
grain_count
} else {
store
.db
.query("SELECT COUNT(*) FROM grains", vec![])?
.first()
.and_then(|r| r.i64(0))
.unwrap_or(0)
};
if store.read_only {
if grain_count > 0 {
store.warnings.push(
"link indexes (provenance/run correlation/related_to) need a one-time \
rebuild but this is a read-only open, which never writes — those reads \
will answer empty until an operator opens the memory read-write once, \
or runs `areev reindex`"
.into(),
);
}
} else if grain_count > 0 {
let rows = store.rebuild_link_indexes()?;
store.warnings.push(format!(
"link indexes rebuilt on open ({rows} rows across {grain_count} grains): this \
file was written before reverse provenance, run correlation and related_to \
cross-links were indexed. One-time; later opens skip it"
));
} else {
store.meta_put(LINK_INDEX_KEY, LINK_INDEX_VERSION)?;
}
}
if meta.get(NS_REGISTRY_KEY).map(String::as_str) != Some(NS_REGISTRY_VERSION) {
let grain_count = if seeds {
grain_count
} else {
store
.db
.query("SELECT COUNT(*) FROM grains", vec![])?
.first()
.and_then(|r| r.i64(0))
.unwrap_or(0)
};
if store.read_only {
if grain_count > 0 {
store.warnings.push(
"namespace registry needs a one-time rebuild but this is a read-only \
open, which never writes — prefix-scoped recall (\"org.*\") will answer \
empty until an operator opens the memory read-write once, or runs \
`areev reindex`"
.into(),
);
}
} else {
store.rebuild_ns_registry()?;
if grain_count > 0 {
store.warnings.push(format!(
"namespace registry rebuilt on open ({grain_count} grains): this file was \
written before namespaces were registered for prefix scoping. One-time; \
later opens skip it"
));
}
}
}
Ok(store)
}
fn check_writable(&self, op: &str) -> Result<()> {
if self.read_only {
return Err(AreevError::ReadOnly(format!(
"cannot {op} — this memory handle was opened with read_only: true"
)));
}
Ok(())
}
fn rebuild_ns_registry(&mut self) -> Result<()> {
self.check_writable("reindex the namespace registry")?;
let dbr = self.db.as_ref();
with_txn(dbr, || {
dbr.reserve_write(0, 0, 0, now_ms() << 16, 0)?;
dbr.execute("DELETE FROM ns_reg", vec![])?;
dbr.execute(
"INSERT INTO ns_reg(ns, n) SELECT ns, COUNT(*) FROM grains GROUP BY ns",
vec![],
)?;
Ok(())
})?;
self.meta_put(NS_REGISTRY_KEY, NS_REGISTRY_VERSION)
}
pub fn set_embedder(&mut self, e: Box<dyn EmbedBackend>) {
let (model, dim) = (e.model().to_string(), e.dim());
if self.read_only && self.meta_embed.is_none() {
self.warnings.push(format!(
"vector recall disabled — embedder {model}@{dim} not installed: {}",
AreevError::ReadOnly(
"this memory has no declared embedding provenance yet, and a read-only \
handle cannot create the vector column"
.into()
)
));
return;
}
if !self.read_only {
if let Err(err) = self.db.ensure_embeddings(dim) {
self.warnings.push(format!(
"vector recall disabled — embedder {model}@{dim} not installed: {err}"
));
return;
}
}
match &self.meta_embed {
Some((m, d)) => {
if *d != dim {
self.warnings.push(format!(
"embedding mismatch: file vectors are {m}@{d}, injected backend is \
{model}@{dim} — vector recall may be degraded"
));
} else if *m != model && m != "unspecified" && model != "unspecified" {
self.warnings.push(format!(
"embedding model differs: file declares {m}, injected {model} (same dim {dim})"
));
}
}
None => {
let ok = self
.db
.execute(
"INSERT OR REPLACE INTO meta(k, v) VALUES ('embedding_model', ?1)",
vec![pt(&model)],
)
.and_then(|_| {
self.db.execute(
"INSERT OR REPLACE INTO meta(k, v) VALUES ('embedding_dim', ?1)",
vec![pt(&dim.to_string())],
)
});
if ok.is_ok() {
self.meta_embed = Some((model, dim));
}
}
}
self.embedder = Some(e);
}
pub fn set_run_id(&mut self, run_id: Option<&str>) {
self.ambient_run_id = run_id.filter(|id| !id.is_empty()).map(str::to_string);
}
pub fn set_reranker(&mut self, r: Box<dyn RerankBackend>) {
self.reranker = Some(r);
}
pub fn has_reranker(&self) -> bool {
self.reranker.is_some()
}
pub fn set_query_expander(&mut self, e: Box<dyn QueryExpander>) {
self.expander = Some(e);
}
pub fn indexed_documents(&self) -> i64 {
match self.db.collection_stats() {
Ok(Some((docs, _))) => docs,
_ => self.fts_docs,
}
}
pub fn index_text_enabled(&self) -> bool {
self.index_text
}
pub fn meta_scan(&self, prefix: &str) -> Result<Vec<(String, String)>> {
let pattern = format!("{}%", like_escape(prefix));
let mut out = Vec::new();
for row in self.db.query(
"SELECT k, v FROM meta WHERE k LIKE ?1 ESCAPE '\\'",
vec![pt(&pattern)],
)? {
if let (Some(k), Some(v)) = (row.text(0), row.text(1)) {
if let Some(rest) = k.strip_prefix(prefix) {
out.push((rest.to_string(), v.to_string()));
}
}
}
Ok(out)
}
pub fn meta_get(&self, key: &str) -> Result<Option<String>> {
let rows = self.db.query("SELECT v FROM meta WHERE k = ?1", vec![pt(key)])?;
Ok(rows.first().and_then(|row| row.text(0)).map(str::to_string))
}
pub fn meta_put(&self, key: &str, value: &str) -> Result<()> {
self.check_writable("write a meta row")?;
self.db.execute(
"INSERT OR REPLACE INTO meta(k, v) VALUES (?1, ?2)",
vec![pt(key), pt(value)],
)?;
Ok(())
}
pub fn meta_delete(&self, key: &str) -> Result<()> {
self.check_writable("delete a meta row")?;
self.db.execute("DELETE FROM meta WHERE k = ?1", vec![pt(key)])?;
Ok(())
}
pub fn meta_cas(&self, key: &str, expected: Option<&str>, new: &str) -> Result<bool> {
self.check_writable("write a meta row")?;
let affected = match expected {
None => self.db.execute(
"INSERT OR IGNORE INTO meta(k, v) VALUES (?1, ?2)",
vec![pt(key), pt(new)],
)?,
Some(prev) => self.db.execute(
"UPDATE meta SET v = ?3 WHERE k = ?1 AND v = ?2",
vec![pt(key), pt(prev), pt(new)],
)?,
};
Ok(affected > 0)
}
pub fn trigger_state(&self, trigger: &str) -> Result<Option<(TriggerState, String)>> {
let key = format!("{TRG_PREFIX}{trigger}");
let Some(raw) = self.meta_get(&key)? else {
return Ok(None);
};
match serde_json::from_str::<TriggerState>(&raw) {
Ok(st) => Ok(Some((st, raw))),
Err(e) => Err(AreevError::Validation(format!(
"unreadable trigger state for {trigger}: {e}"
))),
}
}
pub fn put_trigger_state(
&self,
trigger: &str,
expected_raw: Option<&str>,
state: &TriggerState,
) -> Result<bool> {
let json = serde_json::to_string(state).map_err(db_err)?;
self.meta_cas(&format!("{TRG_PREFIX}{trigger}"), expected_raw, &json)
}
pub fn trigger_states(&self) -> Result<Vec<(String, TriggerState)>> {
let mut out = Vec::new();
for (suffix, raw) in self.meta_scan(TRG_PREFIX)? {
match serde_json::from_str::<TriggerState>(&raw) {
Ok(st) => out.push((suffix, st)),
Err(e) => {
return Err(AreevError::Validation(format!(
"unreadable trigger state for {suffix}: {e}"
)))
}
}
}
Ok(out)
}
pub fn clear_trigger_state(&self, trigger: &str) -> Result<()> {
self.meta_delete(&format!("{TRG_PREFIX}{trigger}"))
}
pub fn retention_policies(&self) -> Result<Vec<(String, RetentionPolicy)>> {
let mut out = Vec::new();
for (ns, raw) in self.meta_scan(RETENTION_PREFIX)? {
match serde_json::from_str::<RetentionPolicy>(&raw) {
Ok(p) => out.push((ns, p)),
Err(e) => {
return Err(AreevError::Validation(format!(
"unreadable retention policy for namespace '{ns}': {e}"
)))
}
}
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
pub fn set_anon_policy(&mut self, ns: &str, policy_json: &str) -> Result<()> {
if ns.trim().is_empty() {
return Err(AreevError::Validation(
"anonymization policy needs a namespace".into(),
));
}
require_exact_ns("an anonymization policy", ns)?;
let policy = areev_core::anon::AnonPolicy::from_json(policy_json)?;
let needs_memory_key = matches!(policy.mode.as_str(), "ingress" | "both")
|| policy.scope == "memory"
|| policy.vault;
if needs_memory_key && !self.anon.has_memory_key() {
return Err(AreevError::Validation(format!(
"anonymization mode \"{}\" / scope \"{}\" needs key material: \
value-derived pseudonym tokens and the mapping vault are keyed from \
either anon_key (a host-supplied 32-byte key from your KMS — works \
on every backend, including postgres: AreevOptions::anon_key, \
--anon-key-env VAR, or anon_key=/anonKey on the bindings) or, on \
the file backend, the page key (open with --passphrase-env / \
open_encrypted)",
policy.mode, policy.scope
)));
}
let json = serde_json::to_string(&policy).map_err(db_err)?;
self.meta_put(&format!("{}{ns}", anon_gate::ANON_PREFIX), &json)?;
let current = env!("CARGO_PKG_VERSION");
let needs_stamp = match self.meta_get(MIN_READER_VERSION_KEY)? {
Some(existing) => version_lt(&existing, current),
None => true,
};
if needs_stamp {
self.meta_put(MIN_READER_VERSION_KEY, current)?;
self.needs_min_reader_stamp = false;
}
self.reload_anon_policies()
}
pub fn clear_anon_policy(&mut self, ns: &str) -> Result<()> {
self.meta_delete(&format!("{}{ns}", anon_gate::ANON_PREFIX))?;
self.reload_anon_policies()
}
pub fn anon_policies(&self) -> Result<Vec<(String, areev_core::anon::AnonPolicy)>> {
let mut out = Vec::new();
for (ns, json) in self.meta_scan(anon_gate::ANON_PREFIX)? {
let p = areev_core::anon::AnonPolicy::from_json(&json).map_err(|e| {
AreevError::Validation(format!("anon:{ns} policy row is unreadable: {e}"))
})?;
out.push((ns, p));
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
pub fn anon_known_identities(&self, ns: &str) -> Vec<String> {
self.anon.known_for(ns)
}
fn reload_anon_policies(&mut self) -> Result<()> {
let rows = self.meta_scan(anon_gate::ANON_PREFIX)?;
self.anon.reload(&rows);
for (ns, policy) in self.anon.vault_namespaces() {
let _ = policy;
let Some(key) = self.anon.vault_key() else { continue };
let prefix = format!("{VAULT_PREFIX}{ns}:");
let mut entries = Vec::new();
for (token, row) in self.meta_scan(&prefix)? {
let v: serde_json::Value = serde_json::from_str(&row).map_err(|e| {
AreevError::Validation(format!("vault row {prefix}{token} unreadable: {e}"))
})?;
let sealed = hex::decode(v["v"].as_str().unwrap_or_default()).map_err(|e| {
AreevError::Validation(format!("vault row {prefix}{token} not hex: {e}"))
})?;
let row_key = format!("{prefix}{token}");
let value = blobcrypt::open(&key, &row_key, &sealed)?;
let value = String::from_utf8(value).map_err(|e| {
AreevError::Validation(format!("vault row {row_key} not UTF-8: {e}"))
})?;
entries.push((token, value));
}
self.anon.set_vault_seed(&ns, entries);
}
for (ns, _) in &rows {
if let Some(ns_id) = self.term_lookup(ns)? {
let ids: Vec<i64> = self
.db
.query(
"SELECT DISTINCT s FROM triples WHERE ns=?1 LIMIT 1024",
vec![pi(ns_id)],
)?
.iter()
.filter_map(|row| row.i64(0))
.collect();
let mut subjects = Vec::with_capacity(ids.len());
for id in ids {
if let Some(t) = self.term_str(id)? {
subjects.push(t);
}
}
self.anon.seed_known(ns, subjects);
}
}
Ok(())
}
pub fn ingress_transform_text(&mut self, ns: &str, text: &str) -> Result<Option<String>> {
let out = self.anon.ingress_text(ns, text)?;
if out.is_some() {
self.flush_anon_vault()?;
}
Ok(out)
}
pub fn ingress_transform_identity(&mut self, ns: &str, value: &str) -> Result<Option<String>> {
let out = self.anon.ingress_identity(ns, value)?;
if out.is_some() {
self.flush_anon_vault()?;
}
Ok(out)
}
pub fn ingress_active(&self, ns: &str) -> Result<bool> {
self.anon.ingress_active(ns)
}
pub fn set_anonymizer(&mut self, backend: Box<dyn areev_core::anon::DetectorBackend>) {
self.anon.install_backend(backend);
}
pub fn anon_backends(&self) -> Vec<(String, String)> {
self.anon.installed_backends()
}
pub fn anon_preview(&mut self, hash: &Hash) -> Result<DeserializedGrain> {
let mut g = self.get(hash)?;
self.anon.preview_grain(&mut g)?;
Ok(g)
}
fn flush_anon_vault(&mut self) -> Result<()> {
let pending = self.anon.take_vault_pending();
if pending.is_empty() {
return Ok(());
}
let Some(key) = self.anon.vault_key() else {
return Err(AreevError::CryptoError(
"vault-enabled policy on a handle with no vault key".into(),
));
};
let at = now_ms();
for (ns, token, value) in pending {
let row_key = format!("{VAULT_PREFIX}{ns}:{token}");
let sealed = blobcrypt::seal(&key, &row_key, value.as_bytes())?;
let row = serde_json::json!({"v": hex::encode(sealed), "at": at});
self.meta_put(&row_key, &row.to_string())?;
}
Ok(())
}
fn egress_exit(
&mut self,
ns_hint: Option<&str>,
grains: &mut [DeserializedGrain],
) -> Result<()> {
if self.anon.is_idle() {
return Ok(());
}
self.anon.egress_grains(ns_hint, grains)?;
self.flush_anon_vault()
}
fn egress_strings_exit(
&mut self,
ns: &str,
category: Option<&str>,
values: &mut [String],
) -> Result<()> {
if self.anon.is_idle() {
return Ok(());
}
self.anon.egress_strings(ns, category, values)?;
self.flush_anon_vault()
}
pub fn anon_egress_text(&mut self, ns: &str, values: &mut [String]) -> Result<()> {
self.egress_strings_exit(ns, None, values)
}
pub fn anon_mapping_for(&self, ns: &str) -> Result<std::collections::BTreeMap<String, String>> {
let mut out = std::collections::BTreeMap::new();
for (m_ns, _, mapping) in self.anon.mappings()? {
if m_ns == ns {
out.extend(mapping);
}
}
Ok(out)
}
pub fn anon_placeholder(&self, ns: &str) -> Result<Option<String>> {
self.anon.placeholder(ns)
}
pub fn anon_reveal(&mut self, ns: &str, token: &str) -> Result<Option<String>> {
for (m_ns, _, mapping) in self.anon.mappings()? {
if m_ns == ns {
if let Some(v) = mapping.get(token) {
return Ok(Some(v.clone()));
}
}
}
let row_key = format!("{VAULT_PREFIX}{ns}:{token}");
let Some(row) = self.meta_get(&row_key)? else { return Ok(None) };
let Some(key) = self.anon.vault_key() else {
return Err(AreevError::CryptoError(
"vault row present but this handle has no vault key".into(),
));
};
let v: serde_json::Value = serde_json::from_str(&row)
.map_err(|e| AreevError::Validation(format!("vault row {row_key}: {e}")))?;
let sealed = hex::decode(v["v"].as_str().unwrap_or_default())
.map_err(|e| AreevError::Validation(format!("vault row {row_key}: {e}")))?;
let value = blobcrypt::open(&key, &row_key, &sealed)?;
Ok(Some(String::from_utf8(value).map_err(|e| {
AreevError::Validation(format!("vault row {row_key} not UTF-8: {e}"))
})?))
}
fn scrub_anon_identities(&mut self, ns: &str, identities: &[String]) -> Result<usize> {
if self.anon.is_idle() {
return Ok(0);
}
let mut removed = self.anon.scrub_identities(identities);
if let Some(key) = self.anon.vault_key() {
let prefix = format!("{VAULT_PREFIX}{ns}:");
let mut doomed = Vec::new();
for (token, row) in self.meta_scan(&prefix)? {
let row_key = format!("{prefix}{token}");
let v: serde_json::Value = serde_json::from_str(&row)
.map_err(|e| AreevError::Validation(format!("vault row {row_key}: {e}")))?;
let sealed = hex::decode(v["v"].as_str().unwrap_or_default())
.map_err(|e| AreevError::Validation(format!("vault row {row_key}: {e}")))?;
let value = blobcrypt::open(&key, &row_key, &sealed)?;
if identities.iter().any(|i| i.as_bytes() == value.as_slice()) {
doomed.push(row_key);
}
}
for row_key in doomed {
self.meta_delete(&row_key)?;
removed += 1;
}
}
Ok(removed)
}
pub fn sweep_anon_vault(&mut self, now_ms: i64) -> Result<Vec<(String, usize)>> {
let mut out = Vec::new();
for (ns, policy) in self.anon.vault_namespaces() {
let Some(ttl) = policy.vault_ttl_days else { continue };
let cutoff = now_ms - (ttl * 86_400_000.0) as i64;
let prefix = format!("{VAULT_PREFIX}{ns}:");
let mut doomed = Vec::new();
for (token, row) in self.meta_scan(&prefix)? {
let at = serde_json::from_str::<serde_json::Value>(&row)
.ok()
.and_then(|v| v["at"].as_i64())
.unwrap_or(0);
if at < cutoff {
doomed.push(format!("{prefix}{token}"));
}
}
let n = doomed.len();
for row_key in doomed {
self.meta_delete(&row_key)?;
}
if n > 0 {
out.push((ns, n));
}
}
Ok(out)
}
pub fn set_anonymize_egress_floor(&mut self, on: bool) {
self.anon.set_floor(on);
}
pub fn anonymize_egress_floor(&self) -> bool {
self.anon.floor()
}
pub fn anon_active_mode(&self, ns: &str) -> Result<Option<String>> {
self.anon.active_mode(ns)
}
pub fn anon_declared(&self) -> Vec<(String, String)> {
self.anon.declared()
}
pub fn anon_mappings(&self) -> Result<Vec<AnonMappingRow>> {
self.anon.mappings()
}
pub fn anon_last_mapping_id(&self, ns: &str) -> Result<Option<String>> {
self.anon.last_mapping_id(ns)
}
pub fn anon_audit_counts(&self) -> Vec<(String, String, u64)> {
self.anon
.audit_counts()
.iter()
.map(|((ns, cat), n)| (ns.clone(), cat.clone(), *n))
.collect()
}
pub fn set_retention_policy(&self, ns: &str, policy: &RetentionPolicy) -> Result<()> {
if ns.trim().is_empty() {
return Err(AreevError::Validation(
"retention policy needs a namespace".into(),
));
}
require_exact_ns("a retention policy", ns)?;
if policy.days < 0.0 || !policy.days.is_finite() {
return Err(AreevError::Validation(format!(
"retention days must be a non-negative, finite number, got {}",
policy.days
)));
}
let json = serde_json::to_string(policy).map_err(db_err)?;
self.meta_put(&format!("{RETENTION_PREFIX}{ns}"), &json)
}
pub fn clear_retention_policy(&self, ns: &str) -> Result<()> {
self.meta_delete(&format!("{RETENTION_PREFIX}{ns}"))
}
pub fn set_retention_floor(&self, ns: &str, min_days: f64, because: &str) -> Result<()> {
if ns.trim().is_empty() {
return Err(AreevError::Validation("retention floor needs a namespace".into()));
}
require_exact_ns("a retention floor", ns)?;
if min_days <= 0.0 || !min_days.is_finite() {
return Err(AreevError::Validation(format!(
"retention floor min_days must be a positive, finite number, got {min_days}"
)));
}
if because.trim().is_empty() {
return Err(AreevError::Validation(
"retention floor requires a non-empty because".into(),
));
}
let json = serde_json::json!({"min_days": min_days, "because": because}).to_string();
self.meta_put(&format!("{RETENTION_FLOOR_PREFIX}{ns}"), &json)
}
pub fn clear_retention_floor(&self, ns: &str) -> Result<()> {
self.meta_delete(&format!("{RETENTION_FLOOR_PREFIX}{ns}"))
}
pub fn retention_floors(&self) -> Result<Vec<(String, f64, String)>> {
let mut out = Vec::new();
for (ns, raw) in self.meta_scan(RETENTION_FLOOR_PREFIX)? {
let v: serde_json::Value = serde_json::from_str(&raw).map_err(|e| {
AreevError::Validation(format!("unreadable retention floor for '{ns}': {e}"))
})?;
out.push((
ns,
v.get("min_days").and_then(|d| d.as_f64()).unwrap_or(0.0),
v.get("because").and_then(|b| b.as_str()).unwrap_or("").to_string(),
));
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
pub fn place_hold(&self, ns: &str, because: &str, placed_by: &str, at_ms: i64) -> Result<()> {
if ns.trim().is_empty() {
return Err(AreevError::Validation("hold needs a namespace".into()));
}
require_exact_ns("a legal hold", ns)?;
if because.trim().is_empty() || placed_by.trim().is_empty() {
return Err(AreevError::Validation(
"hold requires a non-empty because and placed_by".into(),
));
}
let json = serde_json::json!({
"because": because, "placed_by": placed_by, "at_ms": at_ms
})
.to_string();
self.meta_put(&format!("{HOLD_PREFIX}{ns}"), &json)
}
pub fn release_hold(&self, ns: &str) -> Result<()> {
self.meta_delete(&format!("{HOLD_PREFIX}{ns}"))
}
pub fn holds(&self) -> Result<Vec<(String, String, String)>> {
let mut out = Vec::new();
for (ns, raw) in self.meta_scan(HOLD_PREFIX)? {
let v: serde_json::Value = serde_json::from_str(&raw).map_err(|e| {
AreevError::Validation(format!("unreadable hold for '{ns}': {e}"))
})?;
out.push((
ns,
v.get("because").and_then(|b| b.as_str()).unwrap_or("").to_string(),
v.get("placed_by").and_then(|b| b.as_str()).unwrap_or("").to_string(),
));
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
fn check_age_destruction_allowed(&self, ns: Option<&str>, cutoff_ms: i64) -> Result<()> {
let holds = self.holds()?;
match ns {
Some(n) => {
if let Some((_, because, placed_by)) = holds.iter().find(|(h, _, _)| h == n) {
return Err(AreevError::Validation(format!(
"namespace '{n}' is under a legal hold placed by {placed_by} \
({because}) — age-based destruction refuses until the hold \
is released"
)));
}
}
None => {
if let Some((h, because, placed_by)) = holds.first() {
return Err(AreevError::Validation(format!(
"a global sweep is refused while any legal hold is live \
(namespace '{h}', placed by {placed_by}: {because}) — \
scope the sweep to unheld namespaces"
)));
}
}
}
let now = now_ms();
for (fns, min_days, because) in self.retention_floors()? {
let applies = ns.is_none_or(|n| n == fns);
if !applies {
continue;
}
let floor_cutoff = now - (min_days * 86_400_000.0) as i64;
if cutoff_ms > floor_cutoff {
return Err(AreevError::Validation(format!(
"cutoff is younger than the {min_days}-day retention floor on \
'{fns}' ({because}) — destruction refused"
)));
}
}
Ok(())
}
pub fn sweep_retention(&mut self, now_ms: i64) -> Result<Vec<RetentionSweep>> {
self.sweep_anon_vault(now_ms)?;
let policies = self.retention_policies()?;
let mut out = Vec::new();
for (ns, policy) in policies {
let cutoff = now_ms - (policy.days * 86_400_000.0) as i64;
let gtype = match policy.grain_type.as_deref() {
None => None,
Some(t) => Some(areev_core::types::GrainType::from_str(t).ok_or_else(|| {
AreevError::Validation(format!(
"retention policy for '{ns}' names an unknown grain type: {t:?}"
))
})?),
};
if let Err(e) = self.check_age_destruction_allowed(Some(&ns), cutoff) {
out.push(RetentionSweep {
ns,
days: policy.days,
outcome: RetentionOutcome::Skipped { why: e.to_string() },
});
continue;
}
let report = self.forget_older_than(Some(&ns), cutoff, gtype)?;
out.push(RetentionSweep {
ns,
days: policy.days,
outcome: RetentionOutcome::Swept(report),
});
}
Ok(out)
}
pub fn defer_text_index(&mut self) -> Result<bool> {
if !self.index_text || self.fts_deferred {
return Ok(false);
}
self.fts_deferred = true;
Ok(true)
}
pub fn rebuild_text_index(&mut self) -> Result<usize> {
self.check_writable("reindex the text index")?;
if !self.index_text {
return Err(AreevError::Validation(
"text indexing is off for this file — reopen it with text indexing on \
(open_with index_text, index_text=True from Python, indexText from \
Node, --index-text true on the CLI) before rebuilding the FTS index"
.to_string(),
));
}
for row in self.db.query("SELECT text, blob FROM grains", vec![])? {
let text = match row.text(0) {
Some(t) => Some(t.to_string()),
None => match row.blob(1) {
Some(b) => projected_text(&deserialize_blob(&b)?),
None => None,
},
};
if let Some(t) = text {
for (term, _) in token_freqs(&t).0 {
self.fts_term_id(&term)?;
}
}
}
self.fts_deferred = false;
self.db.begin()?;
let r = (|| -> Result<(usize, i64, i64)> {
self.db.reserve_write(0, 0, 0, now_ms() << 16, 0)?;
let mut updates: Vec<(i64, String)> = Vec::new();
for row in self.db.query("SELECT seq, blob FROM grains WHERE text IS NULL", vec![])? {
let seq = row.i64(0).unwrap_or(0);
if let Some(b) = row.blob(1) {
let view = deserialize_blob(&b)?;
if let Some(t) = projected_text(&view) {
updates.push((seq, t));
}
}
}
for (seq, t) in &updates {
self.db
.execute("UPDATE grains SET text = ?1 WHERE seq = ?2", vec![pt(t), pi(*seq)])?;
}
self.db.execute("DELETE FROM fts_post", vec![])?;
self.db.execute("DELETE FROM fts_doc", vec![])?;
let mut texts: Vec<(i64, i64, String)> = Vec::new();
for row in self
.db
.query("SELECT seq, ns, text FROM grains WHERE text IS NOT NULL", vec![])?
{
let seq = row.i64(0).unwrap_or(0);
let ns = row.i64(1).unwrap_or(0);
if let Some(t) = row.text(2) {
texts.push((seq, ns, t.to_string()));
}
}
let (mut docs, mut total_len) = (0i64, 0i64);
for (seq, ns, text) in &texts {
let (freqs, len) = token_freqs(text);
if freqs.is_empty() {
continue;
}
let mut wrote = false;
for (term, tf) in freqs {
let Some(id) = self.fts_term_lookup(&term)? else {
continue;
};
self.db.execute_hot(
"INSERT INTO fts_post(term,seq,ns,tf) VALUES (?1,?2,?3,?4)",
vec![pi(id), pi(*seq), pi(*ns), pi(tf)],
)?;
wrote = true;
}
if wrote {
self.db.execute_hot(
"INSERT OR REPLACE INTO fts_doc(seq,len) VALUES (?1,?2)",
vec![pi(*seq), pi(len)],
)?;
docs += 1;
total_len += len;
}
}
Ok((updates.len(), docs, total_len))
})();
match r {
Ok((backfilled, docs, total_len)) => {
self.db.commit()?;
self.fts_docs = docs;
self.fts_total_len = total_len;
Ok(backfilled)
}
Err(e) => {
let _ = self.db.rollback();
Err(e)
}
}
}
fn fts_term_lookup(&mut self, term: &str) -> Result<Option<i64>> {
if let Some(id) = self.fts_vocab.get(term) {
return Ok(Some(*id));
}
let rows = self.db.query("SELECT id FROM fts_vocab WHERE term = ?1", vec![pt(term)])?;
match rows.first().and_then(|r| r.i64(0)) {
Some(id) => {
self.fts_vocab.insert(term.to_string(), id);
Ok(Some(id))
}
None => Ok(None),
}
}
pub fn embedder_dim(&self) -> Option<usize> {
self.embedder.as_ref().map(|e| e.dim())
}
pub fn declared_embedding(&self) -> Option<(&str, usize)> {
self.meta_embed.as_ref().map(|(m, d)| (m.as_str(), *d))
}
pub fn open_warnings(&self) -> &[String] {
&self.warnings
}
fn next_hlc(&mut self) -> i64 {
let wall = now_ms() << 16;
self.hlc_last = if wall > self.hlc_last { wall } else { self.hlc_last + 1 };
self.hlc_last
}
fn term_id(&mut self, term: &str) -> Result<i64> {
if let Some(id) = self.dict.get(term) {
return Ok(*id);
}
if let Some(id) = self.db.intern_term(term)? {
self.dict.insert(term.to_string(), id);
return Ok(id);
}
let id = self.next_term;
self.next_term += 1;
self.db
.execute("INSERT INTO terms(id, term) VALUES (?1, ?2)", vec![pi(id), pt(term)])?;
self.dict.insert(term.to_string(), id);
Ok(id)
}
fn term_lookup(&mut self, term: &str) -> Result<Option<i64>> {
if let Some(id) = self.dict.get(term) {
return Ok(Some(*id));
}
if let Some(id) = self.db.lookup_term(term)? {
self.dict.insert(term.to_string(), id);
return Ok(Some(id));
}
Ok(None)
}
fn term_str(&self, id: i64) -> Result<Option<String>> {
if let Some(t) = self.dict.iter().find(|(_, v)| **v == id).map(|(k, _)| k.clone()) {
return Ok(Some(t));
}
self.db.lookup_term_str(id)
}
pub fn add<G: Grain + 'static>(&mut self, grain: &G) -> Result<Hash> {
self.add_batch_inner(std::slice::from_ref(&(grain as &dyn AddableDyn)))
.map(|mut v| v.remove(0))
}
pub fn add_batch(&mut self, grains: &[&dyn AddableDyn]) -> Result<Vec<Hash>> {
self.add_batch_inner(grains)
}
pub fn add_with_embedding<G: Grain + 'static>(
&mut self,
grain: &G,
embedding: &[f32],
) -> Result<Hash> {
self.check_embedding_dim(embedding.len())?;
let hash = self.add(grain)?;
self.set_grain_embedding(&hash, embedding)?;
Ok(hash)
}
pub fn set_grain_embedding(&mut self, hash: &Hash, embedding: &[f32]) -> Result<Hash> {
match self.set_grain_embeddings(&[(*hash, embedding.to_vec())])? {
0 => Err(AreevError::NotFound(*hash)),
_ => Ok(*hash),
}
}
pub fn set_grain_embeddings(&mut self, items: &[(Hash, Vec<f32>)]) -> Result<usize> {
self.check_writable("write embeddings")?;
let Some((_, first)) = items.first() else {
return Ok(0);
};
let dim = first.len();
self.check_embedding_dim(dim)?;
for (h, v) in items {
if v.len() != dim {
return Err(AreevError::Validation(format!(
"embedding for {} has {} dimensions; this batch is {dim}-dimensional",
h.to_hex(),
v.len()
)));
}
}
let mut by_hash: HashMap<Vec<u8>, i64> = HashMap::new();
for chunk in items.chunks(EMBEDDING_BATCH_CHUNK) {
let placeholders: Vec<String> = (1..=chunk.len()).map(|i| format!("?{i}")).collect();
let params: Vec<Value> = chunk.iter().map(|(h, _)| pb(h.as_bytes().to_vec())).collect();
let rows = self.db.query(
&format!("SELECT seq, hash FROM grains WHERE hash IN ({})", placeholders.join(",")),
params,
)?;
for r in &rows {
if let (Some(seq), Some(hash)) = (r.i64(0), r.blob(1)) {
by_hash.insert(hash, seq);
}
}
}
let mut rows: Vec<(i64, String)> = Vec::with_capacity(items.len());
let mut seen: HashMap<i64, usize> = HashMap::new();
for (h, v) in items {
let Some(seq) = by_hash.get(h.as_bytes().as_slice()) else {
return Err(AreevError::NotFound(*h));
};
match seen.get(seq) {
Some(&at) => rows[at].1 = vec_to_json(v),
None => {
seen.insert(*seq, rows.len());
rows.push((*seq, vec_to_json(v)));
}
}
}
self.db.ensure_embeddings(dim)?;
let dbr = self.db.as_ref();
let written = with_txn(dbr, || {
let mut written = 0usize;
for chunk in rows.chunks(EMBEDDING_BATCH_CHUNK) {
let seqs: Vec<i64> = chunk.iter().map(|(seq, _)| *seq).collect();
let alive: HashSet<i64> = dbr
.query(
&format!(
"SELECT seq FROM grains WHERE seq IN ({}){}",
seq_csv(&seqs),
dbr.for_update()
),
vec![],
)?
.iter()
.filter_map(|r| r.i64(0))
.collect();
let live: Vec<&(i64, String)> =
chunk.iter().filter(|(seq, _)| alive.contains(seq)).collect();
if live.is_empty() {
continue;
}
let live_seqs: Vec<i64> = live.iter().map(|(seq, _)| *seq).collect();
dbr.execute(
&format!("DELETE FROM embeddings WHERE seq IN ({})", seq_csv(&live_seqs)),
vec![],
)?;
let mut values: Vec<String> = Vec::with_capacity(live.len());
let mut params: Vec<Value> = Vec::with_capacity(live.len() * 2);
for (i, (seq, qjson)) in live.iter().enumerate() {
values.push(format!("(?{}, vector32(?{}))", 2 * i + 1, 2 * i + 2));
params.push(pi(*seq));
params.push(pt(qjson));
}
dbr.execute(
&format!("INSERT INTO embeddings(seq, vec) VALUES {}", values.join(", ")),
params,
)?;
written += live.len();
}
Ok(written)
})?;
if written > 0 && self.meta_embed.is_none() {
self.meta_put("embedding_model", "external")?;
self.meta_put("embedding_dim", &dim.to_string())?;
self.meta_embed = Some(("external".to_string(), dim));
}
Ok(written)
}
pub fn vector_recall_check(
&mut self,
ns: &str,
queries: &[Vec<f32>],
k: usize,
) -> Result<VectorRecallReport> {
if k == 0 {
return Err(AreevError::Validation("recall@k needs k >= 1".into()));
}
if queries.is_empty() {
return Err(AreevError::Validation(
"recall@k needs at least one query vector".into(),
));
}
for q in queries {
self.check_embedding_dim(q.len())?;
}
let ef_search = self.db.ann_ef_search()?;
let Some(index) = self.vector_index()? else {
return Ok(VectorRecallReport {
index: None,
ef_search,
k,
queries: 0,
recall: Some(1.0),
});
};
let mut approx = Vec::with_capacity(queries.len());
for q in queries {
approx.push(self.nearest_vector(ns, None, None, q, k)?);
}
self.db.set_exact_vector_scan(true)?;
let exact: Result<Vec<Vec<(Hash, f32)>>> =
queries.iter().map(|q| self.nearest_vector(ns, None, None, q, k)).collect();
self.db.set_exact_vector_scan(false)?;
let exact = exact?;
let (mut hits, mut possible) = (0usize, 0usize);
for (a, e) in approx.iter().zip(&exact) {
possible += e.len();
let truth: HashSet<&Hash> = e.iter().map(|(h, _)| h).collect();
hits += a.iter().filter(|(h, _)| truth.contains(h)).count();
}
let recall = (possible > 0).then(|| hits as f32 / possible as f32);
Ok(VectorRecallReport { index: Some(index), ef_search, k, queries: queries.len(), recall })
}
pub fn set_vector_ef_search(&mut self, ef_search: usize) -> Result<()> {
self.db.set_ann_ef_search(ef_search)
}
pub fn ensure_vector_index(
&mut self,
m: usize,
ef_construction: usize,
ef_search: usize,
) -> Result<()> {
self.check_writable("build a vector index")?;
self.db.ensure_ann_index(m, ef_construction, ef_search)
}
pub fn drop_vector_index(&mut self) -> Result<()> {
self.check_writable("drop a vector index")?;
self.db.drop_ann_index()
}
pub fn vector_index(&mut self) -> Result<Option<String>> {
self.db.ann_index_name()
}
pub fn add_if_novel<G: Grain + 'static>(&mut self, grain: &G) -> Result<(Hash, bool)> {
self.add_dyn_if_novel(grain as &dyn AddableDyn)
}
fn add_dyn_if_novel(&mut self, grain: &dyn AddableDyn) -> Result<(Hash, bool)> {
self.check_writable("add a grain")?;
let (blob, _hash) = grain.serialize_dyn()?;
let gv = extract_view(&deserialize_blob(&blob)?);
if let (Some(sj), Some(rl), Some(ob)) =
(gv.subject.as_deref(), gv.relation.as_deref(), gv.object.as_deref())
{
if let (Some(ns_id), Some(s_id), Some(p_id), Some(o_id)) = (
self.term_lookup(&gv.ns)?,
self.term_lookup(sj)?,
self.term_lookup(rl)?,
self.term_lookup(ob)?,
) {
if let Some(existing) = self.head_hash_for_object(ns_id, s_id, p_id, o_id)? {
return Ok((existing, false));
}
}
}
let (preps, first_seq, first_op, hlc0) =
self.prep_and_reserve(std::slice::from_ref(&grain))?;
if self.needs_min_reader_stamp
&& preps
.iter()
.any(|p| p.blob.get(2).is_some_and(|b| *b > LEGACY_MAX_GRAIN_BYTE))
{
self.meta_put(MIN_READER_VERSION_KEY, env!("CARGO_PKG_VERSION"))?;
self.needs_min_reader_stamp = false;
}
let key = match (preps[0].s, preps[0].p, preps[0].o) {
(Some(s), Some(p), Some(o)) => Some((preps[0].ns_id, s, p, o)),
_ => None,
};
let hash = preps[0].hash;
let (d_docs, d_len) = fts_delta(&preps);
let dbr = self.db.as_ref();
let duplicate = with_txn(dbr, || {
dbr.reserve_write(0, 0, 0, now_ms() << 16, 0)?;
if let Some((ns, s, p, o)) = key {
let rows = dbr.query(
"SELECT hash FROM entity_latest WHERE ns=?1 AND s=?2 AND p=?3 AND o=?4",
vec![pi(ns), pi(s), pi(p), pi(o)],
)?;
if let Some(b) = rows.first().and_then(|r| r.blob(0)) {
return Ok(Some(Hash::try_from_bytes(&b)?));
}
}
let (s0, o0, h0) = match dbr.reserve_write(1, 1, 1, now_ms() << 16, 0)? {
Some(w) => (w.seq0, w.op0, w.hlc0),
None => (first_seq, first_op, hlc0),
};
insert_prepped(dbr, &preps, s0, o0, h0)?;
Ok(None)
})?;
match duplicate {
Some(existing) => Ok((existing, false)),
None => {
self.fts_docs += d_docs;
self.fts_total_len += d_len;
Ok((hash, true))
}
}
}
fn head_hash_for_object(&mut self, ns: i64, s: i64, p: i64, o: i64) -> Result<Option<Hash>> {
let rows = self.db.query(
"SELECT hash FROM entity_latest WHERE ns=?1 AND s=?2 AND p=?3 AND o=?4",
vec![pi(ns), pi(s), pi(p), pi(o)],
)?;
match rows.first().and_then(|row| row.blob(0)) {
Some(b) => Ok(Some(Hash::try_from_bytes(&b)?)),
None => Ok(None),
}
}
fn prep_from_blob(&mut self, blob: Vec<u8>, hash: Hash, new_write: bool) -> Result<GrainPrep> {
let view = deserialize_blob(&blob)?;
let gv = extract_view(&view);
if new_write {
require_writable_ns(&gv.ns)?;
}
if !self.anon.is_idle() {
if let Some(subject) = &gv.subject {
self.anon.note_written_identity(&gv.ns, subject);
}
}
let ns_id = self.term_id(&gv.ns)?;
let (mut s, mut p, mut o, mut osp) = (None, None, None, false);
if let (Some(sj), Some(rl), Some(ob)) = (&gv.subject, &gv.relation, &gv.object) {
s = Some(self.term_id(sj)?);
p = Some(self.term_id(rl)?);
o = Some(self.term_id(ob)?);
osp = rl == "mg:harness" || self.entity_rels.contains(rl.as_str());
}
let subject_only = match (&gv.subject, s) {
(Some(sj), None) => Some(self.term_id(sj)?),
_ => None,
};
let session = match &gv.session {
Some(x) => Some(self.term_id(x)?),
None => None,
};
let projected = projected_text(&view);
let text = if self.index_text { projected.clone() } else { None };
let (tokens, doc_len) = match (&text, self.fts_deferred) {
(Some(t), false) => {
let (freqs, len) = token_freqs(t);
let mut ids = Vec::with_capacity(freqs.len());
for (term, tf) in freqs {
ids.push((self.fts_term_id(&term)?, tf));
}
(ids, len)
}
_ => (Vec::new(), 0),
};
let embed_text = projected;
let embedding = match (&self.embedder, &embed_text) {
(Some(e), Some(t)) => Some(e.embed(t)?),
_ => None,
};
let mut links = Vec::with_capacity(gv.links.len());
if !gv.links.is_empty() {
let self_id = self.term_id(&hash.to_hex())?;
for (rel, target) in &gv.links {
links.push((self_id, self.term_id(rel)?, self.term_id(target)?));
}
}
let run = match &gv.run {
Some(r) => Some(self.term_id(r)?),
None => None,
};
let parent = gv
.parent
.as_deref()
.and_then(|p| Hash::from_hex(p).ok())
.map(|h| h.as_bytes().to_vec());
Ok(GrainPrep {
blob,
hash,
ns_id,
s,
p,
o,
osp,
session,
vf: gv.vf,
vt: gv.vt,
created: gv.created_at,
gtype: gv.gtype as i64,
text,
embedding,
tokens,
doc_len,
links,
subject_only,
run,
parent,
corpus_export: gv.corpus_export,
})
}
fn fts_term_id(&mut self, term: &str) -> Result<i64> {
if let Some(id) = self.fts_vocab.get(term) {
return Ok(*id);
}
self.db
.execute("INSERT OR IGNORE INTO fts_vocab(term) VALUES (?1)", vec![pt(term)])?;
let rows = self
.db
.query("SELECT id FROM fts_vocab WHERE term = ?1", vec![pt(term)])?;
let id = match rows.first().and_then(|row| row.i64(0)) {
Some(id) => id,
None => return Err(AreevError::Storage("fts_vocab insert vanished".into())),
};
self.fts_vocab.insert(term.to_string(), id);
Ok(id)
}
fn add_batch_inner(&mut self, grains: &[&dyn AddableDyn]) -> Result<Vec<Hash>> {
self.check_writable("add a grain")?;
let mut prepped: Vec<Option<GrainPrep>> = Vec::with_capacity(grains.len());
let mut hashes: Vec<Hash> = Vec::with_capacity(grains.len());
let mut seen_in_batch: HashSet<Hash> = HashSet::new();
for g in grains {
let (blob, hash) = g.serialize_dyn()?;
hashes.push(hash);
if !seen_in_batch.insert(hash) || self.has(&hash)? {
prepped.push(None);
continue;
}
prepped.push(Some(self.prep_from_blob(blob, hash, true)?));
}
let preps: Vec<GrainPrep> = prepped.into_iter().flatten().collect();
if preps.is_empty() {
return Ok(hashes);
}
let (preps, first_seq, first_op, hlc0) = self.reserve_for(preps);
if self.needs_min_reader_stamp
&& preps
.iter()
.any(|p| p.blob.get(2).is_some_and(|b| *b > LEGACY_MAX_GRAIN_BYTE))
{
self.meta_put(MIN_READER_VERSION_KEY, env!("CARGO_PKG_VERSION"))?;
self.needs_min_reader_stamp = false;
}
let (d_docs, d_len) = fts_delta(&preps);
let n = preps.len() as i64;
let dbr = self.db.as_ref();
with_txn(dbr, || {
let (s0, o0, h0) = match dbr.reserve_write(n, n, n, now_ms() << 16, 0)? {
Some(w) => (w.seq0, w.op0, w.hlc0),
None => (first_seq, first_op, hlc0),
};
insert_prepped(dbr, &preps, s0, o0, h0)
})?;
self.fts_docs += d_docs;
self.fts_total_len += d_len;
Ok(hashes)
}
fn prep_and_reserve(
&mut self,
grains: &[&dyn AddableDyn],
) -> Result<(Vec<GrainPrep>, i64, i64, i64)> {
let mut preps = Vec::with_capacity(grains.len());
for g in grains {
let (blob, hash) = g.serialize_dyn()?;
preps.push(self.prep_from_blob(blob, hash, true)?);
}
Ok(self.reserve_for(preps))
}
fn reserve_for(&mut self, preps: Vec<GrainPrep>) -> (Vec<GrainPrep>, i64, i64, i64) {
let first_seq = self.next_seq;
self.next_seq += preps.len() as i64;
let first_op = self.next_op;
self.next_op += preps.len() as i64;
let hlc0 = self.next_hlc();
self.hlc_last = hlc0 + preps.len() as i64 - 1;
(preps, first_seq, first_op, hlc0)
}
pub fn rebuild_link_indexes(&mut self) -> Result<usize> {
self.check_writable("reindex the link indexes")?;
for row in self.db.query("SELECT blob FROM grains ORDER BY seq", vec![])? {
let Some(blob) = row.blob(0) else { continue };
let view = deserialize_blob(&blob)?;
let gv = extract_view(&view);
if let Some(r) = &gv.run {
self.term_id(r)?;
}
if !gv.links.is_empty() {
self.term_id(&view.hash.to_hex())?;
for (rel, target) in &gv.links {
self.term_id(rel)?;
self.term_id(target)?;
}
}
if let (Some(sj), true) = (&gv.subject, gv.relation.is_none() || gv.object.is_none()) {
self.term_id(sj)?;
}
}
struct Row {
seq: i64,
ns: i64,
run: Option<i64>,
parent: Option<Vec<u8>>,
corpus_export: bool,
links: Vec<(i64, i64, i64)>,
subject_only: Option<i64>,
superseded: bool,
}
self.db.begin()?;
let r = (|| -> Result<usize> {
self.db.reserve_write(0, 0, 0, now_ms() << 16, 0)?;
self.db.execute("DELETE FROM prov_idx", vec![])?;
self.db.execute("DELETE FROM run_idx", vec![])?;
self.db.execute("DELETE FROM corpus_idx", vec![])?;
let mut rows: Vec<(i64, i64, Vec<u8>, bool)> = Vec::new();
for row in self
.db
.query("SELECT seq, ns, blob, svt FROM grains ORDER BY seq", vec![])?
{
let (Some(seq), Some(ns), Some(blob)) = (row.i64(0), row.i64(1), row.blob(2))
else {
continue;
};
rows.push((seq, ns, blob, row.i64(3).is_some()));
}
let mut plan: Vec<Row> = Vec::with_capacity(rows.len());
for (seq, ns, blob, superseded) in &rows {
let view = deserialize_blob(blob)?;
let gv = extract_view(&view);
let run = match &gv.run {
Some(r) => self.term_lookup(r)?,
None => None,
};
let parent = gv
.parent
.as_deref()
.and_then(|p| Hash::from_hex(p).ok())
.map(|h| h.as_bytes().to_vec());
let mut links = Vec::with_capacity(gv.links.len());
if !gv.links.is_empty() {
if let Some(self_id) = self.term_lookup(&view.hash.to_hex())? {
for (rel, target) in &gv.links {
if let (Some(rl), Some(tg)) =
(self.term_lookup(rel)?, self.term_lookup(target)?)
{
links.push((self_id, rl, tg));
}
}
}
}
let subject_only = match &gv.subject {
Some(sj) if gv.relation.is_none() || gv.object.is_none() => {
self.term_lookup(sj)?
}
_ => None,
};
plan.push(Row {
seq: *seq,
ns: *ns,
run,
parent,
corpus_export: gv.corpus_export,
links,
subject_only,
superseded: *superseded,
});
}
let mut n = 0usize;
for r in &plan {
if let Some(run) = r.run {
self.db.execute(
"INSERT INTO run_idx(ns,run,seq) VALUES (?1,?2,?3)",
vec![pi(r.ns), pi(run), pi(r.seq)],
)?;
n += 1;
}
if let Some(ref p) = r.parent {
self.db.execute(
"INSERT INTO prov_idx(ns,parent,seq) VALUES (?1,?2,?3)",
vec![pi(r.ns), pb(p.clone()), pi(r.seq)],
)?;
n += 1;
}
if r.corpus_export {
self.db.execute(
"INSERT INTO corpus_idx(seq) VALUES (?1)",
vec![pi(r.seq)],
)?;
n += 1;
}
if let Some(s) = r.subject_only {
self.db.execute(
"DELETE FROM triples WHERE ns=?1 AND s=?2 AND p IS NULL AND o IS NULL AND seq=?3",
vec![pi(r.ns), pi(s), pi(r.seq)],
)?;
self.db.execute(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,NULL,NULL,?3,?4)",
vec![
pi(r.ns),
pi(s),
pi(r.seq),
pi(if r.superseded { 0 } else { 1 }),
],
)?;
n += 1;
}
for (ls, lp, lo) in &r.links {
self.db.execute(
"DELETE FROM triples WHERE ns=?1 AND s=?2 AND p=?3 AND o=?4 AND seq=?5",
vec![pi(r.ns), pi(*ls), pi(*lp), pi(*lo), pi(r.seq)],
)?;
self.db.execute(
"DELETE FROM osp WHERE ns=?1 AND o=?2 AND s=?3 AND p=?4 AND seq=?5",
vec![pi(r.ns), pi(*lo), pi(*ls), pi(*lp), pi(r.seq)],
)?;
self.db.execute(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(r.ns), pi(*ls), pi(*lp), pi(*lo), pi(r.seq)],
)?;
self.db.execute(
"INSERT INTO osp(ns,o,s,p,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(r.ns), pi(*lo), pi(*ls), pi(*lp), pi(r.seq)],
)?;
n += 1;
}
}
Ok(n)
})();
let written = match r {
Ok(n) => {
self.db.commit()?;
n
}
Err(e) => {
let _ = self.db.rollback();
return Err(e);
}
};
self.meta_put(LINK_INDEX_KEY, LINK_INDEX_VERSION)?;
Ok(written)
}
pub fn grains_derived_from(&mut self, parent: &Hash) -> Result<Vec<DeserializedGrain>> {
let key = parent.as_bytes().to_vec();
let mut out: Vec<DeserializedGrain> = self
.db
.query(
"SELECT g.blob FROM prov_idx p JOIN grains g ON g.seq = p.seq
WHERE p.parent = ?1 ORDER BY p.seq DESC",
vec![pb(key)],
)?
.iter()
.filter_map(|row| row.blob(0))
.map(|b| deserialize_blob(&b))
.collect::<Result<_>>()?;
self.egress_exit(None, &mut out)?;
Ok(out)
}
pub fn run_trace(&mut self, ns: &str, run_id: &str, limit: usize) -> Result<Vec<DeserializedGrain>> {
require_exact_ns("run_trace", ns)?;
let (Some(ns_id), Some(run)) = (self.term_lookup(ns)?, self.term_lookup(run_id)?) else {
return Ok(Vec::new());
};
let limit = limit.min(1024) as i64;
let mut out: Vec<DeserializedGrain> = self
.db
.query(
"SELECT g.blob FROM run_idx r JOIN grains g ON g.seq = r.seq
WHERE r.ns = ?1 AND r.run = ?2 ORDER BY r.seq DESC LIMIT ?3",
vec![pi(ns_id), pi(run), pi(limit)],
)?
.iter()
.filter_map(|row| row.blob(0))
.map(|b| deserialize_blob(&b))
.collect::<Result<_>>()?;
self.egress_exit(Some(ns), &mut out)?;
Ok(out)
}
pub fn run_grains(
&mut self,
ns: &str,
run_id: &str,
after_seq: i64,
limit: usize,
) -> Result<Vec<(i64, DeserializedGrain)>> {
require_exact_ns("run_grains", ns)?;
let (Some(ns_id), Some(run)) = (self.term_lookup(ns)?, self.term_lookup(run_id)?) else {
return Ok(Vec::new());
};
let limit = limit.clamp(1, 1024) as i64;
self.db
.query(
"SELECT r.seq, g.blob FROM run_idx r JOIN grains g ON g.seq = r.seq
WHERE r.ns = ?1 AND r.run = ?2 AND r.seq > ?3
ORDER BY r.seq ASC LIMIT ?4",
vec![pi(ns_id), pi(run), pi(after_seq), pi(limit)],
)?
.iter()
.filter_map(|row| Some((row.i64(0)?, row.blob(1)?)))
.map(|(seq, b)| Ok((seq, deserialize_blob(&b)?)))
.collect()
}
pub fn run_yield(&mut self, ns: &str, run_id: &str, limit: usize) -> Result<Vec<DeserializedGrain>> {
require_exact_ns("run_yield", ns)?;
let trace = self.run_trace(ns, run_id, limit)?;
let in_run: HashSet<Hash> = trace.iter().map(|g| g.hash).collect();
let mut seen: HashSet<Hash> = in_run.clone();
let mut out = Vec::new();
for g in &trace {
for child in self.grains_derived_from(&g.hash)? {
if !in_run.contains(&child.hash) && seen.insert(child.hash) {
out.push(child);
if out.len() >= limit {
return Ok(out);
}
}
}
}
Ok(out)
}
pub fn runs_touching(&mut self, ns: &str, hash: &Hash, depth: usize) -> Result<Vec<String>> {
require_exact_ns("runs_touching", ns)?;
let depth = depth.min(8);
let mut runs: Vec<String> = Vec::new();
let mut seen_run: HashSet<String> = HashSet::new();
let mut visited: HashSet<Hash> = HashSet::new();
let mut frontier = vec![*hash];
visited.insert(*hash);
for _ in 0..=depth {
let mut next: Vec<Hash> = Vec::new();
for h in std::mem::take(&mut frontier) {
let Ok(g) = self.get(&h) else { continue };
if let Some(r) = g.get_str("run_id") {
if seen_run.insert(r.to_string()) {
runs.push(r.to_string());
}
}
if let Some(p) = g.get_str("derived_from").and_then(|p| Hash::from_hex(p).ok()) {
if !visited.contains(&p)
&& self
.get(&p)
.is_ok_and(|pg| pg.get_str("namespace") == Some(ns))
{
visited.insert(p);
next.push(p);
}
}
for child in self.grains_derived_from(&h)? {
if child.get_str("namespace") == Some(ns) && visited.insert(child.hash) {
next.push(child.hash);
}
}
}
if next.is_empty() {
break;
}
frontier = next;
}
for run in self.runs_executing(ns, hash)? {
if seen_run.insert(run.clone()) {
runs.push(run);
}
}
Ok(runs)
}
pub fn runs_executing(&mut self, ns: &str, hash: &Hash) -> Result<Vec<String>> {
require_exact_ns("runs_executing", ns)?;
let is_definition = matches!(self.get(hash), Ok(g)
if g.grain_type == areev_core::types::GrainType::Tool
&& g.get_str("kind") == Some("definition"));
if !is_definition {
return Ok(Vec::new());
}
let hex = hash.to_hex();
let mut runs: Vec<String> = Vec::new();
let mut seen: HashSet<String> = HashSet::new();
const SPEC_SCAN: usize = 200_000;
for g in self.recent(ns, Some(areev_core::types::GrainType::Tool), SPEC_SCAN)? {
if g.get_str("spec_hash") != Some(hex.as_str()) {
continue;
}
if let Some(r) = g.get_str("run_id") {
if seen.insert(r.to_string()) {
runs.push(r.to_string());
}
}
}
Ok(runs)
}
pub fn namespaces(&mut self) -> Result<Vec<(String, i64)>> {
let mut out: Vec<(String, i64)> = self
.db
.query("SELECT t.term, r.n FROM ns_reg r JOIN terms t ON t.id = r.ns", vec![])?
.iter()
.filter_map(|row| Some((row.text(0)?.to_string(), row.i64(1)?)))
.collect();
out.sort();
Ok(out)
}
pub fn namespaces_in_scope(&mut self, scope: &NsScope) -> Result<Vec<String>> {
if let NsScope::Exact(ns) = scope {
return Ok(vec![ns.clone()]);
}
let mut out: Vec<String> = self
.db
.query("SELECT t.term FROM ns_reg r JOIN terms t ON t.id = r.ns", vec![])?
.iter()
.filter_map(|row| row.text(0))
.filter(|term| scope.matches(term))
.map(str::to_string)
.collect();
out.sort();
Ok(out)
}
fn ns_param_ids(&mut self, ns: &str) -> Result<Vec<i64>> {
if !NsScope::is_pattern(ns) {
return Ok(self.term_lookup(ns)?.into_iter().collect());
}
let scope = NsScope::parse(ns)?;
let names = self.namespaces_in_scope(&scope)?;
let mut ids = Vec::with_capacity(names.len());
for n in &names {
if let Some(id) = self.term_lookup(n)? {
ids.push(id);
}
}
Ok(ids)
}
pub fn recent_ordered(
&mut self,
ns: &str,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
order: RecentOrder,
) -> Result<Vec<DeserializedGrain>> {
let ids = self.ns_param_ids(ns)?;
let hint = (!NsScope::is_pattern(ns)).then_some(ns);
self.recent_ids_ordered(&ids, hint, gtype, limit, live_only, order)
}
pub fn recent_ordered_scoped(
&mut self,
ns_list: &[String],
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
order: RecentOrder,
) -> Result<Vec<DeserializedGrain>> {
let mut ids = Vec::with_capacity(ns_list.len());
for ns in ns_list {
require_exact_ns("a resolved recall scope element", ns)?;
if let Some(id) = self.term_lookup(ns)? {
ids.push(id);
}
}
let single = (ns_list.len() == 1).then(|| ns_list[0].as_str());
self.recent_ids_ordered(&ids, single, gtype, limit, live_only, order)
}
pub fn recent(
&mut self,
ns: &str,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
) -> Result<Vec<DeserializedGrain>> {
self.recent_inner(ns, gtype, limit, false)
}
pub fn recent_live(
&mut self,
ns: &str,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
) -> Result<Vec<DeserializedGrain>> {
self.recent_inner(ns, gtype, limit, true)
}
pub fn recent_scoped(
&mut self,
ns_list: &[String],
gtype: Option<areev_core::types::GrainType>,
limit: usize,
) -> Result<Vec<DeserializedGrain>> {
self.recent_scoped_inner(ns_list, gtype, limit, false)
}
pub fn recent_live_scoped(
&mut self,
ns_list: &[String],
gtype: Option<areev_core::types::GrainType>,
limit: usize,
) -> Result<Vec<DeserializedGrain>> {
self.recent_scoped_inner(ns_list, gtype, limit, true)
}
fn recent_scoped_inner(
&mut self,
ns_list: &[String],
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
) -> Result<Vec<DeserializedGrain>> {
let mut ids = Vec::with_capacity(ns_list.len());
for ns in ns_list {
require_exact_ns("a resolved recall scope element", ns)?;
if let Some(id) = self.term_lookup(ns)? {
ids.push(id);
}
}
let single = (ns_list.len() == 1).then(|| ns_list[0].as_str());
self.recent_ids(&ids, single, gtype, limit, live_only)
}
fn recent_inner(
&mut self,
ns: &str,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
) -> Result<Vec<DeserializedGrain>> {
let ids = self.ns_param_ids(ns)?;
let hint = (!NsScope::is_pattern(ns)).then_some(ns);
self.recent_ids(&ids, hint, gtype, limit, live_only)
}
fn recent_ids(
&mut self,
ns_ids: &[i64],
egress_hint: Option<&str>,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
) -> Result<Vec<DeserializedGrain>> {
self.recent_ids_ordered(ns_ids, egress_hint, gtype, limit, live_only, RecentOrder::Seq)
}
fn recent_ids_ordered(
&mut self,
ns_ids: &[i64],
egress_hint: Option<&str>,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
order: RecentOrder,
) -> Result<Vec<DeserializedGrain>> {
if ns_ids.is_empty() {
return Ok(Vec::new());
}
let live = if live_only {
" AND superseded_by IS NULL"
} else {
""
};
let gt_ord = gtype.map(|g| g as u8 as i64);
let (sort_col, sort_sql) = match order {
RecentOrder::Seq => ("seq", "seq DESC"),
RecentOrder::CreatedAtDesc => ("created_at", "created_at DESC, seq DESC"),
RecentOrder::CreatedAtAsc => ("created_at", "created_at ASC, seq ASC"),
};
let mut merged: Vec<((i64, i64), Vec<u8>)> = Vec::new();
for ns_id in ns_ids {
let rows = match gt_ord {
Some(gt) => self.db.query(
&format!("SELECT {sort_col}, seq, blob FROM grains WHERE ns=?1 AND gtype=?2{live} ORDER BY {sort_sql} LIMIT ?3"),
vec![pi(*ns_id), pi(gt), pi(limit as i64)],
)?,
None => self.db.query(
&format!("SELECT {sort_col}, seq, blob FROM grains WHERE ns=?1{live} ORDER BY {sort_sql} LIMIT ?2"),
vec![pi(*ns_id), pi(limit as i64)],
)?,
};
for row in &rows {
if let (Some(key), Some(seq), Some(b)) = (row.i64(0), row.i64(1), row.blob(2)) {
merged.push(((key, seq), b));
}
}
}
if ns_ids.len() > 1 {
match order {
RecentOrder::CreatedAtAsc => merged.sort_unstable_by_key(|(k, _)| *k),
_ => merged.sort_unstable_by_key(|(k, _)| std::cmp::Reverse(*k)),
}
merged.truncate(limit);
}
let mut out: Vec<DeserializedGrain> = merged
.iter()
.map(|(_, b)| deserialize_blob(b))
.collect::<Result<_>>()?;
self.egress_exit(egress_hint, &mut out)?;
Ok(out)
}
pub fn recent_in_session(
&mut self,
ns: &str,
session: &str,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
) -> Result<Vec<DeserializedGrain>> {
let ids = self.ns_param_ids(ns)?;
let hint = (!NsScope::is_pattern(ns)).then_some(ns);
self.session_ids(&ids, hint, session, gtype, limit, live_only)
}
pub fn recent_in_session_scoped(
&mut self,
ns_list: &[String],
session: &str,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
) -> Result<Vec<DeserializedGrain>> {
let mut ids = Vec::with_capacity(ns_list.len());
for ns in ns_list {
require_exact_ns("a resolved recall scope element", ns)?;
if let Some(id) = self.term_lookup(ns)? {
ids.push(id);
}
}
let single = (ns_list.len() == 1).then(|| ns_list[0].as_str());
self.session_ids(&ids, single, session, gtype, limit, live_only)
}
fn session_ids(
&mut self,
ns_ids: &[i64],
egress_hint: Option<&str>,
session: &str,
gtype: Option<areev_core::types::GrainType>,
limit: usize,
live_only: bool,
) -> Result<Vec<DeserializedGrain>> {
let sess_id = match self.term_lookup(session)? {
Some(x) => x,
None => return Ok(Vec::new()),
};
if ns_ids.is_empty() {
return Ok(Vec::new());
}
let live = if live_only {
" AND g.superseded_by IS NULL"
} else {
""
};
let gt_ord = gtype.map(|g| g as u8 as i64);
let mut merged: Vec<(i64, Vec<u8>)> = Vec::new();
for ns_id in ns_ids {
let rows = match gt_ord {
Some(gt) => self.db.query(
&format!(
"SELECT ti.seq, g.blob FROM thread_idx ti JOIN grains g ON g.seq = ti.seq \
WHERE ti.ns=?1 AND ti.session=?2 AND g.gtype=?3{live} \
ORDER BY ti.seq DESC LIMIT ?4"
),
vec![pi(*ns_id), pi(sess_id), pi(gt), pi(limit as i64)],
)?,
None => self.db.query(
&format!(
"SELECT ti.seq, g.blob FROM thread_idx ti JOIN grains g ON g.seq = ti.seq \
WHERE ti.ns=?1 AND ti.session=?2{live} \
ORDER BY ti.seq DESC LIMIT ?3"
),
vec![pi(*ns_id), pi(sess_id), pi(limit as i64)],
)?,
};
for row in &rows {
if let (Some(seq), Some(b)) = (row.i64(0), row.blob(1)) {
merged.push((seq, b));
}
}
}
if ns_ids.len() > 1 {
merged.sort_unstable_by_key(|(seq, _)| std::cmp::Reverse(*seq));
merged.truncate(limit);
}
let mut out: Vec<DeserializedGrain> = merged
.iter()
.map(|(_, b)| deserialize_blob(b))
.collect::<Result<_>>()?;
self.egress_exit(egress_hint, &mut out)?;
Ok(out)
}
pub fn get(&mut self, hash: &Hash) -> Result<DeserializedGrain> {
let rows = self.db.query(
"SELECT blob FROM grains WHERE hash = ?1",
vec![pb(hash.as_bytes().to_vec())],
)?;
let blob = match rows.first() {
Some(row) => row
.blob(0)
.ok_or_else(|| AreevError::Storage("blob column not a blob".into()))?,
None => return Err(AreevError::NotFound(*hash)),
};
let mut g = deserialize_blob(&blob)?;
self.egress_one(None, &mut g)?;
Ok(g)
}
#[inline]
fn egress_one(&mut self, ns_hint: Option<&str>, g: &mut DeserializedGrain) -> Result<()> {
if self.anon.is_idle() {
return Ok(());
}
self.anon.egress_grains(ns_hint, std::slice::from_mut(g))?;
self.flush_anon_vault()
}
pub fn recall(
&mut self,
ns: &str,
subject: &str,
relation: Option<&str>,
k: usize,
) -> Result<Vec<DeserializedGrain>> {
if NsScope::is_pattern(ns) {
let ids = self.ns_param_ids(ns)?;
return self.recall_multi_raw(ns, &ids, subject, relation, k);
}
let mut out = self.recall_raw(ns, subject, relation, k)?;
self.egress_exit(Some(ns), &mut out)?;
Ok(out)
}
fn recall_multi_raw(
&mut self,
label: &str,
ns_ids: &[i64],
subject: &str,
relation: Option<&str>,
k: usize,
) -> Result<Vec<DeserializedGrain>> {
let start = std::time::Instant::now();
let s_id = match self.term_lookup(subject)? {
Some(x) => x,
None => {
self.record_recall_event(label, Some(subject), relation, None, &[], start);
return Ok(Vec::new());
}
};
let p_id = match relation {
Some(r) => match self.term_lookup(r)? {
Some(x) => Some(x),
None => {
self.record_recall_event(label, Some(subject), relation, None, &[], start);
return Ok(Vec::new());
}
},
None => None,
};
let mut merged: Vec<(i64, Vec<u8>)> = Vec::new();
for ns_id in ns_ids {
let rows = match p_id {
Some(p) => self.db.query_hot(
"SELECT t.seq, g.blob FROM triples t JOIN grains g ON g.seq = t.seq
WHERE t.ns=?1 AND t.s=?2 AND t.p=?3 AND t.cur=1
ORDER BY t.seq DESC LIMIT ?4",
vec![pi(*ns_id), pi(s_id), pi(p), pi(k as i64)],
)?,
None => self.db.query_hot(
"SELECT t.seq, g.blob FROM triples t JOIN grains g ON g.seq = t.seq
WHERE t.ns=?1 AND t.s=?2 AND t.cur=1
ORDER BY t.seq DESC LIMIT ?3",
vec![pi(*ns_id), pi(s_id), pi(k as i64)],
)?,
};
for row in &rows {
if let (Some(seq), Some(b)) = (row.i64(0), row.blob(1)) {
merged.push((seq, b));
}
}
}
merged.sort_unstable_by_key(|(seq, _)| std::cmp::Reverse(*seq));
merged.truncate(k);
let mut out = Vec::with_capacity(merged.len());
for (_, b) in &merged {
out.push(deserialize_blob(b)?);
}
self.record_recall_event(label, Some(subject), relation, None, &out, start);
self.egress_exit(None, &mut out)?;
Ok(out)
}
fn recall_raw(
&mut self,
ns: &str,
subject: &str,
relation: Option<&str>,
k: usize,
) -> Result<Vec<DeserializedGrain>> {
let start = std::time::Instant::now();
let (ns_id, s_id) = match (self.term_lookup(ns)?, self.term_lookup(subject)?) {
(Some(a), Some(b)) => (a, b),
_ => return Ok(Vec::new()),
};
let p_id = match relation {
Some(r) => match self.term_lookup(r)? {
Some(x) => Some(x),
None => return Ok(Vec::new()),
},
None => None,
};
let rows = match p_id {
Some(p) => self.db.query_hot(
"SELECT g.blob FROM triples t JOIN grains g ON g.seq = t.seq
WHERE t.ns=?1 AND t.s=?2 AND t.p=?3 AND t.cur=1
ORDER BY t.seq DESC LIMIT ?4",
vec![pi(ns_id), pi(s_id), pi(p), pi(k as i64)],
)?,
None => self.db.query_hot(
"SELECT g.blob FROM triples t JOIN grains g ON g.seq = t.seq
WHERE t.ns=?1 AND t.s=?2 AND t.cur=1
ORDER BY t.seq DESC LIMIT ?3",
vec![pi(ns_id), pi(s_id), pi(k as i64)],
)?,
};
let mut out = Vec::with_capacity(rows.len());
for row in &rows {
if let Some(b) = row.blob(0) {
out.push(deserialize_blob(&b)?);
}
}
self.record_recall_event(ns, Some(subject), relation, None, &out, start);
Ok(out)
}
pub fn latest(&mut self, ns: &str, subject: &str, relation: &str) -> Result<Option<DeserializedGrain>> {
require_exact_ns("latest", ns)?;
let (ns_id, s_id, p_id) = match (
self.term_lookup(ns)?,
self.term_lookup(subject)?,
self.term_lookup(relation)?,
) {
(Some(a), Some(b), Some(c)) => (a, b, c),
_ => return Ok(None),
};
let hash = self
.db
.query_hot(
"SELECT hash FROM entity_latest WHERE ns=?1 AND s=?2 AND p=?3",
vec![pi(ns_id), pi(s_id), pi(p_id)],
)?
.first()
.and_then(|row| row.blob(0));
match hash {
Some(h) => {
let h = Hash::try_from_bytes(&h)?;
Ok(Some(self.get(&h)?))
}
None => Ok(None),
}
}
pub fn thread_tail(&mut self, ns: &str, session: &str, n: usize) -> Result<Vec<DeserializedGrain>> {
require_exact_ns("thread_tail", ns)?;
let (ns_id, sess_id) = match (self.term_lookup(ns)?, self.term_lookup(session)?) {
(Some(a), Some(b)) => (a, b),
_ => return Ok(Vec::new()),
};
let rows = self.db.query(
"SELECT g.blob FROM thread_idx ti JOIN grains g ON g.seq = ti.seq
WHERE ti.ns=?1 AND ti.session=?2 ORDER BY ti.seq DESC LIMIT ?3",
vec![pi(ns_id), pi(sess_id), pi(n as i64)],
)?;
let mut out = Vec::with_capacity(rows.len());
for row in rows.iter().rev() {
if let Some(b) = row.blob(0) {
out.push(deserialize_blob(&b)?);
}
}
self.egress_exit(Some(ns), &mut out)?;
Ok(out)
}
pub fn supersede<G: Grain + 'static>(&mut self, old: &Hash, new_grain: &mut G) -> Result<Hash> {
self.check_writable("supersede a grain")?;
let rows = self.db.query(
"SELECT seq, ns, s, p, svt FROM grains WHERE hash = ?1",
vec![pb(old.as_bytes().to_vec())],
)?;
let (old_seq, _ns, old_s, old_p, old_svt) = match rows.first() {
Some(row) => (
row.i64(0).unwrap_or(0),
row.i64(1),
row.i64(2),
row.i64(3),
row.i64(4),
),
None => return Err(AreevError::NotFound(*old)),
};
if old_svt.is_some() {
return Err(AreevError::SupersessionConflict(*old));
}
new_grain.common_mut().derived_from = Some(old.to_hex());
let new_dyn: &dyn AddableDyn = &*new_grain;
let (preps, first_seq, first_op, hlc0) = self.prep_and_reserve(&[new_dyn])?;
let new_hash = preps[0].hash;
let now = now_ms();
let ram_op = self.next_op;
self.next_op += 1;
let ram_hlc = self.next_hlc();
let (d_docs, d_len) = fts_delta(&preps);
let dbr = self.db.as_ref();
with_txn(dbr, || {
let (s0, o0, h0, op_seq, hlc) = match dbr.reserve_write(1, 2, 2, now_ms() << 16, 0)? {
Some(w) => (w.seq0, w.op0, w.hlc0, w.op0 + 1, w.hlc0 + 1),
None => (first_seq, first_op, hlc0, ram_op, ram_hlc),
};
let recheck = dbr.query(
&format!("SELECT seq, svt FROM grains WHERE hash=?1{}", dbr.for_update()),
vec![pb(old.as_bytes().to_vec())],
)?;
let old_seq = match recheck.first() {
None => return Err(AreevError::NotFound(*old)),
Some(row) if row.i64(1).is_some() => {
return Err(AreevError::SupersessionConflict(*old))
}
Some(row) => row.i64(0).unwrap_or(old_seq),
};
insert_prepped(dbr, &preps, s0, o0, h0)?;
dbr.execute(
"UPDATE grains SET superseded_by=?1, svt=?2 WHERE seq=?3",
vec![pb(new_hash.as_bytes().to_vec()), pi(now), pi(old_seq)],
)?;
dbr.execute(
"UPDATE grains SET supersedes=?1 WHERE hash=?2",
vec![pb(old.as_bytes().to_vec()), pb(new_hash.as_bytes().to_vec())],
)?;
dbr.execute("UPDATE triples SET cur=0 WHERE seq=?1", vec![pi(old_seq)])?;
dbr.execute("UPDATE osp SET cur=0 WHERE seq=?1", vec![pi(old_seq)])?;
if let (Some(ns), Some(s), Some(p)) = (_ns, old_s, old_p) {
dbr.execute(
"DELETE FROM heads WHERE ns=?1 AND s=?2 AND p=?3 AND seq=?4",
vec![pi(ns), pi(s), pi(p), pi(old_seq)],
)?;
dbr.execute(
"DELETE FROM entity_latest WHERE ns=?1 AND s=?2 AND p=?3 AND seq=?4",
vec![pi(ns), pi(s), pi(p), pi(old_seq)],
)?;
let rows = dbr.query(
"SELECT t.o, h.seq, h.hash, h.created_at
FROM heads h JOIN triples t
ON t.seq=h.seq AND t.ns=h.ns AND t.s=h.s AND t.p=h.p
WHERE h.ns=?1 AND h.s=?2 AND h.p=?3
ORDER BY h.created_at DESC, h.hash DESC LIMIT 1",
vec![pi(ns), pi(s), pi(p)],
)?;
if let Some(row) = rows.first() {
let o = row.i64(0).unwrap_or(0);
let sq = row.i64(1).unwrap_or(0);
let h = row.blob(2).unwrap_or_default();
dbr.execute(
"INSERT OR REPLACE INTO entity_latest(ns,s,p,o,seq,hash) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(ns), pi(s), pi(p), pi(o), pi(sq), pb(h)],
)?;
}
}
dbr.execute(
"INSERT INTO oplog(op_seq,hlc,op,hash) VALUES (?1,?2,?3,?4)",
vec![pi(op_seq), pi(hlc), pi(OP_SUPERSEDE), pb(new_hash.as_bytes().to_vec())],
)?;
Ok(())
})?;
self.fts_docs += d_docs;
self.fts_total_len += d_len;
Ok(new_hash)
}
const AUTHZ_GRANT_CAP: usize = 256;
pub fn authz_grants(&mut self, principal: &str) -> Result<Vec<authz::Grant>> {
let grains = self.recall_raw(
authz::AUTHZ_NS,
principal,
Some(authz::REL_PERMITS),
Self::AUTHZ_GRANT_CAP,
)?;
let mut grants = Vec::new();
for g in &grains {
if let Some(obj) = g.get_str("object") {
if let Ok(grant) = authz::Grant::from_object_string(obj) {
grants.push(grant);
}
}
}
Ok(grants)
}
pub fn forget(&mut self, hash: &Hash) -> Result<()> {
self.check_writable("forget a grain")?;
let rows = self.db.query(
"SELECT ns, s, p, blob FROM grains WHERE hash = ?1",
vec![pb(hash.as_bytes().to_vec())],
)?;
let (ns, s, p) = match rows.first() {
Some(row) => (row.i64(0), row.i64(1), row.i64(2)),
None => return Err(AreevError::NotFound(*hash)),
};
let mut cas_candidates: Vec<String> = rows
.first()
.and_then(|row| row.blob(3))
.and_then(|b| deserialize_blob(&b).ok())
.map(|view| cas_hexes(&view))
.unwrap_or_default();
cas_candidates.sort_unstable();
cas_candidates.dedup();
let ram_op = self.next_op;
self.next_op += 1;
let ram_hlc = self.next_hlc();
let dbr = self.db.as_ref();
let txn_out = with_txn(dbr, || {
let (op_seq, hlc) = match dbr.reserve_write(0, 1, 1, now_ms() << 16, 0)? {
Some(w) => (w.op0, w.hlc0),
None => (ram_op, ram_hlc),
};
let recheck = dbr.query(
&format!("SELECT seq FROM grains WHERE hash=?1{}", dbr.for_update()),
vec![pb(hash.as_bytes().to_vec())],
)?;
let seq = match recheck.first().and_then(|r| r.i64(0)) {
Some(s) => s,
None => return Err(AreevError::NotFound(*hash)),
};
let doc_len = dbr
.query("SELECT len FROM fts_doc WHERE seq = ?1", vec![pi(seq)])?
.first()
.and_then(|row| row.i64(0));
dbr.execute("DELETE FROM triples WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM osp WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM embeddings WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM thread_idx WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM prov_idx WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM run_idx WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM corpus_idx WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM fts_post WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM fts_doc WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("DELETE FROM grains WHERE seq=?1", vec![pi(seq)])?;
if let Some(ns) = ns {
drop_ns_registry(dbr, ns, 1)?;
}
if let (Some(ns), Some(s), Some(p)) = (ns, s, p) {
dbr.execute(
"DELETE FROM heads WHERE ns=?1 AND s=?2 AND p=?3 AND seq=?4",
vec![pi(ns), pi(s), pi(p), pi(seq)],
)?;
dbr.execute(
"DELETE FROM entity_latest WHERE ns=?1 AND s=?2 AND p=?3 AND seq=?4",
vec![pi(ns), pi(s), pi(p), pi(seq)],
)?;
let rows = dbr.query(
"SELECT t.o, h.seq, h.hash, h.created_at
FROM heads h JOIN triples t
ON t.seq=h.seq AND t.ns=h.ns AND t.s=h.s AND t.p=h.p
WHERE h.ns=?1 AND h.s=?2 AND h.p=?3
ORDER BY h.created_at DESC, h.hash DESC LIMIT 1",
vec![pi(ns), pi(s), pi(p)],
)?;
if let Some(row) = rows.first() {
let o = row.i64(0).unwrap_or(0);
let sq = row.i64(1).unwrap_or(0);
let h = row.blob(2).unwrap_or_default();
dbr.execute(
"INSERT OR REPLACE INTO entity_latest(ns,s,p,o,seq,hash) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(ns), pi(s), pi(p), pi(o), pi(sq), pb(h)],
)?;
}
}
dbr.execute(
"INSERT INTO oplog(op_seq,hlc,op,hash) VALUES (?1,?2,?3,?4)",
vec![pi(op_seq), pi(hlc), pi(OP_FORGET), pb(hash.as_bytes().to_vec())],
)?;
let mut fs_blobs: Vec<String> = Vec::new();
if !cas_candidates.is_empty() {
let mut referenced: HashSet<String> = HashSet::new();
for row in dbr.query("SELECT blob FROM grains", vec![])? {
if let Some(b) = row.blob(0) {
if let Ok(view) = deserialize_blob(&b) {
referenced.extend(cas_hexes(&view));
}
}
}
for hex in &cas_candidates {
if referenced.contains(hex) {
continue;
}
match &self.blob_store {
BlobStore::Table => {
if let Ok(raw) = hex::decode(hex) {
dbr.execute("DELETE FROM blobs WHERE hash=?1", vec![pb(raw)])?;
}
}
BlobStore::Fs(_) => fs_blobs.push(hex.clone()),
}
}
}
Ok((doc_len, fs_blobs))
})?;
let (doc_len, fs_blobs) = txn_out;
if let Some(len) = doc_len {
self.fts_docs = (self.fts_docs - 1).max(0);
self.fts_total_len = (self.fts_total_len - len).max(0);
}
if let BlobStore::Fs(dir) = &self.blob_store {
for hex in &fs_blobs {
let _ = std::fs::remove_file(fs_blob_path(dir, hex));
}
}
if let Some(tel) = self.telemetry.as_mut() {
let _ = tel.scrub(hash);
}
Ok(())
}
pub fn forget_subject(&mut self, ns: &str, subject: &str) -> Result<ErasureReport> {
self.forget_subject_with(ns, subject, ErasureOptions::default())
}
pub fn forget_subject_with(
&mut self,
ns: &str,
subject: &str,
opts: ErasureOptions,
) -> Result<ErasureReport> {
require_exact_ns("forget_subject", ns)?;
self.check_subject_selector("forget_subject", subject, opts.text_mentions)?;
let Some(ns_id) = self.term_lookup(ns)? else {
return Ok(ErasureReport::default());
};
let mut report = self.erase_where(
ErasureSelector::Identity {
ns_id,
subject: subject.to_string(),
text_mentions: opts.text_mentions,
},
Some(subject.to_string()),
)?;
let mut scrub_names = vec![subject.to_string()];
if let Some(alias) = self.anon.ingress_alias(ns, subject)? {
scrub_names.push(alias);
}
self.scrub_anon_identities(ns, &scrub_names)?;
if let Some(alias) = self.anon.ingress_alias(ns, subject)? {
if let Some(ns_id) = self.term_lookup(ns)? {
let alias_report = self.erase_where(
ErasureSelector::Identity {
ns_id,
subject: alias.clone(),
text_mentions: opts.text_mentions,
},
Some(alias),
)?;
report.grains_erased += alias_report.grains_erased;
report.terms_removed += alias_report.terms_removed;
report.vocab_removed += alias_report.vocab_removed;
report.blobs_reclaimed += alias_report.blobs_reclaimed;
report.erased_hashes.extend(alias_report.erased_hashes);
}
}
Ok(report)
}
fn check_subject_selector(&self, op: &str, subject: &str, text_mentions: bool) -> Result<()> {
if subject.trim().is_empty() {
return Err(AreevError::Validation(format!("{op} needs a non-empty subject")));
}
if text_mentions && (!self.index_text || self.fts_deferred) {
return Err(AreevError::Validation(
"text-mention erasure needs the text index on and fully built for this \
memory (the BM25 index is what makes prose mentions findable; finish the \
deferred rebuild first)"
.into(),
));
}
Ok(())
}
pub fn subject_report(&mut self, ns: &str, subject: &str) -> Result<SubjectReport> {
self.subject_report_with(ns, subject, ErasureOptions::default())
}
pub fn subject_report_with(
&mut self,
ns: &str,
subject: &str,
opts: ErasureOptions,
) -> Result<SubjectReport> {
require_exact_ns("subject_report", ns)?;
self.check_subject_selector("subject_report", subject, opts.text_mentions)?;
let Some(ns_id) = self.term_lookup(ns)? else {
return Ok(SubjectReport::default());
};
let mut sel = self.select_identity(ns_id, subject, opts.text_mentions)?;
if let Some(alias) = self.anon.ingress_alias(ns, subject)? {
let alias_sel = self.select_identity(ns_id, &alias, opts.text_mentions)?;
sel.seqs.extend(alias_sel.seqs);
sel.identity_names.extend(alias_sel.identity_names);
}
let sel = sel;
let mut seqs = sel.seqs;
seqs.sort_unstable();
seqs.dedup();
let mut grains = Vec::with_capacity(seqs.len());
if !seqs.is_empty() {
let csv = seq_csv(&seqs);
for row in self
.db
.query(&format!("SELECT blob FROM grains WHERE seq IN ({csv}) ORDER BY seq"), vec![])?
{
if let Some(b) = row.blob(0) {
grains.push(deserialize_blob(&b)?);
}
}
}
Ok(SubjectReport { identity_names: sel.identity_names, grains })
}
pub fn subject_bundle_with(
&mut self,
ns: &str,
subject: &str,
opts: ErasureOptions,
path: &str,
) -> Result<BundleStats> {
require_exact_ns("subject_bundle", ns)?;
self.check_subject_selector("subject_bundle", subject, opts.text_mentions)?;
let mut out: Vec<u8> = Vec::with_capacity(16 * 1024);
out.extend_from_slice(BUNDLE_MAGIC);
let mut ops = 0usize;
let mut last = 0i64;
if let Some(ns_id) = self.term_lookup(ns)? {
let sel = self.select_identity(ns_id, subject, opts.text_mentions)?;
let mut seqs = sel.seqs;
seqs.sort_unstable();
seqs.dedup();
if !seqs.is_empty() {
let csv = seq_csv(&seqs);
for row in self.db.query(
&format!(
"SELECT o.op_seq, o.hlc, o.op, g.hash, g.blob \
FROM oplog o JOIN grains g ON g.hash = o.hash \
WHERE g.seq IN ({csv}) AND o.op != ?1 ORDER BY o.op_seq"
),
vec![pi(OP_FORGET)],
)? {
let (Some(op_seq), Some(hlc), Some(op), Some(hash), Some(blob)) =
(row.i64(0), row.i64(1), row.i64(2), row.blob(3), row.blob(4))
else {
continue;
};
out.push(op as u8);
out.extend_from_slice(&hlc.to_le_bytes());
out.extend_from_slice(&hash);
out.extend_from_slice(&(blob.len() as u32).to_le_bytes());
out.extend_from_slice(&blob);
ops += 1;
last = op_seq;
}
}
}
std::fs::write(path, &out).map_err(db_err)?;
Ok(BundleStats {
ops,
bytes: std::fs::metadata(path).map(|m| m.len()).unwrap_or(0),
last_op_seq: last,
})
}
pub fn forget_older_than(
&mut self,
ns: Option<&str>,
cutoff_ms: i64,
gtype: Option<areev_core::types::GrainType>,
) -> Result<ErasureReport> {
self.forget_older_than_capped(ns, cutoff_ms, gtype, None)
}
pub fn forget_older_than_capped(
&mut self,
ns: Option<&str>,
cutoff_ms: i64,
gtype: Option<areev_core::types::GrainType>,
limit: Option<usize>,
) -> Result<ErasureReport> {
if limit == Some(0) {
return Ok(ErasureReport::default());
}
if let Some(n) = ns {
require_exact_ns("an age-based sweep", n)?;
}
self.check_age_destruction_allowed(ns, cutoff_ms)?;
let ns_id = match ns {
Some(n) => match self.term_lookup(n)? {
Some(x) => Some(x),
None => return Ok(ErasureReport::default()),
},
None => None,
};
let gt = gtype.map(|g| g as u8 as i64);
self.erase_where(
ErasureSelector::OlderThan { ns_id, cutoff_ms, gtype: gt, limit },
None,
)
}
fn select_identity(
&mut self,
ns_id: i64,
subject: &str,
text_mentions: bool,
) -> Result<IdentitySelection> {
let mut sel = IdentitySelection::default();
let like = format!("{}%", like_escape(subject));
for row in self.db.query(
"SELECT id, term FROM terms WHERE term LIKE ?1 ESCAPE '\\'",
vec![pt(&like)],
)? {
let (Some(id), Some(term)) = (row.i64(0), row.text(1)) else {
continue;
};
let matched = term == subject
|| (term.starts_with(subject)
&& term[subject.len()..]
.chars()
.next()
.is_some_and(|c| !c.is_alphanumeric()));
if matched {
sel.identity_ids.push(id);
sel.identity_names.push(term.to_string());
}
}
if !sel.identity_names.iter().any(|n| n == subject) {
sel.identity_names.push(subject.to_string());
}
if !sel.identity_ids.is_empty() {
let idcsv = seq_csv(&sel.identity_ids);
sel.seqs.extend(
self.db
.query(
&format!(
"SELECT DISTINCT seq FROM triples WHERE ns=?1 AND \
(s IN ({idcsv}) OR o IN ({idcsv})) \
UNION SELECT seq FROM thread_idx WHERE ns=?1 AND session IN ({idcsv}) \
UNION SELECT seq FROM run_idx WHERE ns=?1 AND run IN ({idcsv})"
),
vec![pi(ns_id)],
)?
.iter()
.filter_map(|r| r.i64(0)),
);
}
if text_mentions {
let mut tokens = tokenize(subject);
tokens.sort_unstable();
tokens.dedup();
let mut inter: Option<HashSet<i64>> = None;
for t in &tokens {
let set: HashSet<i64> = match self.fts_term_lookup(t)? {
Some(vid) => self
.db
.query(
"SELECT seq FROM fts_post WHERE term=?1 AND ns=?2",
vec![pi(vid), pi(ns_id)],
)?
.iter()
.filter_map(|r| r.i64(0))
.collect(),
None => HashSet::new(),
};
inter = Some(match inter {
None => set,
Some(p) => p.intersection(&set).copied().collect(),
});
if inter.as_ref().is_some_and(|s| s.is_empty()) {
break;
}
}
if let Some(set) = inter {
sel.seqs.extend(set);
}
}
Ok(sel)
}
fn erase_where(
&mut self,
selector: ErasureSelector,
identity: Option<String>,
) -> Result<ErasureReport> {
struct Target {
seq: i64,
hash: Vec<u8>,
ns: i64,
s: Option<i64>,
p: Option<i64>,
o: Option<i64>,
doc_len: Option<i64>,
blob: Vec<u8>,
}
struct EraseOut {
report: ErasureReport,
hashes: Vec<Vec<u8>>,
docs: i64,
len_sum: i64,
fs_blobs: Vec<String>,
scrubbed_terms: Vec<i64>,
identity_names: Vec<String>,
}
self.check_writable("erase (forget-subject / retention sweep)")?;
self.db.begin()?;
let r = (|| -> Result<EraseOut> {
let mut out = EraseOut {
report: ErasureReport::default(),
hashes: Vec::new(),
docs: 0,
len_sum: 0,
fs_blobs: Vec::new(),
scrubbed_terms: Vec::new(),
identity_names: Vec::new(),
};
self.db.reserve_write(0, 0, 0, now_ms() << 16, 0)?;
let mut identity_ids: Vec<i64> = Vec::new();
let mut seqs: Vec<i64> = match &selector {
ErasureSelector::Identity { ns_id, subject, text_mentions } => {
let sel = self.select_identity(*ns_id, subject, *text_mentions)?;
identity_ids = sel.identity_ids;
out.identity_names = sel.identity_names;
sel.seqs
}
ErasureSelector::OlderThan { ns_id, cutoff_ms, gtype, limit } => {
let mut sql = String::from("SELECT seq FROM grains WHERE created_at < ?1");
let mut params = vec![pi(*cutoff_ms)];
if let Some(n) = ns_id {
sql.push_str(" AND ns = ?2");
params.push(pi(*n));
}
if let Some(g) = gtype {
sql.push_str(if ns_id.is_some() { " AND gtype = ?3" } else { " AND gtype = ?2" });
params.push(pi(*g));
}
if let Some(n) = limit {
sql.push_str(&format!(" ORDER BY created_at ASC, seq ASC LIMIT {n}"));
}
self.db.query(&sql, params)?.iter().filter_map(|r| r.i64(0)).collect()
}
};
seqs.sort_unstable();
seqs.dedup();
if seqs.is_empty() {
for id in &identity_ids {
let gone = self.scrub_term_if_unreferenced(*id)?;
out.report.terms_removed += gone as usize;
if gone > 0 {
out.scrubbed_terms.push(*id);
}
}
return Ok(out);
}
let n = seqs.len() as i64;
let (op0, hlc0) = match self.db.reserve_write(0, n, n, now_ms() << 16, 0)? {
Some(w) => (w.op0, w.hlc0),
None => {
let op0 = self.next_op;
self.next_op += n;
let hlc0 = self.next_hlc();
self.hlc_last = hlc0 + n - 1;
(op0, hlc0)
}
};
let csv = seq_csv(&seqs);
let vocab_candidates: Vec<i64> = self
.db
.query(&format!("SELECT DISTINCT term FROM fts_post WHERE seq IN ({csv})"), vec![])?
.iter()
.filter_map(|r| r.i64(0))
.collect();
let mut term_candidates: Vec<i64> = Vec::new();
for sql in [
format!("SELECT DISTINCT session FROM thread_idx WHERE seq IN ({csv})"),
format!("SELECT DISTINCT run FROM run_idx WHERE seq IN ({csv})"),
] {
for row in self.db.query(&sql, vec![])? {
if let Some(id) = row.i64(0) {
term_candidates.push(id);
}
}
}
term_candidates.extend(identity_ids.iter().copied());
let mut targets: Vec<Target> = Vec::with_capacity(seqs.len());
for row in self.db.query(
&format!(
"SELECT g.seq, g.hash, g.ns, g.s, g.p, g.o, d.len, g.blob \
FROM grains g LEFT JOIN fts_doc d ON d.seq = g.seq \
WHERE g.seq IN ({csv})"
),
vec![],
)? {
let (Some(seq), Some(hash), Some(ns), Some(blob)) =
(row.i64(0), row.blob(1), row.i64(2), row.blob(7))
else {
continue;
};
targets.push(Target {
seq,
hash,
ns,
s: row.i64(3),
p: row.i64(4),
o: row.i64(5),
doc_len: row.i64(6),
blob,
});
}
for t in &targets {
for id in [t.s, t.p, t.o].into_iter().flatten() {
term_candidates.push(id);
}
}
term_candidates.sort_unstable();
term_candidates.dedup();
let mut affected_keys: Vec<(i64, i64, i64)> = Vec::new();
for (i, t) in targets.iter().enumerate() {
for table in [
"triples",
"osp",
"embeddings",
"thread_idx",
"prov_idx",
"run_idx",
"corpus_idx",
"fts_post",
"fts_doc",
]
{
self.db
.execute(&format!("DELETE FROM {table} WHERE seq=?1"), vec![pi(t.seq)])?;
}
self.db.execute("DELETE FROM grains WHERE seq=?1", vec![pi(t.seq)])?;
if let (Some(s), Some(p)) = (t.s, t.p) {
self.db.execute(
"DELETE FROM heads WHERE ns=?1 AND s=?2 AND p=?3 AND seq=?4",
vec![pi(t.ns), pi(s), pi(p), pi(t.seq)],
)?;
if !affected_keys.contains(&(t.ns, s, p)) {
affected_keys.push((t.ns, s, p));
}
}
self.db.execute(
"INSERT INTO oplog(op_seq,hlc,op,hash) VALUES (?1,?2,?3,?4)",
vec![
pi(op0 + i as i64),
pi(hlc0 + i as i64),
pi(OP_FORGET),
pb(t.hash.clone()),
],
)?;
}
let mut ns_counts: Vec<(i64, i64)> = Vec::new();
for t in &targets {
match ns_counts.iter_mut().find(|(ns, _)| *ns == t.ns) {
Some((_, k)) => *k += 1,
None => ns_counts.push((t.ns, 1)),
}
}
for (ns, k) in &ns_counts {
drop_ns_registry(self.db.as_ref(), *ns, *k)?;
}
for (ns, s, p) in &affected_keys {
self.db.execute(
&format!(
"DELETE FROM entity_latest WHERE ns=?1 AND s=?2 AND p=?3 AND seq IN ({csv})"
),
vec![pi(*ns), pi(*s), pi(*p)],
)?;
let rows = self.db.query(
"SELECT t.o, h.seq, h.hash, h.created_at
FROM heads h JOIN triples t
ON t.seq=h.seq AND t.ns=h.ns AND t.s=h.s AND t.p=h.p
WHERE h.ns=?1 AND h.s=?2 AND h.p=?3
ORDER BY h.created_at DESC, h.hash DESC LIMIT 1",
vec![pi(*ns), pi(*s), pi(*p)],
)?;
if let Some(row) = rows.first() {
let o = row.i64(0).unwrap_or(0);
let sq = row.i64(1).unwrap_or(0);
let h = row.blob(2).unwrap_or_default();
self.db.execute(
"INSERT OR REPLACE INTO entity_latest(ns,s,p,o,seq,hash) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(*ns), pi(*s), pi(*p), pi(o), pi(sq), pb(h)],
)?;
}
}
for term in vocab_candidates {
let gone = self.db.execute(
"DELETE FROM fts_vocab WHERE id=?1 AND NOT EXISTS \
(SELECT 1 FROM fts_post WHERE term=?1)",
vec![pi(term)],
)?;
out.report.vocab_removed += gone as usize;
}
for id in term_candidates {
let gone = self.scrub_term_if_unreferenced(id)?;
out.report.terms_removed += gone as usize;
if gone > 0 {
out.scrubbed_terms.push(id);
}
}
let mut cas_candidates: Vec<String> = Vec::new();
for t in &targets {
if let Ok(view) = deserialize_blob(&t.blob) {
cas_candidates.extend(cas_hexes(&view));
}
}
cas_candidates.sort_unstable();
cas_candidates.dedup();
if !cas_candidates.is_empty() {
let mut referenced: HashSet<String> = HashSet::new();
for row in self.db.query("SELECT blob FROM grains", vec![])? {
if let Some(b) = row.blob(0) {
if let Ok(view) = deserialize_blob(&b) {
referenced.extend(cas_hexes(&view));
}
}
}
for hex in cas_candidates {
if referenced.contains(&hex) {
continue;
}
match &self.blob_store {
BlobStore::Table => {
let Ok(raw) = hex::decode(&hex) else { continue };
let gone = self
.db
.execute("DELETE FROM blobs WHERE hash=?1", vec![pb(raw)])?;
out.report.blobs_reclaimed += gone as usize;
}
BlobStore::Fs(_) => out.fs_blobs.push(hex),
}
}
}
out.report.grains_erased = targets.len();
out.docs = targets.iter().filter(|t| t.doc_len.is_some()).count() as i64;
out.len_sum = targets.iter().filter_map(|t| t.doc_len).sum();
out.hashes = targets.into_iter().map(|t| t.hash).collect();
Ok(out)
})();
match r {
Ok(mut out) => {
self.db.commit()?;
out.report.erased_hashes = out
.hashes
.iter()
.filter_map(|bytes| Hash::try_from_bytes(bytes).ok())
.map(|hash| hash.to_hex())
.collect();
self.fts_docs = (self.fts_docs - out.docs).max(0);
self.fts_total_len = (self.fts_total_len - out.len_sum).max(0);
if !out.scrubbed_terms.is_empty() {
self.dict.retain(|_, id| !out.scrubbed_terms.contains(id));
}
if out.report.vocab_removed > 0 {
self.fts_vocab.clear();
}
if let BlobStore::Fs(dir) = &self.blob_store {
let dir = dir.clone();
for hex in &out.fs_blobs {
if std::fs::remove_file(fs_blob_path(&dir, hex)).is_ok() {
out.report.blobs_reclaimed += 1;
}
}
}
if self.telemetry.is_some() {
let hashes: Vec<Hash> = out
.hashes
.iter()
.filter_map(|h| Hash::try_from_bytes(h).ok())
.collect();
if let Some(tel) = self.telemetry.as_mut() {
for h in &hashes {
let _ = tel.scrub(h);
}
if identity.is_some() {
for name in &out.identity_names {
let _ = tel.scrub_subject(name);
}
}
}
}
Ok(out.report)
}
Err(e) => {
let _ = self.db.rollback();
Err(e)
}
}
}
fn scrub_term_if_unreferenced(&mut self, id: i64) -> Result<u64> {
let placeholder = format!("\u{1}erased:{id}");
self.db.scrub_term(id, &placeholder)
}
fn log_op(&mut self, op: i64, hash: &Hash, hlc: i64) -> Result<()> {
let ram_op = self.next_op;
self.next_op += 1;
let dbr = self.db.as_ref();
with_txn(dbr, || {
let op_seq = match dbr.reserve_write(0, 1, 0, now_ms() << 16, hlc)? {
Some(w) => w.op0,
None => ram_op,
};
dbr.execute(
"INSERT INTO oplog(op_seq,hlc,op,hash) VALUES (?1,?2,?3,?4)",
vec![pi(op_seq), pi(hlc), pi(op), pb(hash.as_bytes().to_vec())],
)?;
Ok(())
})
}
pub fn telemetry_mode(&self) -> TelemetryMode {
self.telemetry
.as_ref()
.map(|t| t.mode())
.unwrap_or(TelemetryMode::Off)
}
pub fn telemetry_flush(&mut self) -> Result<()> {
if let Some(tel) = self.telemetry.as_mut() {
tel.flush()?;
}
Ok(())
}
pub fn telemetry_note_budget(&mut self, overflow: bool) -> Result<()> {
if let Some(tel) = self.telemetry.as_mut() {
tel.note_budget(overflow)?;
}
Ok(())
}
pub fn telemetry_access_stats(&mut self, ns: Option<&str>) -> Result<Vec<AccessStat>> {
self.telemetry_flush()?;
match self.telemetry.as_ref() {
Some(tel) => tel.access_stats(ns),
None => Ok(Vec::new()),
}
}
pub fn telemetry_query_stats(&mut self, ns: Option<&str>) -> Result<Vec<QueryStat>> {
self.telemetry_flush()?;
match self.telemetry.as_ref() {
Some(tel) => tel.query_stats(ns),
None => Ok(Vec::new()),
}
}
pub fn telemetry_recall_log(
&mut self,
run_id: Option<&str>,
) -> Result<Vec<RecallLogEntry>> {
self.telemetry_flush()?;
match self.telemetry.as_ref() {
Some(tel) => tel.recall_log(run_id),
None => Ok(Vec::new()),
}
}
pub fn telemetry_budget_stats(&mut self) -> Result<BudgetStat> {
self.telemetry_flush()?;
match self.telemetry.as_ref() {
Some(tel) => tel.budget_stats(),
None => Ok(BudgetStat::default()),
}
}
pub fn step_actions(
&mut self,
ns: &str,
workflow_hash: &Hash,
node_id: Option<&str>,
limit: usize,
) -> Result<Vec<(String, Hash)>> {
require_exact_ns("step_actions", ns)?;
let Some(ns_id) = self.term_lookup(ns)? else {
return Ok(Vec::new());
};
let Some(target_id) = self.term_lookup(&workflow_hash.to_hex())? else {
return Ok(Vec::new());
};
let pred_ids: Vec<i64> = match node_id {
Some(n) => match self.term_lookup(&step_action_relation(n))? {
Some(id) => vec![id],
None => return Ok(Vec::new()),
},
None => self
.terms_with_prefix(STEP_ACTION_PREFIX)?
.into_iter()
.map(|(id, _)| id)
.collect(),
};
if pred_ids.is_empty() {
return Ok(Vec::new());
}
let limit = limit.min(1024);
let mut rows: Vec<(i64, i64, i64)> = Vec::new();
for p in &pred_ids {
for row in self.db.query(
"SELECT seq, s, p FROM osp WHERE ns=?1 AND o=?2 AND p=?3 AND cur=1
ORDER BY seq DESC LIMIT ?4",
vec![pi(ns_id), pi(target_id), pi(*p), pi(limit as i64)],
)? {
let (Some(seq), Some(s), Some(p)) = (row.i64(0), row.i64(1), row.i64(2)) else {
continue;
};
rows.push((seq, s, p));
}
}
rows.sort_unstable_by_key(|(seq, _, _)| std::cmp::Reverse(*seq));
let mut result = Vec::with_capacity(rows.len());
for (_, s, p) in rows {
let (Some(subj), Some(rel)) = (self.term_str(s)?, self.term_str(p)?) else {
continue;
};
let (Some(node), Ok(h)) = (step_action_node(&rel), Hash::from_hex(&subj)) else {
continue;
};
result.push((node.to_string(), h));
}
result.truncate(limit);
Ok(result)
}
pub fn step_actions_page(
&mut self,
ns: &str,
workflow_hash: &Hash,
node_id: Option<&str>,
after_seq: i64,
limit: usize,
) -> Result<Vec<(i64, String, Hash)>> {
require_exact_ns("step_actions_page", ns)?;
let Some(ns_id) = self.term_lookup(ns)? else {
return Ok(Vec::new());
};
let Some(target_id) = self.term_lookup(&workflow_hash.to_hex())? else {
return Ok(Vec::new());
};
let pred_ids: Vec<i64> = match node_id {
Some(n) => match self.term_lookup(&step_action_relation(n))? {
Some(id) => vec![id],
None => return Ok(Vec::new()),
},
None => self
.terms_with_prefix(STEP_ACTION_PREFIX)?
.into_iter()
.map(|(id, _)| id)
.collect(),
};
if pred_ids.is_empty() {
return Ok(Vec::new());
}
let limit = limit.clamp(1, 1024);
let mut rows: Vec<(i64, i64, i64)> = Vec::new();
for p in &pred_ids {
for row in self.db.query(
"SELECT seq, s, p FROM osp WHERE ns=?1 AND o=?2 AND p=?3 AND cur=1
AND seq > ?4 ORDER BY seq ASC LIMIT ?5",
vec![pi(ns_id), pi(target_id), pi(*p), pi(after_seq), pi(limit as i64)],
)? {
let (Some(seq), Some(s), Some(p)) = (row.i64(0), row.i64(1), row.i64(2)) else {
continue;
};
rows.push((seq, s, p));
}
}
rows.sort_unstable_by_key(|(seq, _, _)| *seq);
rows.truncate(limit);
let mut result = Vec::with_capacity(rows.len());
for (seq, s, p) in rows {
let (Some(subj), Some(rel)) = (self.term_str(s)?, self.term_str(p)?) else {
continue;
};
let (Some(node), Ok(h)) = (step_action_node(&rel), Hash::from_hex(&subj)) else {
continue;
};
result.push((seq, node.to_string(), h));
}
Ok(result)
}
fn terms_with_prefix(&self, prefix: &str) -> Result<Vec<(i64, String)>> {
if let Some(rows) = self.db.terms_with_prefix(prefix)? {
return Ok(rows);
}
Ok(self
.dict
.iter()
.filter(|(term, _)| term.starts_with(prefix))
.map(|(term, id)| (*id, term.clone()))
.collect())
}
pub fn grains_by_object(
&mut self,
ns: &str,
object: &str,
limit: usize,
) -> Result<Vec<DeserializedGrain>> {
require_exact_ns("grains_by_object", ns)?;
let (Some(ns_id), Some(obj_id)) = (self.term_lookup(ns)?, self.term_lookup(object)?) else {
return Ok(Vec::new());
};
let limit = limit.min(1024) as i64;
let mut out: Vec<DeserializedGrain> = self
.db
.query(
"SELECT g.blob FROM osp o JOIN grains g ON g.seq = o.seq
WHERE o.ns = ?1 AND o.o = ?2 AND o.cur = 1 ORDER BY o.seq DESC LIMIT ?3",
vec![pi(ns_id), pi(obj_id), pi(limit)],
)?
.iter()
.filter_map(|row| row.blob(0))
.map(|b| deserialize_blob(&b))
.collect::<Result<_>>()?;
self.egress_exit(Some(ns), &mut out)?;
Ok(out)
}
pub fn related(
&mut self,
ns: &str,
start: &str,
relations: &[&str],
dir: Direction,
depth: usize,
cap: usize,
) -> Result<Vec<String>> {
require_exact_ns("related", ns)?;
let ns_id = match self.term_lookup(ns)? {
Some(x) => x,
None => return Ok(Vec::new()),
};
let start_id = match self.term_lookup(start)? {
Some(x) => x,
None => return Ok(Vec::new()),
};
let mut rel_ids: Vec<i64> = Vec::new();
for r in relations {
if let Some(id) = self.term_lookup(r)? {
rel_ids.push(id);
}
}
if rel_ids.is_empty() {
return Ok(Vec::new());
}
let depth = depth.min(4);
let cap = cap.min(512);
let reached = 'bfs: {
let mut seen: HashSet<i64> = HashSet::new();
seen.insert(start_id);
let mut order: Vec<i64> = Vec::new();
let mut frontier = vec![start_id];
for _ in 0..depth {
let mut next = Vec::new();
for node in &frontier {
for p in &rel_ids {
if matches!(dir, Direction::Out | Direction::Both) {
for row in self.db.query(
"SELECT o FROM triples WHERE ns=?1 AND s=?2 AND p=?3 AND cur=1 LIMIT 64",
vec![pi(ns_id), pi(*node), pi(*p)],
)? {
if let Some(o) = row.i64(0) {
if seen.insert(o) {
order.push(o);
next.push(o);
if order.len() >= cap {
break 'bfs order;
}
}
}
}
}
if matches!(dir, Direction::In | Direction::Both) {
for row in self.db.query(
"SELECT s FROM osp WHERE ns=?1 AND o=?2 AND p=?3 AND cur=1 LIMIT 64",
vec![pi(ns_id), pi(*node), pi(*p)],
)? {
if let Some(s) = row.i64(0) {
if seen.insert(s) {
order.push(s);
next.push(s);
if order.len() >= cap {
break 'bfs order;
}
}
}
}
}
}
}
if next.is_empty() {
break;
}
frontier = next;
}
order
};
let mut out = Vec::with_capacity(reached.len());
for id in reached {
if let Some(t) = self.term_str(id)? {
out.push(t);
}
}
self.egress_strings_exit(ns, None, &mut out)?;
Ok(out)
}
pub fn path(
&mut self,
ns: &str,
from: &str,
to: &str,
relations: &[&str],
max_depth: usize,
) -> Result<Option<Vec<String>>> {
require_exact_ns("path", ns)?;
let ns_id = match self.term_lookup(ns)? {
Some(x) => x,
None => return Ok(None),
};
let (a, b) = match (self.term_lookup(from)?, self.term_lookup(to)?) {
(Some(a), Some(b)) => (a, b),
_ => return Ok(None),
};
let mut rel_ids: Vec<i64> = Vec::new();
for r in relations {
if let Some(id) = self.term_lookup(r)? {
rel_ids.push(id);
}
}
if rel_ids.is_empty() {
return Ok(None);
}
let max_depth = max_depth.min(6);
let parents = {
let mut parent: HashMap<i64, i64> = HashMap::new();
let mut q = VecDeque::from([a]);
let mut found = false;
let mut hops = 0usize;
let mut visited: HashSet<i64> = HashSet::from([a]);
'outer: while !q.is_empty() && hops < max_depth {
let level: Vec<i64> = q.drain(..).collect();
for node in level {
for p in &rel_ids {
for row in self.db.query(
"SELECT o FROM triples WHERE ns=?1 AND s=?2 AND p=?3 AND cur=1 LIMIT 64",
vec![pi(ns_id), pi(node), pi(*p)],
)? {
if let Some(o) = row.i64(0) {
if visited.insert(o) {
parent.insert(o, node);
if o == b {
found = true;
break 'outer;
}
q.push_back(o);
if visited.len() > 2048 {
break 'outer;
}
}
}
}
}
}
hops += 1;
}
if found { Some(parent) } else { None }
};
let Some(parent) = parents else {
return Ok(None);
};
let mut chain = vec![b];
let mut cur = b;
while let Some(pr) = parent.get(&cur) {
chain.push(*pr);
cur = *pr;
if cur == a {
break;
}
}
chain.reverse();
let mut out = Vec::with_capacity(chain.len());
for id in chain {
if let Some(t) = self.term_str(id)? {
out.push(t);
}
}
self.egress_strings_exit(ns, None, &mut out)?;
Ok(Some(out))
}
pub fn entity_at(
&mut self,
ns: &str,
subject: &str,
relation: &str,
t: i64,
axis: Axis,
) -> Result<Option<DeserializedGrain>> {
require_exact_ns("entity_at", ns)?;
match axis {
Axis::Knowledge => {
let head = match self.latest(ns, subject, relation)? {
Some(g) => g.hash,
None => return Ok(None),
};
let mut cur = head;
loop {
let rows = self.db.query(
"SELECT svf, supersedes, blob FROM grains WHERE hash = ?1",
vec![pb(cur.as_bytes().to_vec())],
)?;
let (svf, sup, blob) = match rows.first() {
Some(row) => (row.i64(0), row.blob(1), row.blob(2)),
None => return Ok(None),
};
if svf.unwrap_or(i64::MIN) <= t {
let mut g = deserialize_blob(&blob.unwrap_or_default())?;
self.egress_one(Some(ns), &mut g)?;
return Ok(Some(g));
}
match sup {
Some(prev) => cur = Hash::try_from_bytes(&prev)?,
None => return Ok(None),
}
}
}
Axis::World => {
let (ns_id, s_id, p_id) = match (
self.term_lookup(ns)?,
self.term_lookup(subject)?,
self.term_lookup(relation)?,
) {
(Some(a), Some(b), Some(c)) => (a, b, c),
_ => return Ok(None),
};
let blob = self
.db
.query(
"SELECT g.blob FROM triples tr JOIN grains g ON g.seq = tr.seq
WHERE tr.ns=?1 AND tr.s=?2 AND tr.p=?3
AND g.svt IS NULL
AND (g.vf IS NULL OR g.vf <= ?4)
AND (g.vt IS NULL OR g.vt > ?4)
ORDER BY tr.seq DESC LIMIT 1",
vec![pi(ns_id), pi(s_id), pi(p_id), pi(t)],
)?
.first()
.and_then(|row| row.blob(0));
match blob {
Some(b) => {
let mut g = deserialize_blob(&b)?;
self.egress_one(Some(ns), &mut g)?;
Ok(Some(g))
}
None => Ok(None),
}
}
}
}
pub fn has(&mut self, hash: &Hash) -> Result<bool> {
self.has_grain(hash)
}
pub fn search_text(&mut self, ns: &str, query: &str, k: usize) -> Result<Vec<i64>> {
let ids = self.ns_param_ids(ns)?;
self.search_text_inner(&ids, query, k, false)
}
pub fn search_text_all(&mut self, ns: &str, query: &str, k: usize) -> Result<Vec<i64>> {
let ids = self.ns_param_ids(ns)?;
self.search_text_inner(&ids, query, k, true)
}
fn search_text_inner(
&mut self,
ns_ids: &[i64],
query: &str,
k: usize,
include_superseded: bool,
) -> Result<Vec<i64>> {
if !self.index_text || k == 0 || ns_ids.is_empty() {
return Ok(Vec::new()); }
let terms = tokenize(query);
if terms.is_empty() {
return Ok(Vec::new());
}
let (docs_i, len_i) = match self.db.collection_stats()? {
Some(s) => s,
None => (self.fts_docs, self.fts_total_len),
};
let n_docs = docs_i.max(1) as f64;
let avgdl = if docs_i > 0 { (len_i as f64 / docs_i as f64).max(1.0) } else { 1.0 };
let mut distinct: Vec<&String> = Vec::new();
{
let mut seen: HashSet<&str> = HashSet::new();
for t in &terms {
if seen.insert(t.as_str()) {
distinct.push(t);
}
}
}
let mut by_term: HashMap<String, Vec<(i64, i64, i64)>> = HashMap::new();
if (distinct.len() == 1 || !self.db.prefers_batched_reads()) && ns_ids.len() == 1 {
let ns_id = ns_ids[0];
for t in &distinct {
for row in &self.db.query_hot(
"SELECT v.term, p.seq, p.tf, d.len FROM fts_post p
JOIN fts_vocab v ON v.id = p.term
JOIN fts_doc d ON d.seq = p.seq
WHERE v.term = ?1 AND p.ns = ?2",
vec![pt(t), pi(ns_id)],
)? {
if let (Some(t), Some(seq), Some(tf), Some(dl)) =
(row.text(0), row.i64(1), row.i64(2), row.i64(3))
{
by_term.entry(t.to_string()).or_default().push((seq, tf, dl.max(1)));
}
}
}
} else {
let placeholders: Vec<String> =
(1..=distinct.len()).map(|i| format!("?{i}")).collect();
let sql = format!(
"SELECT v.term, p.seq, p.tf, d.len FROM fts_post p
JOIN fts_vocab v ON v.id = p.term
JOIN fts_doc d ON d.seq = p.seq
WHERE p.ns IN ({}) AND v.term IN ({})",
seq_csv(ns_ids),
placeholders.join(",")
);
let params: Vec<Value> = distinct.iter().map(|t| pt(t)).collect();
for row in &self.db.query(&sql, params)? {
if let (Some(t), Some(seq), Some(tf), Some(dl)) =
(row.text(0), row.i64(1), row.i64(2), row.i64(3))
{
by_term.entry(t.to_string()).or_default().push((seq, tf, dl.max(1)));
}
}
}
let mut scores: HashMap<i64, f64> = HashMap::new();
for term in terms {
let Some(postings) = by_term.get(&term) else {
continue;
};
let df = postings.len() as f64;
let idf = (1.0 + (n_docs - df + 0.5) / (df + 0.5)).ln();
for (seq, tf, dl) in postings {
let tf = *tf as f64;
let norm = BM25_K1 * (1.0 - BM25_B + BM25_B * (*dl as f64 / avgdl));
*scores.entry(*seq).or_insert(0.0) += idf * (tf * (BM25_K1 + 1.0)) / (tf + norm);
}
}
if scores.is_empty() {
return Ok(Vec::new());
}
let mut ranked: Vec<(i64, f64)> = scores.into_iter().collect();
ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)
.then(b.0.cmp(&a.0)));
if include_superseded {
ranked.truncate(k);
return Ok(ranked.into_iter().map(|(s, _)| s).collect());
}
ranked.truncate(k.saturating_mul(4).max(k));
let live = self.live_seqs(ranked.iter().map(|(s, _)| *s))?;
Ok(ranked
.into_iter()
.filter(|(s, _)| live.contains(s))
.map(|(s, _)| s)
.take(k)
.collect())
}
fn live_seqs(&mut self, seqs: impl Iterator<Item = i64>) -> Result<HashSet<i64>> {
let list: Vec<String> = seqs.map(|s| s.to_string()).collect();
if list.is_empty() {
return Ok(HashSet::new());
}
let sql = format!(
"SELECT seq FROM grains WHERE svt IS NULL AND seq IN ({})",
list.join(",")
);
Ok(self
.db
.query(&sql, vec![])?
.iter()
.filter_map(|row| row.i64(0))
.collect())
}
pub fn supersession_map(&mut self, hashes: &[Hash]) -> Result<HashMap<Hash, Hash>> {
let mut out = HashMap::new();
if hashes.is_empty() {
return Ok(out);
}
if !self.db.prefers_batched_reads() {
for h in hashes {
if let Some(sup) = self
.db
.query_hot(
"SELECT superseded_by FROM grains WHERE hash = ?1",
vec![pb(h.as_bytes().to_vec())],
)?
.first()
.and_then(|row| row.blob(0))
.and_then(|b| Hash::try_from_bytes(&b).ok())
{
out.insert(*h, sup);
}
}
return Ok(out);
}
let placeholders: Vec<String> = (1..hashes.len() + 1).map(|i| format!("?{i}")).collect();
let sql = format!(
"SELECT hash, superseded_by FROM grains
WHERE superseded_by IS NOT NULL AND hash IN ({})",
placeholders.join(",")
);
let params: Vec<Value> = hashes.iter().map(|h| pb(h.as_bytes().to_vec())).collect();
for row in &self.db.query(&sql, params)? {
if let (Some(h), Some(sup)) = (
row.blob(0).and_then(|b| Hash::try_from_bytes(&b).ok()),
row.blob(1).and_then(|b| Hash::try_from_bytes(&b).ok()),
) {
out.insert(h, sup);
}
}
Ok(out)
}
pub fn current_head(&self, hash: &Hash) -> Result<Hash> {
let mut cur = *hash;
for _ in 0..MAX_SUPERSESSION_CHAIN_HOPS {
let rows = self.db.query(
"SELECT superseded_by FROM grains WHERE hash = ?1",
vec![pb(cur.as_bytes().to_vec())],
)?;
match rows.first() {
None => return Ok(cur),
Some(row) => match row.blob(0).and_then(|b| Hash::try_from_bytes(&b).ok()) {
Some(next) => cur = next,
None => return Ok(cur),
},
}
}
Err(AreevError::SupersessionChainTooDeep(*hash))
}
pub fn supersession_chain(&self, hash: &Hash) -> Result<Vec<Hash>> {
let mut out = vec![*hash];
let mut cur = *hash;
for _ in 0..MAX_SUPERSESSION_CHAIN_HOPS {
let rows = self.db.query(
"SELECT supersedes FROM grains WHERE hash = ?1",
vec![pb(cur.as_bytes().to_vec())],
)?;
let supersedes = match rows.first() {
Some(row) => row.blob(0).and_then(|b| Hash::try_from_bytes(&b).ok()),
None => return Ok(out),
};
match supersedes {
Some(parent) => {
out.push(parent);
cur = parent;
}
None => return Ok(out),
}
}
Err(AreevError::SupersessionChainTooDeep(*hash))
}
fn check_embedding_dim(&self, dim: usize) -> Result<()> {
if dim == 0 {
return Err(AreevError::Validation(
"embedding vector must be non-empty".into(),
));
}
if let Some((model, declared)) = &self.meta_embed {
if *declared != dim {
return Err(AreevError::Validation(format!(
"embedding dimension mismatch: file declares {model}@{declared}, \
caller supplied a {dim}-dimensional vector"
)));
}
}
Ok(())
}
pub fn nearest_vector(
&mut self,
ns: &str,
subject: Option<&str>,
relation: Option<&str>,
vector: &[f32],
k: usize,
) -> Result<Vec<(Hash, f32)>> {
self.check_embedding_dim(vector.len())?;
let qjson = vec_to_json(vector);
self.nearest_scoped(ns, subject, relation, &qjson, k)
}
pub fn nearest_semantic(
&mut self,
ns: &str,
subject: Option<&str>,
relation: Option<&str>,
text: &str,
k: usize,
) -> Result<Vec<(Hash, f32)>> {
if self.embedder.is_none() {
return Err(AreevError::Validation(
"semantic nearest-neighbour search requires an installed embedder; \
none is installed"
.into(),
));
}
let embedder = self.embedder.as_ref().expect("checked above");
let qjson = vec_to_json(&embedder.embed(text)?);
self.nearest_scoped(ns, subject, relation, &qjson, k)
}
fn nearest_scoped(
&mut self,
ns: &str,
subject: Option<&str>,
relation: Option<&str>,
qjson: &str,
k: usize,
) -> Result<Vec<(Hash, f32)>> {
let ns_ids = self.ns_param_ids(ns)?;
if ns_ids.is_empty() {
return Ok(Vec::new());
}
let s_id = match subject {
Some(s) => match self.term_lookup(s)? {
Some(x) => Some(x),
None => return Ok(Vec::new()),
},
None => None,
};
let p_id = match relation {
Some(r) => match self.term_lookup(r)? {
Some(x) => Some(x),
None => return Ok(Vec::new()),
},
None => None,
};
let rows = if let [ns_id] = ns_ids[..] {
let base = "SELECT g.hash, vector_distance_cos(e.vec, vector32(?2)) AS dist \
FROM embeddings e JOIN grains g ON g.seq = e.seq \
WHERE g.ns = ?1 AND g.svt IS NULL";
match (s_id, p_id) {
(Some(s), Some(p)) => self.db.query(
&format!("{base} AND g.s = ?3 AND g.p = ?4 ORDER BY dist LIMIT ?5"),
vec![pi(ns_id), pt(qjson), pi(s), pi(p), pi(k as i64)],
)?,
(Some(s), None) => self.db.query(
&format!("{base} AND g.s = ?3 ORDER BY dist LIMIT ?4"),
vec![pi(ns_id), pt(qjson), pi(s), pi(k as i64)],
)?,
(None, Some(p)) => self.db.query(
&format!("{base} AND g.p = ?3 ORDER BY dist LIMIT ?4"),
vec![pi(ns_id), pt(qjson), pi(p), pi(k as i64)],
)?,
(None, None) => self.db.query(
&format!("{base} ORDER BY dist LIMIT ?3"),
vec![pi(ns_id), pt(qjson), pi(k as i64)],
)?,
}
} else {
let base = format!(
"SELECT g.hash, vector_distance_cos(e.vec, vector32(?1)) AS dist \
FROM embeddings e JOIN grains g ON g.seq = e.seq \
WHERE g.ns IN ({}) AND g.svt IS NULL",
seq_csv(&ns_ids)
);
match (s_id, p_id) {
(Some(s), Some(p)) => self.db.query(
&format!("{base} AND g.s = ?2 AND g.p = ?3 ORDER BY dist LIMIT ?4"),
vec![pt(qjson), pi(s), pi(p), pi(k as i64)],
)?,
(Some(s), None) => self.db.query(
&format!("{base} AND g.s = ?2 ORDER BY dist LIMIT ?3"),
vec![pt(qjson), pi(s), pi(k as i64)],
)?,
(None, Some(p)) => self.db.query(
&format!("{base} AND g.p = ?2 ORDER BY dist LIMIT ?3"),
vec![pt(qjson), pi(p), pi(k as i64)],
)?,
(None, None) => self.db.query(
&format!("{base} ORDER BY dist LIMIT ?2"),
vec![pt(qjson), pi(k as i64)],
)?,
}
};
let mut out = Vec::new();
for row in rows {
let h = row.blob(0).and_then(|b| Hash::try_from_bytes(&b).ok());
let dist = row.f64(1).unwrap_or(1.0);
if let Some(h) = h {
out.push((h, (1.0 - dist) as f32));
}
}
Ok(out)
}
pub fn search_vector(&mut self, ns: &str, query: &str, k: usize) -> Result<Vec<i64>> {
let ids = self.ns_param_ids(ns)?;
self.search_vector_inner(&ids, query, k, false)
}
pub fn search_vector_all(&mut self, ns: &str, query: &str, k: usize) -> Result<Vec<i64>> {
let ids = self.ns_param_ids(ns)?;
self.search_vector_inner(&ids, query, k, true)
}
fn search_vector_inner(
&mut self,
ns_ids: &[i64],
query: &str,
k: usize,
include_superseded: bool,
) -> Result<Vec<i64>> {
if ns_ids.is_empty() {
return Ok(Vec::new());
}
let Some(embedder) = &self.embedder else {
return Ok(Vec::new());
};
let qv = embedder.embed(query)?;
let qjson = vec_to_json(&qv);
if let [ns_id] = ns_ids {
return Ok(self
.db
.query(
if include_superseded {
"SELECT e.seq FROM embeddings e JOIN grains g ON g.seq = e.seq
WHERE g.ns = ?1
ORDER BY vector_distance_cos(e.vec, vector32(?2)) LIMIT ?3"
} else {
"SELECT e.seq FROM embeddings e JOIN grains g ON g.seq = e.seq
WHERE g.ns = ?1 AND g.svt IS NULL
ORDER BY vector_distance_cos(e.vec, vector32(?2)) LIMIT ?3"
},
vec![pi(*ns_id), pt(&qjson), pi(k as i64)],
)?
.iter()
.filter_map(|row| row.i64(0))
.collect());
}
let live = if include_superseded { "" } else { " AND g.svt IS NULL" };
let sql = format!(
"SELECT e.seq FROM embeddings e JOIN grains g ON g.seq = e.seq
WHERE g.ns IN ({}){live}
ORDER BY vector_distance_cos(e.vec, vector32(?1)) LIMIT ?2",
seq_csv(ns_ids)
);
Ok(self
.db
.query(&sql, vec![pt(&qjson), pi(k as i64)])?
.iter()
.filter_map(|row| row.i64(0))
.collect())
}
pub fn recall_hybrid(
&mut self,
ns: &str,
subject: Option<&str>,
relation: Option<&str>,
query: Option<&str>,
k: usize,
deadline: Option<std::time::Duration>,
) -> Result<Vec<DeserializedGrain>> {
self.recall_hybrid_tuned(ns, subject, relation, query, k, deadline, RecallTuning::default())
}
#[allow(clippy::too_many_arguments)] pub fn recall_hybrid_tuned(
&mut self,
ns: &str,
subject: Option<&str>,
relation: Option<&str>,
query: Option<&str>,
k: usize,
deadline: Option<std::time::Duration>,
tuning: RecallTuning,
) -> Result<Vec<DeserializedGrain>> {
let ns_ids = self.ns_param_ids(ns)?;
let hint = (!NsScope::is_pattern(ns)).then_some(ns);
self.recall_hybrid_ids(ns, &ns_ids, hint, subject, relation, query, k, deadline, tuning)
}
#[allow(clippy::too_many_arguments)]
pub fn recall_hybrid_scoped(
&mut self,
ns_list: &[String],
subject: Option<&str>,
relation: Option<&str>,
query: Option<&str>,
k: usize,
deadline: Option<std::time::Duration>,
tuning: RecallTuning,
) -> Result<Vec<DeserializedGrain>> {
let mut ns_ids = Vec::with_capacity(ns_list.len());
for ns in ns_list {
require_exact_ns("a resolved recall scope element", ns)?;
if let Some(id) = self.term_lookup(ns)? {
ns_ids.push(id);
}
}
let label = ns_list.join(",");
let hint = (ns_list.len() == 1).then(|| ns_list[0].as_str());
self.recall_hybrid_ids(&label, &ns_ids, hint, subject, relation, query, k, deadline, tuning)
}
#[allow(clippy::too_many_arguments)]
fn recall_hybrid_ids(
&mut self,
label: &str,
ns_ids: &[i64],
egress_hint: Option<&str>,
subject: Option<&str>,
relation: Option<&str>,
query: Option<&str>,
k: usize,
deadline: Option<std::time::Duration>,
tuning: RecallTuning,
) -> Result<Vec<DeserializedGrain>> {
let start = std::time::Instant::now();
let over = |start: &std::time::Instant| match deadline {
Some(d) => start.elapsed() >= d,
None => false,
};
let refine = tuning.rerank || tuning.diversity_lambda.is_some();
let leg_k = if refine {
k.max(REFINE_POOL)
} else {
k.saturating_mul(2)
};
let structural: Vec<i64> = match subject {
Some(s) => self.recall_seqs(ns_ids, s, relation, leg_k, tuning.include_superseded)?,
None => Vec::new(),
};
let mut fts_legs: Vec<Vec<i64>> = Vec::new();
if let Some(q) = query {
if !over(&start) {
fts_legs.push(
self.search_text_inner(ns_ids, q, leg_k, tuning.include_superseded)
.unwrap_or_default(),
);
if tuning.query_expansion && self.index_text {
for variant in self.expand_query(q) {
if over(&start) {
break;
}
let hits = self
.search_text_inner(ns_ids, &variant, leg_k, tuning.include_superseded)
.unwrap_or_default();
if !hits.is_empty() {
fts_legs.push(hits);
}
}
}
}
}
let vecs: Vec<i64> = match query {
Some(q) if self.embedder.is_some() && !over(&start) => {
self.search_vector_inner(ns_ids, q, leg_k, tuning.include_superseded)
.unwrap_or_default()
}
_ => Vec::new(),
};
if structural.is_empty() && fts_legs.iter().all(|l| l.is_empty()) && vecs.is_empty() {
self.record_recall_event(label, subject, relation, query, &[], start);
return Ok(Vec::new());
}
let mut scores: HashMap<i64, f64> = HashMap::new();
for (rank, seq) in structural.iter().enumerate() {
*scores.entry(*seq).or_insert(0.0) += 1.0 / (RRF_K0 + rank as f64);
}
for leg in &fts_legs {
for (rank, seq) in leg.iter().enumerate() {
*scores.entry(*seq).or_insert(0.0) += 1.0 / (RRF_K0 + rank as f64);
}
}
for (rank, seq) in vecs.iter().enumerate() {
*scores.entry(*seq).or_insert(0.0) += 1.0 / (RRF_K0 + rank as f64);
}
let mut ranked: Vec<(i64, f64)> = scores.into_iter().collect();
ranked.sort_by(|a, b| {
b.1.partial_cmp(&a.1)
.unwrap_or(std::cmp::Ordering::Equal)
.then(b.0.cmp(&a.0))
});
let ordered: Vec<i64> = if let Some(q) =
query.filter(|_| tuning.rerank && self.reranker.is_some() && !over(&start))
{
self.rerank_pool(q, &ranked, k)?
} else if let (Some(lambda), Some(q)) = (tuning.diversity_lambda, query) {
if self.embedder.is_some() && !over(&start) {
self.mmr_pool(q, &ranked, lambda, k)?
} else {
ranked.iter().take(k).map(|(s, _)| *s).collect()
}
} else {
ranked.iter().take(k).map(|(s, _)| *s).collect()
};
let mut out = Vec::new();
if self.db.prefers_batched_reads() && !ordered.is_empty() && !over(&start) {
for chunk in ordered.chunks(16) {
if over(&start) {
break;
}
let blobs = self.blobs_by_seqs(chunk)?;
for seq in chunk {
if over(&start) {
break;
}
if let Some(b) = blobs.get(seq) {
out.push(deserialize_blob(b)?);
}
}
}
} else {
for seq in ordered {
if over(&start) {
break; }
if let Some(b) = self.blob_by_seq(seq)? {
out.push(deserialize_blob(&b)?);
}
}
}
self.record_recall_event(label, subject, relation, query, &out, start);
self.egress_exit(egress_hint, &mut out)?;
Ok(out)
}
#[inline]
fn record_recall_event(
&mut self,
ns: &str,
subject: Option<&str>,
relation: Option<&str>,
query: Option<&str>,
out: &[DeserializedGrain],
start: std::time::Instant,
) {
let run_id = self.ambient_run_id.clone();
if let Some(tel) = self.telemetry.as_mut() {
tel.record(RecallEvent {
ts_ms: now_ms(),
run_id,
ns: ns.to_string(),
subject: subject.map(str::to_string),
relation: relation.map(str::to_string),
query: query.map(str::to_string),
n_results: out.len(),
latency_us: start.elapsed().as_micros() as i64,
hashes: out.iter().map(|g| g.hash).collect(),
});
}
}
fn expand_query(&self, q: &str) -> Vec<String> {
match &self.expander {
Some(e) => e.expand(q),
None => EnglishExpander::default().expand(q),
}
}
fn candidate_text(&mut self, seq: i64) -> Result<String> {
let Some(b) = self.blob_by_seq(seq)? else {
return Ok(String::new());
};
let g = deserialize_blob(&b)?;
Ok(projected_text(&g).unwrap_or_default())
}
fn rerank_pool(&mut self, query: &str, ranked: &[(i64, f64)], k: usize) -> Result<Vec<i64>> {
let pool_n = ranked.len().min(k.max(REFINE_POOL));
let pool: Vec<i64> = ranked.iter().take(pool_n).map(|(s, _)| *s).collect();
if pool.is_empty() {
return Ok(pool);
}
let mut docs: Vec<String> = Vec::with_capacity(pool.len());
for &seq in &pool {
docs.push(self.candidate_text(seq)?);
}
let refs: Vec<&str> = docs.iter().map(|s| s.as_str()).collect();
let reranker = self.reranker.as_ref().expect("caller checked reranker present");
match reranker.rerank(query, &refs) {
Ok(scores) if scores.len() == refs.len() => {
let mut scored: Vec<(i64, f32)> = pool.iter().copied().zip(scores).collect();
scored.sort_by(|a, b| {
b.1.partial_cmp(&a.1)
.unwrap_or(std::cmp::Ordering::Equal)
.then(a.0.cmp(&b.0))
});
Ok(scored.into_iter().take(k).map(|(s, _)| s).collect())
}
_ => Ok(pool.into_iter().take(k).collect()),
}
}
fn mmr_pool(
&mut self,
query: &str,
ranked: &[(i64, f64)],
lambda: f32,
k: usize,
) -> Result<Vec<i64>> {
let lambda = lambda.clamp(0.0, 1.0);
let pool_n = ranked.len().min(k.max(REFINE_POOL));
let pool: Vec<i64> = ranked.iter().take(pool_n).map(|(s, _)| *s).collect();
if pool.len() < 2 {
return Ok(pool);
}
let qv = match &self.embedder {
Some(e) => e.embed(query)?,
None => return Ok(pool.into_iter().take(k).collect()),
};
let qjson = vec_to_json(&qv);
let rel = self.vec_rel_map(&pool, &qjson)?;
let embedded: Vec<i64> = pool.iter().copied().filter(|s| rel.contains_key(s)).collect();
if embedded.len() < 2 {
return Ok(pool.into_iter().take(k).collect());
}
let sim = self.vec_pairwise_map(&embedded)?;
let sim_of = |a: i64, b: i64| -> f32 {
if a == b {
1.0
} else {
let key = if a < b { (a, b) } else { (b, a) };
*sim.get(&key).unwrap_or(&0.0)
}
};
let target = k.min(embedded.len());
let mut selected: Vec<i64> = Vec::with_capacity(target);
let mut remaining: Vec<i64> = embedded.clone();
while selected.len() < target && !remaining.is_empty() {
let mut best_idx = 0usize;
let mut best_score = f32::MIN;
for (i, &c) in remaining.iter().enumerate() {
let relevance = *rel.get(&c).unwrap_or(&0.0);
let max_sim = selected
.iter()
.map(|&s| sim_of(c, s))
.fold(0.0f32, f32::max);
let mmr = lambda * relevance - (1.0 - lambda) * max_sim;
if mmr > best_score {
best_score = mmr;
best_idx = i;
}
}
selected.push(remaining.remove(best_idx));
}
for s in pool {
if selected.len() >= k {
break;
}
if !selected.contains(&s) {
selected.push(s);
}
}
selected.truncate(k);
Ok(selected)
}
fn vec_rel_map(&mut self, seqs: &[i64], qjson: &str) -> Result<HashMap<i64, f32>> {
if seqs.is_empty() {
return Ok(HashMap::new());
}
let sql = format!(
"SELECT seq, vector_distance_cos(vec, vector32(?1)) FROM embeddings WHERE seq IN ({})",
seq_csv(seqs)
);
let mut out = HashMap::new();
for row in self.db.query(&sql, vec![pt(qjson)])? {
if let (Some(s), Some(d)) = (row.i64(0), row.f64(1)) {
out.insert(s, 1.0 - d as f32);
}
}
Ok(out)
}
fn vec_pairwise_map(&mut self, seqs: &[i64]) -> Result<HashMap<(i64, i64), f32>> {
if seqs.len() < 2 {
return Ok(HashMap::new());
}
let csv = seq_csv(seqs);
let sql = format!(
"SELECT a.seq, b.seq, vector_distance_cos(a.vec, b.vec) \
FROM embeddings a JOIN embeddings b ON a.seq < b.seq \
WHERE a.seq IN ({csv}) AND b.seq IN ({csv})"
);
let mut out = HashMap::new();
for row in self.db.query(&sql, vec![])? {
if let (Some(a), Some(b), Some(d)) = (row.i64(0), row.i64(1), row.f64(2)) {
out.insert((a, b), 1.0 - d as f32);
}
}
Ok(out)
}
fn recall_seqs(
&mut self,
ns_ids: &[i64],
subject: &str,
relation: Option<&str>,
k: usize,
all_versions: bool,
) -> Result<Vec<i64>> {
if ns_ids.is_empty() {
return Ok(Vec::new());
}
let s_id = match self.term_lookup(subject)? {
Some(x) => x,
None => return Ok(Vec::new()),
};
let p_id = match relation {
Some(r) => match self.term_lookup(r)? {
Some(x) => Some(x),
None => return Ok(Vec::new()),
},
None => None,
};
let mut all: Vec<i64> = Vec::new();
for ns_id in ns_ids {
all.extend(self.recall_seqs_probe(*ns_id, s_id, p_id, k, all_versions)?);
}
if ns_ids.len() > 1 {
all.sort_unstable_by(|a, b| b.cmp(a));
all.truncate(k);
}
Ok(all)
}
fn recall_seqs_probe(
&mut self,
ns_id: i64,
s_id: i64,
p_id: Option<i64>,
k: usize,
all_versions: bool,
) -> Result<Vec<i64>> {
let rows = match (p_id, all_versions) {
(Some(p), true) => self.db.query_hot(
"SELECT seq FROM triples WHERE ns=?1 AND s=?2 AND p=?3 ORDER BY seq DESC LIMIT ?4",
vec![pi(ns_id), pi(s_id), pi(p), pi(k as i64)],
)?,
(Some(p), false) => self.db.query_hot(
"SELECT seq FROM triples WHERE ns=?1 AND s=?2 AND p=?3 AND cur=1 ORDER BY seq DESC LIMIT ?4",
vec![pi(ns_id), pi(s_id), pi(p), pi(k as i64)],
)?,
(None, true) => self.db.query_hot(
"SELECT seq FROM triples WHERE ns=?1 AND s=?2 ORDER BY seq DESC LIMIT ?3",
vec![pi(ns_id), pi(s_id), pi(k as i64)],
)?,
(None, false) => self.db.query_hot(
"SELECT seq FROM triples WHERE ns=?1 AND s=?2 AND cur=1 ORDER BY seq DESC LIMIT ?3",
vec![pi(ns_id), pi(s_id), pi(k as i64)],
)?,
};
Ok(rows.iter().filter_map(|row| row.i64(0)).collect())
}
fn blob_by_seq(&mut self, seq: i64) -> Result<Option<Vec<u8>>> {
Ok(self
.db
.query_hot("SELECT blob FROM grains WHERE seq = ?1", vec![pi(seq)])?
.first()
.and_then(|row| row.blob(0)))
}
fn blobs_by_seqs(&mut self, seqs: &[i64]) -> Result<HashMap<i64, Vec<u8>>> {
let mut out = HashMap::new();
if seqs.is_empty() {
return Ok(out);
}
let sql = format!("SELECT seq, blob FROM grains WHERE seq IN ({})", seq_csv(seqs));
for row in &self.db.query(&sql, vec![])? {
if let (Some(s), Some(b)) = (row.i64(0), row.blob(1)) {
out.insert(s, b);
}
}
Ok(out)
}
pub fn subjects_with_relation(&mut self, ns: &str, relation: &str) -> Result<Vec<String>> {
let (ns_id, p_id) = match (self.term_lookup(ns)?, self.term_lookup(relation)?) {
(Some(a), Some(b)) => (a, b),
_ => return Ok(Vec::new()),
};
let ids: Vec<i64> = self
.db
.query(
"SELECT DISTINCT s FROM triples WHERE ns=?1 AND p=?2 AND cur=1",
vec![pi(ns_id), pi(p_id)],
)?
.iter()
.filter_map(|row| row.i64(0))
.collect();
let mut subjects: Vec<String> = Vec::with_capacity(ids.len());
for id in ids {
if let Some(t) = self.term_str(id)? {
subjects.push(t);
}
}
subjects.sort();
self.egress_strings_exit(ns, Some("person"), &mut subjects)?;
Ok(subjects)
}
pub fn capture(&mut self, ns: &str, content: &str, meta: &Capture<'_>) -> Result<Hash> {
use areev_core::types::{Event, Role};
let ingressed = self.ingress_transform_text(ns, content)?;
let content: &str = ingressed.as_deref().unwrap_or(content);
let mut e = Event::new(content);
e.common.namespace = Some(ns.to_string());
e.session_id = meta.session_id.map(str::to_string);
e.role = meta.role.and_then(Role::from_str);
e.run_id = match meta.run_id.filter(|r| !r.is_empty()) {
Some(r) => {
areev_core::types::validate_run_id(r)?;
Some(r.to_string())
}
None => None,
};
if let Some(observer) = meta.observer.filter(|o| !o.is_empty()) {
e.common
.extra_fields
.insert("observer".to_string(), serde_json::json!(observer));
}
self.add(&e)
}
pub fn attach_facts(
&mut self,
ns: &str,
source: &Hash,
drafts: &[FactDraft],
attr: &FactAttribution<'_>,
) -> Result<Vec<Hash>> {
let source_hex = source.to_hex();
let mut facts = Vec::with_capacity(drafts.len());
for draft in drafts {
let subject = match self.ingress_transform_identity(ns, &draft.subject)? {
Some(t) => t,
None => draft.subject.clone(),
};
let object = match self.ingress_transform_text(ns, &draft.object)? {
Some(t) => t,
None => draft.object.clone(),
};
let mut fact = areev_core::types::Fact::new(&subject, &draft.relation, &object);
fact.common.confidence = draft.confidence.clamp(0.0, 1.0);
fact.common.namespace = Some(ns.to_string());
fact.common.derived_from = Some(source_hex.clone());
fact.common.source_type = Some("derived".to_string());
if let Some(status) = attr.verification_status {
fact.common.verification_status = Some(status.to_string());
}
if let Some(model) = attr.extractor_model {
fact.common
.extra_fields
.insert("extractor_model".to_string(), serde_json::json!(model));
}
facts.push(self.add(&fact)?);
}
Ok(facts)
}
#[allow(clippy::type_complexity)] pub fn remember(
&mut self,
ns: &str,
content: &str,
observer: &str,
extractor: Option<&dyn Fn(&str) -> Vec<FactDraft>>,
) -> Result<RememberResult> {
let ingressed = self.ingress_transform_text(ns, content)?;
let content: &str = ingressed.as_deref().unwrap_or(content);
let event = self.capture(ns, content, &Capture { observer: Some(observer), ..Default::default() })?;
let drafts = extractor.map(|f| f(content)).unwrap_or_default();
let facts = self.attach_facts(ns, &event, &drafts, &FactAttribution::default())?;
Ok(RememberResult { event, facts })
}
pub fn count(&mut self) -> Result<usize> {
Ok(self
.db
.query("SELECT COUNT(*) FROM grains", vec![])?
.first()
.and_then(|row| row.i64(0))
.unwrap_or(0) as usize)
}
pub fn open_forks(&mut self) -> Result<Vec<ForkGroup>> {
let groups: Vec<(i64, i64, i64)> = self
.db
.query(
"SELECT ns, s, p FROM heads GROUP BY ns, s, p HAVING COUNT(*) > 1",
vec![],
)?
.iter()
.filter_map(|row| match (row.i64(0), row.i64(1), row.i64(2)) {
(Some(ns), Some(s), Some(p)) => Some((ns, s, p)),
_ => None,
})
.collect();
let mut forks = Vec::new();
for (ns_id, s_id, p_id) in groups {
let (Some(namespace), Some(subject), Some(relation)) =
(self.term_str(ns_id)?, self.term_str(s_id)?, self.term_str(p_id)?)
else {
continue;
};
let heads = self
.heads(&namespace, &subject, &relation)?
.into_iter()
.map(|(h, _)| h)
.collect();
forks.push(ForkGroup {
namespace,
subject,
relation,
heads,
});
}
if !self.anon.is_idle() {
for f in forks.iter_mut() {
let mut s = vec![std::mem::take(&mut f.subject)];
let ns = f.namespace.clone();
self.egress_strings_exit(&ns, Some("person"), &mut s)?;
f.subject = s.pop().unwrap_or_default();
}
}
Ok(forks)
}
pub fn heads(&mut self, ns: &str, subject: &str, relation: &str) -> Result<Vec<(Hash, i64)>> {
require_exact_ns("heads", ns)?;
let (Some(ns_id), Some(s_id), Some(p_id)) = (
self.term_lookup(ns)?,
self.term_lookup(subject)?,
self.term_lookup(relation)?,
) else {
return Ok(Vec::new());
};
let mut out = Vec::new();
for row in self.db.query(
"SELECT hash, created_at FROM heads WHERE ns=?1 AND s=?2 AND p=?3
ORDER BY created_at DESC, hash DESC",
vec![pi(ns_id), pi(s_id), pi(p_id)],
)? {
let h = row.blob(0).unwrap_or_default();
let c = row.i64(1).unwrap_or(0);
if let Ok(h) = Hash::try_from_bytes(&h) {
out.push((h, c));
}
}
Ok(out)
}
pub fn merge_heads<G: Grain + 'static>(
&mut self,
ns: &str,
subject: &str,
relation: &str,
merged: &mut G,
) -> Result<Hash> {
self.check_writable("merge heads")?;
let tips = self.heads(ns, subject, relation)?;
if tips.len() < 2 {
return Err(AreevError::Validation(format!(
"merge_heads needs an open fork; {} head(s) present",
tips.len()
)));
}
merged.common_mut().derived_from = Some(tips[0].0.to_hex());
let parents: Vec<String> = tips.iter().map(|(h, _)| h.to_hex()).collect();
let mut ctx = match merged.common().context.clone() {
Some(serde_json::Value::Object(m)) => m,
_ => serde_json::Map::new(),
};
ctx.insert("merge_parents".into(), serde_json::json!(parents));
merged.common_mut().context = Some(serde_json::Value::Object(ctx));
let merged_dyn: &dyn AddableDyn = &*merged;
let (preps, first_seq, first_op, hlc0) = self.prep_and_reserve(&[merged_dyn])?;
let merge_hash = preps[0].hash;
let now = now_ms();
let ram_op = self.next_op;
self.next_op += 1;
let ram_hlc = self.next_hlc();
let (d_docs, d_len) = fts_delta(&preps);
let dbr = self.db.as_ref();
with_txn(dbr, || {
let (s0, o0, h0, op_seq, hlc) = match dbr.reserve_write(1, 2, 2, now_ms() << 16, 0)? {
Some(w) => (w.seq0, w.op0, w.hlc0, w.op0 + 1, w.hlc0 + 1),
None => (first_seq, first_op, hlc0, ram_op, ram_hlc),
};
insert_prepped(dbr, &preps, s0, o0, h0)?; for (tip, _) in &tips {
let seq = dbr
.query(
"SELECT seq, svt FROM grains WHERE hash=?1",
vec![pb(tip.as_bytes().to_vec())],
)?
.first()
.and_then(|row| {
let seq = row.i64(0).unwrap_or(0);
row.i64(1).is_none().then_some(seq)
});
if let Some(seq) = seq {
dbr.execute(
"UPDATE grains SET superseded_by=?1, svt=?2 WHERE seq=?3",
vec![pb(merge_hash.as_bytes().to_vec()), pi(now), pi(seq)],
)?;
dbr.execute("UPDATE triples SET cur=0 WHERE seq=?1", vec![pi(seq)])?;
dbr.execute("UPDATE osp SET cur=0 WHERE seq=?1", vec![pi(seq)])?;
}
}
dbr.execute(
"INSERT INTO oplog(op_seq,hlc,op,hash) VALUES (?1,?2,?3,?4)",
vec![pi(op_seq), pi(hlc), pi(OP_SUPERSEDE), pb(merge_hash.as_bytes().to_vec())],
)?;
Ok(())
})?;
self.fts_docs += d_docs;
self.fts_total_len += d_len;
Ok(merge_hash)
}
pub fn history(&mut self, ns: &str, subject: &str, relation: &str) -> Result<Vec<HistoryEntry>> {
let head = match self.latest(ns, subject, relation)? {
Some(g) => g.hash,
None => return Ok(Vec::new()),
};
let mut out = Vec::new();
let mut cur = Some(head);
while let Some(h) = cur {
let rows = self.db.query(
"SELECT blob, superseded_by, supersedes FROM grains WHERE hash = ?1",
vec![pb(h.as_bytes().to_vec())],
)?;
let (blob, sup_by, supersedes) = match rows.first() {
Some(row) => (row.blob(0), row.blob(1), row.blob(2)),
None => break,
};
if let Some(b) = blob {
let g = deserialize_blob(&b)?;
out.push(HistoryEntry {
hash: h,
object: g.get_str("object").unwrap_or_default().to_string(),
created_at: g.get_i64("created_at").unwrap_or(0),
confidence: g.get_f64("confidence").unwrap_or(0.0),
superseded_by: sup_by.and_then(|b| Hash::try_from_bytes(&b).ok()),
});
}
cur = supersedes.and_then(|b| Hash::try_from_bytes(&b).ok());
if out.len() > 512 {
break; }
}
if !self.anon.is_idle() {
let mut objects: Vec<String> = out.iter().map(|e| e.object.clone()).collect();
self.egress_strings_exit(ns, None, &mut objects)?;
for (e, o) in out.iter_mut().zip(objects) {
e.object = o;
}
}
Ok(out)
}
pub fn verify(&mut self) -> Result<VerifyReport> {
let mut real: Vec<String> = Vec::new();
let mut fts_notes: Vec<String> = Vec::new();
for row in self.db.query("PRAGMA integrity_check", vec![])? {
if let Some(s) = row.text(0) {
if s == "ok" {
continue;
} else if s.contains("__turso_internal_fts") {
fts_notes.push(s.to_string());
} else {
real.push(s.to_string());
}
}
}
let integrity = if real.is_empty() { "ok".to_string() } else { real.join("; ") };
let rows: Vec<(Vec<u8>, Vec<u8>)> = self
.db
.query("SELECT hash, blob FROM grains", vec![])?
.iter()
.map(|row| (row.blob(0).unwrap_or_default(), row.blob(1).unwrap_or_default()))
.collect();
let mut report = VerifyReport {
integrity,
fts_notes,
grains: rows.len(),
hash_mismatches: 0,
undecodable: 0,
};
for (stored, blob) in rows {
match deserialize_blob(&blob) {
Ok(g) => {
if g.hash.as_bytes().as_slice() != stored.as_slice() {
report.hash_mismatches += 1;
}
}
Err(_) => report.undecodable += 1,
}
}
Ok(report)
}
pub fn record_corpus_export(
&mut self,
selector: &str,
destination: &str,
recipient: Option<&str>,
exported_at_ms: i64,
subject_fingerprints: &[String],
source_hashes: &[String],
) -> Result<Hash> {
use sha2::{Digest, Sha256};
let mut fingerprints = subject_fingerprints.to_vec();
fingerprints.sort();
fingerprints.dedup();
let mut sources = source_hashes.to_vec();
sources.sort();
sources.dedup();
let mut id_material = selector.as_bytes().to_vec();
id_material.extend_from_slice(destination.as_bytes());
if let Some(recipient) = recipient {
id_material.extend_from_slice(recipient.as_bytes());
}
id_material.extend_from_slice(&exported_at_ms.to_le_bytes());
for hash in &sources {
id_material.extend_from_slice(hash.as_bytes());
}
let export_id = format!("export:{}", hex::encode(&Sha256::digest(id_material)[..12]));
let selector_digest = hex::encode(Sha256::digest(selector.as_bytes()));
let mut observation = Observation::new("areev:corpus", "corpus_export")
.subject(&export_id)
.object("openai_chat_jsonl")
.created_at(exported_at_ms);
observation.common.namespace = Some(authz::HARNESS_NS.to_string());
observation.common.context = Some(serde_json::json!({
"corpus_export": true,
"selector_sha256": selector_digest,
"selector_kind": "cal_read",
"destination": destination,
"recipient": recipient,
"exported_at_ms": exported_at_ms,
"subject_fingerprints": fingerprints,
"source_count": sources.len(),
"registry_status": "active",
}));
observation.common.related_to = sources
.iter()
.map(|hash| RelatedTo {
hash: hash.clone(),
relation_type: "mg:corpus_source".to_string(),
weight: None,
})
.collect();
self.add(&observation)
}
pub fn corpus_exports(&mut self) -> Result<Vec<CorpusExportRegistry>> {
let mut exports = Vec::new();
let rows = self.db.query(
"SELECT g.blob FROM corpus_idx ci JOIN grains g ON g.seq=ci.seq ORDER BY g.seq DESC",
vec![],
)?;
for blob in rows.iter().filter_map(|row| row.blob(0)) {
let grain = deserialize_blob(&blob)?;
let Some(context) = grain.fields.get("context").and_then(|v| v.as_object()) else {
continue;
};
if context.get("corpus_export").and_then(|v| v.as_bool()) != Some(true) {
continue;
}
let export_id = grain.get_str("subject").unwrap_or_default().to_string();
let destination = context
.get("destination")
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string();
let recipient = context
.get("recipient")
.and_then(|v| v.as_str())
.map(str::to_string);
let exported_at_ms = context
.get("exported_at_ms")
.and_then(|v| v.as_i64())
.unwrap_or_else(|| grain.get_i64("created_at").unwrap_or(0));
let mut subject_fingerprints: Vec<String> = context
.get("subject_fingerprints")
.and_then(|v| v.as_array())
.into_iter()
.flatten()
.filter_map(|v| v.as_str().map(str::to_string))
.collect();
subject_fingerprints.sort();
subject_fingerprints.dedup();
let mut source_hashes: Vec<String> = grain
.fields
.get("related_to")
.and_then(|v| v.as_array())
.into_iter()
.flatten()
.filter(|link| {
link.get("relation_type").and_then(|v| v.as_str())
== Some("mg:corpus_source")
})
.filter_map(|link| link.get("hash").and_then(|v| v.as_str()).map(str::to_string))
.collect();
source_hashes.sort();
source_hashes.dedup();
exports.push(CorpusExportRegistry {
manifest_hash: grain.hash.to_hex(),
export_id,
destination,
recipient,
exported_at_ms,
subject_fingerprints,
source_hashes,
});
}
exports.sort_by(|a, b| {
b.exported_at_ms
.cmp(&a.exported_at_ms)
.then_with(|| a.manifest_hash.cmp(&b.manifest_hash))
});
Ok(exports)
}
pub fn corpus_exports_touching_subject(
&mut self,
subject: &str,
) -> Result<Vec<CorpusExportRegistry>> {
let fingerprint = authz::subject_fingerprint(subject);
Ok(self
.corpus_exports()?
.into_iter()
.filter(|entry| entry.subject_fingerprints.contains(&fingerprint))
.collect())
}
pub fn corpus_exports_touching_hashes(
&mut self,
hashes: &[String],
) -> Result<Vec<CorpusExportRegistry>> {
Ok(exports_touching_hashes(&self.corpus_exports()?, hashes))
}
pub fn stats(&mut self) -> Result<StoreStats> {
let one = |sql: &'static str| -> Result<i64> {
Ok(self
.db
.query(sql, vec![])?
.first()
.and_then(|row| row.i64(0))
.unwrap_or(0))
};
Ok(StoreStats {
grains: one("SELECT COUNT(*) FROM grains")? as usize,
current: one("SELECT COUNT(*) FROM grains WHERE svt IS NULL")? as usize,
triples: one("SELECT COUNT(*) FROM triples")? as usize,
terms: one("SELECT COUNT(*) FROM terms")? as usize,
ops: one("SELECT COUNT(*) FROM oplog")? as usize,
events_indexed: one("SELECT COUNT(*) FROM thread_idx")? as usize,
})
}
pub fn head_op_seq(&mut self) -> Result<i64> {
let rows = self
.db
.query("SELECT COALESCE(MAX(op_seq), 0) FROM oplog", vec![])?;
Ok(rows.first().and_then(|r| r.i64(0)).unwrap_or(0))
}
pub fn changes_since(&mut self, after_op_seq: i64, limit: usize) -> Result<Vec<OpRecord>> {
let mut out = Vec::new();
for row in self.db.query(
"SELECT op_seq, hlc, op, hash FROM oplog WHERE op_seq > ?1 ORDER BY op_seq LIMIT ?2",
vec![pi(after_op_seq), pi(limit as i64)],
)? {
out.push(OpRecord {
op_seq: row.i64(0).unwrap_or(0),
hlc: row.i64(1).unwrap_or(0),
op: row.i64(2).unwrap_or(0),
hash: Hash::try_from_bytes(&row.blob(3).unwrap_or_default())?,
});
}
Ok(out)
}
}
impl Areev {
fn cas_hex(uri: &str) -> Result<&str> {
let hex = uri
.strip_prefix("cas://sha256:")
.ok_or_else(|| AreevError::Validation(format!("not a cas uri: {uri}")))?;
if hex.len() != 64 || !hex.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err(AreevError::Validation(format!("malformed cas uri: {uri}")));
}
Ok(hex)
}
pub fn put_blob(&mut self, bytes: &[u8]) -> Result<String> {
self.check_writable("put a blob")?;
use sha2::{Digest, Sha256};
let digest = Sha256::digest(bytes);
let hex = hex::encode(digest);
let stored = blobcrypt::write_bytes(self.blob_key.as_deref(), &hex, bytes)?;
match &self.blob_store {
BlobStore::Fs(dir) => {
let path = fs_blob_path(dir, &hex);
if !path.exists() {
std::fs::create_dir_all(path.parent().unwrap()).map_err(db_err)?;
let tmp = path.with_extension("tmp");
std::fs::write(&tmp, &stored).map_err(db_err)?;
std::fs::rename(&tmp, &path).map_err(db_err)?;
}
}
BlobStore::Table => {
self.db.execute(
"INSERT OR IGNORE INTO blobs(hash, body) VALUES (?1, ?2)",
vec![pb(digest.to_vec()), pb(stored)],
)?;
}
}
Ok(format!("cas://sha256:{hex}"))
}
pub fn get_blob(&mut self, uri: &str) -> Result<Vec<u8>> {
use sha2::{Digest, Sha256};
let hex = Self::cas_hex(uri)?;
let bytes = match &self.blob_store {
BlobStore::Fs(dir) => std::fs::read(fs_blob_path(dir, hex))
.map_err(|_| AreevError::Storage(format!("blob missing: {uri}")))?,
BlobStore::Table => {
let raw = hex::decode(hex).map_err(db_err)?;
self.db
.query("SELECT body FROM blobs WHERE hash = ?1", vec![pb(raw)])?
.first()
.and_then(|row| row.blob(0))
.ok_or_else(|| AreevError::Storage(format!("blob missing: {uri}")))?
}
};
let bytes = blobcrypt::read_maybe_sealed(self.blob_key.as_deref(), hex, bytes)?;
if hex::encode(Sha256::digest(&bytes)) != hex {
return Err(AreevError::Storage(format!("blob corrupt: {uri}")));
}
Ok(bytes)
}
pub fn encrypt_blobs(&mut self) -> Result<(usize, usize)> {
self.check_writable("encrypt blobs")?;
let Some(key) = self.blob_key.as_deref().copied() else {
return Err(AreevError::Validation(
"encrypt_blobs needs an encrypted memory — open it with its key \
(open_encrypted / --passphrase-env) first"
.into(),
));
};
let mut encrypted = 0usize;
let mut already = 0usize;
match &self.blob_store {
BlobStore::Fs(dir) => {
let dir = dir.clone();
let Ok(shards) = std::fs::read_dir(&dir) else { return Ok((0, 0)) };
for shard in shards.flatten() {
let prefix = shard.file_name().to_string_lossy().to_string();
let Ok(files) = std::fs::read_dir(shard.path()) else { continue };
for f in files.flatten() {
let path = f.path();
let rest = f.file_name().to_string_lossy().to_string();
let hex = format!("{prefix}{rest}");
if hex.len() != 64 || !hex.bytes().all(|b| b.is_ascii_hexdigit()) {
continue;
}
let raw = std::fs::read(&path).map_err(db_err)?;
if blobcrypt::is_sealed(&raw) {
already += 1;
continue;
}
let sealed = blobcrypt::seal(&key, &hex, &raw)?;
let tmp = path.with_extension("tmp");
std::fs::write(&tmp, &sealed).map_err(db_err)?;
std::fs::rename(&tmp, &path).map_err(db_err)?;
encrypted += 1;
}
}
}
BlobStore::Table => {
return Err(AreevError::Validation(
"blob encryption is file-backend-only; on postgres the blobs table is \
covered by the database's own encryption at rest"
.into(),
));
}
}
Ok((encrypted, already))
}
pub fn gc_blobs(&mut self) -> Result<usize> {
self.check_writable("garbage-collect blobs")?;
let blobs: Vec<Vec<u8>> = self
.db
.query("SELECT blob FROM grains", vec![])?
.iter()
.filter_map(|row| row.blob(0))
.collect();
let mut referenced: HashSet<String> = HashSet::new();
for b in &blobs {
if let Ok(view) = deserialize_blob(b) {
referenced.extend(cas_hexes(&view));
}
}
let mut removed = 0usize;
match &self.blob_store {
BlobStore::Fs(dir) => {
if let Ok(shards) = std::fs::read_dir(dir) {
for shard in shards.flatten() {
let prefix = shard.file_name().to_string_lossy().to_string();
if let Ok(files) = std::fs::read_dir(shard.path()) {
for f in files.flatten() {
let rest = f.file_name().to_string_lossy().to_string();
let hex = format!("{prefix}{rest}");
if !referenced.contains(&hex)
&& std::fs::remove_file(f.path()).is_ok()
{
removed += 1;
}
}
}
}
}
}
BlobStore::Table => {
let stored: Vec<Vec<u8>> = self
.db
.query("SELECT hash FROM blobs", vec![])?
.iter()
.filter_map(|row| row.blob(0))
.collect();
for raw in stored {
if !referenced.contains(&hex::encode(&raw)) {
self.db.execute("DELETE FROM blobs WHERE hash = ?1", vec![pb(raw)])?;
removed += 1;
}
}
}
}
Ok(removed)
}
fn replicable_meta_rows(&self) -> Result<Vec<(String, String)>> {
let mut rows = Vec::new();
for prefix in REPLICABLE_META_PREFIXES {
for (suffix, value) in self.meta_scan(prefix)? {
let export_value = match serde_json::from_str::<serde_json::Value>(&value) {
Ok(mut v) => {
if let Some(obj) = v.as_object_mut() {
obj.remove("last_run_at");
}
v.to_string()
}
Err(_) => value,
};
rows.push((format!("{prefix}{suffix}"), export_value));
}
}
rows.sort();
Ok(rows)
}
pub fn bundle_since(&mut self, after_op_seq: i64, path: &str) -> Result<BundleStats> {
let ops = self.changes_since(after_op_seq, usize::MAX / 2)?;
let meta_rows = self.replicable_meta_rows()?;
let mut out: Vec<u8> = Vec::with_capacity(64 * 1024);
if meta_rows.is_empty() {
out.extend_from_slice(BUNDLE_MAGIC);
} else {
out.extend_from_slice(BUNDLE_MAGIC_V2);
let meta_json = serde_json::to_string(
&meta_rows
.iter()
.map(|(k, v)| (k.as_str(), v.as_str()))
.collect::<std::collections::BTreeMap<_, _>>(),
)
.map_err(|e| AreevError::Format(format!("bundle meta segment: {e}")))?;
out.extend_from_slice(&(meta_json.len() as u32).to_le_bytes());
out.extend_from_slice(meta_json.as_bytes());
}
let mut last = after_op_seq;
for rec in &ops {
let blob: Option<Vec<u8>> = if rec.op == OP_FORGET {
None
} else {
self.db
.query_hot(
"SELECT blob FROM grains WHERE hash = ?1",
vec![pb(rec.hash.as_bytes().to_vec())],
)?
.first()
.and_then(|row| row.blob(0))
};
out.push(rec.op as u8);
out.extend_from_slice(&rec.hlc.to_le_bytes());
out.extend_from_slice(rec.hash.as_bytes());
let b = blob.unwrap_or_default();
out.extend_from_slice(&(b.len() as u32).to_le_bytes());
out.extend_from_slice(&b);
last = rec.op_seq;
}
std::fs::write(path, &out).map_err(db_err)?;
Ok(BundleStats {
ops: ops.len(),
bytes: std::fs::metadata(path).map(|m| m.len()).unwrap_or(0),
last_op_seq: last,
})
}
fn blob_by_hash(&mut self, hash: &Hash) -> Result<Option<Vec<u8>>> {
Ok(self
.db
.query_hot(
"SELECT blob FROM grains WHERE hash = ?1",
vec![pb(hash.as_bytes().to_vec())],
)?
.first()
.and_then(|row| row.blob(0)))
}
fn has_grain(&mut self, hash: &Hash) -> Result<bool> {
Ok(!self
.db
.query_hot(
"SELECT 1 FROM grains WHERE hash = ?1",
vec![pb(hash.as_bytes().to_vec())],
)?
.is_empty())
}
fn insert_blob(&mut self, blob: Vec<u8>, hash: Hash, op: i64, hlc_in: i64) -> Result<()> {
self.check_writable("import a bundle")?;
let pr = self.prep_from_blob(blob, hash, false)?;
let (d_docs, d_len) = fts_delta(std::slice::from_ref(&pr));
let ram_seq = self.next_seq;
self.next_seq += 1;
let ram_op = self.next_op;
self.next_op += 1;
self.hlc_last = self.hlc_last.max(hlc_in);
let dbr = self.db.as_ref();
with_txn(dbr, || {
let (seq, op_seq) = match dbr.reserve_write(1, 1, 0, now_ms() << 16, hlc_in)? {
Some(w) => (w.seq0, w.op0),
None => (ram_seq, ram_op),
};
bump_ns_registry(dbr, pr.ns_id, 1)?;
dbr.execute(
"INSERT INTO grains(seq,hash,ns,gtype,created_at,s,p,o,vf,vt,svf,svt,superseded_by,supersedes,text,blob)
VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,NULL,NULL,NULL,?12,?13)",
vec![
pi(seq),
pb(pr.hash.as_bytes().to_vec()),
pi(pr.ns_id),
pi(pr.gtype),
pi(pr.created),
opt_i(pr.s),
opt_i(pr.p),
opt_i(pr.o),
opt_i(pr.vf),
opt_i(pr.vt),
pi(pr.created),
match &pr.text { Some(t) => pt(t), None => Value::Null },
pb(pr.blob.clone()),
],
)?;
if let (Some(s), Some(p), Some(o)) = (pr.s, pr.p, pr.o) {
dbr.execute(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(s), pi(p), pi(o), pi(seq)],
)?;
if pr.osp {
dbr.execute(
"INSERT INTO osp(ns,o,s,p,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(o), pi(s), pi(p), pi(seq)],
)?;
}
dbr.execute(
"INSERT OR REPLACE INTO heads(ns,s,p,seq,hash,created_at) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(pr.ns_id), pi(s), pi(p), pi(seq), pb(pr.hash.as_bytes().to_vec()), pi(pr.created)],
)?;
let cur = dbr
.query(
"SELECT h.created_at, h.hash FROM heads h JOIN entity_latest e
ON e.ns=h.ns AND e.s=h.s AND e.p=h.p AND e.seq=h.seq
WHERE e.ns=?1 AND e.s=?2 AND e.p=?3",
vec![pi(pr.ns_id), pi(s), pi(p)],
)?
.first()
.map(|row| (row.i64(0).unwrap_or(0), row.blob(1).unwrap_or_default()));
let wins = match &cur {
Some((c, h)) => (pr.created, pr.hash.as_bytes().as_slice()) > (*c, h.as_slice()),
None => true,
};
if wins {
dbr.execute(
"INSERT OR REPLACE INTO entity_latest(ns,s,p,o,seq,hash) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(pr.ns_id), pi(s), pi(p), pi(o), pi(seq), pb(pr.hash.as_bytes().to_vec())],
)?;
}
}
if let Some(s) = pr.subject_only {
dbr.execute(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,NULL,NULL,?3,1)",
vec![pi(pr.ns_id), pi(s), pi(seq)],
)?;
}
for (ls, lp, lo) in &pr.links {
dbr.execute(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(*ls), pi(*lp), pi(*lo), pi(seq)],
)?;
dbr.execute(
"INSERT INTO osp(ns,o,s,p,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(*lo), pi(*ls), pi(*lp), pi(seq)],
)?;
}
if let Some(sess) = pr.session {
dbr.execute(
"INSERT INTO thread_idx(ns,session,seq) VALUES (?1,?2,?3)",
vec![pi(pr.ns_id), pi(sess), pi(seq)],
)?;
}
if let Some(run) = pr.run {
dbr.execute(
"INSERT INTO run_idx(ns,run,seq) VALUES (?1,?2,?3)",
vec![pi(pr.ns_id), pi(run), pi(seq)],
)?;
}
if pr.corpus_export {
dbr.execute("INSERT INTO corpus_idx(seq) VALUES (?1)", vec![pi(seq)])?;
}
if let Some(ref parent) = pr.parent {
dbr.execute(
"INSERT INTO prov_idx(ns,parent,seq) VALUES (?1,?2,?3)",
vec![pi(pr.ns_id), pb(parent.clone()), pi(seq)],
)?;
}
if let Some(ref emb) = pr.embedding {
dbr.execute(
"INSERT INTO embeddings(seq, vec) VALUES (?1, vector32(?2))",
vec![pi(seq), pt(&vec_to_json(emb))],
)?;
}
if !pr.tokens.is_empty() {
for (term, tf) in &pr.tokens {
dbr.execute_hot(
"INSERT INTO fts_post(term,seq,ns,tf) VALUES (?1,?2,?3,?4)",
vec![pi(*term), pi(seq), pi(pr.ns_id), pi(*tf)],
)?;
}
dbr.execute_hot(
"INSERT OR REPLACE INTO fts_doc(seq,len) VALUES (?1,?2)",
vec![pi(seq), pi(pr.doc_len)],
)?;
}
dbr.execute(
"INSERT INTO oplog(op_seq,hlc,op,hash) VALUES (?1,?2,?3,?4)",
vec![pi(op_seq), pi(hlc_in), pi(op), pb(pr.hash.as_bytes().to_vec())],
)?;
Ok(())
})?;
self.fts_docs += d_docs;
self.fts_total_len += d_len;
Ok(())
}
fn apply_supersede_flip(&mut self, old: &Hash, new_hash: &Hash) -> Result<bool> {
let dbr = self.db.as_ref();
with_txn(dbr, || {
dbr.reserve_write(0, 0, 0, now_ms() << 16, 0)?;
let rows = dbr.query(
&format!(
"SELECT seq, svt, ns, s, p FROM grains WHERE hash = ?1{}",
dbr.for_update()
),
vec![pb(old.as_bytes().to_vec())],
)?;
let (old_seq, old_svt, old_ns, old_s, old_p) = match rows.first() {
Some(row) => (
row.i64(0).unwrap_or(0),
row.i64(1),
row.i64(2).unwrap_or(0),
row.i64(3),
row.i64(4),
),
None => return Ok(false), };
if old_svt.is_some() {
let existing = dbr
.query(
"SELECT superseded_by FROM grains WHERE seq=?1",
vec![pi(old_seq)],
)?
.first()
.and_then(|row| row.blob(0));
if existing.as_deref() == Some(new_hash.as_bytes().as_slice()) {
return Ok(false); }
let inc = dbr
.query(
"SELECT seq, ns, s, p, o, created_at FROM grains WHERE hash=?1",
vec![pb(new_hash.as_bytes().to_vec())],
)?
.first()
.map(|row| {
(
row.i64(0).unwrap_or(0),
row.i64(1).unwrap_or(0),
row.i64(2).unwrap_or(0),
row.i64(3).unwrap_or(0),
row.i64(4).unwrap_or(0),
row.i64(5).unwrap_or(0),
)
});
let Some((inc_seq, ns, s, p, o, inc_created)) = inc else { return Ok(false) };
dbr.execute(
"INSERT OR REPLACE INTO heads(ns,s,p,seq,hash,created_at) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(ns), pi(s), pi(p), pi(inc_seq), pb(new_hash.as_bytes().to_vec()), pi(inc_created)],
)?;
let cur = dbr
.query(
"SELECT h.created_at, h.hash FROM heads h JOIN entity_latest e
ON e.ns=h.ns AND e.s=h.s AND e.p=h.p AND e.seq=h.seq
WHERE e.ns=?1 AND e.s=?2 AND e.p=?3",
vec![pi(ns), pi(s), pi(p)],
)?
.first()
.map(|row| (row.i64(0).unwrap_or(0), row.blob(1).unwrap_or_default()));
let incoming_wins = match &cur {
Some((c_created, c_hash)) => {
(inc_created, new_hash.as_bytes().as_slice()) > (*c_created, c_hash.as_slice())
}
None => true,
};
if incoming_wins {
dbr.execute(
"INSERT OR REPLACE INTO entity_latest(ns,s,p,o,seq,hash) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(ns), pi(s), pi(p), pi(o), pi(inc_seq), pb(new_hash.as_bytes().to_vec())],
)?;
}
return Ok(true); }
let now = now_ms();
dbr.execute(
"UPDATE grains SET superseded_by=?1, svt=?2 WHERE seq=?3",
vec![pb(new_hash.as_bytes().to_vec()), pi(now), pi(old_seq)],
)?;
dbr.execute(
"UPDATE grains SET supersedes=?1 WHERE hash=?2",
vec![pb(old.as_bytes().to_vec()), pb(new_hash.as_bytes().to_vec())],
)?;
dbr.execute("UPDATE triples SET cur=0 WHERE seq=?1", vec![pi(old_seq)])?;
dbr.execute("UPDATE osp SET cur=0 WHERE seq=?1", vec![pi(old_seq)])?;
if let (Some(s), Some(p)) = (old_s, old_p) {
dbr.execute(
"DELETE FROM heads WHERE ns=?1 AND s=?2 AND p=?3 AND seq=?4",
vec![pi(old_ns), pi(s), pi(p), pi(old_seq)],
)?;
}
Ok(true)
})
}
pub fn import_bundle(&mut self, path: &str) -> Result<ImportStats> {
self.import_bundle_until(path, None)
}
pub fn import_bundle_until(&mut self, path: &str, max_hlc: Option<i64>) -> Result<ImportStats> {
let data = std::fs::read(path).map_err(db_err)?;
if data.len() < 4 {
return Err(AreevError::Format("not a MGB1/MGB2 bundle".into()));
}
let mut stats = ImportStats::default();
let mut meta_entries: Vec<(String, String)> = Vec::new();
let mut i = 4usize;
match &data[..4] {
m if m == BUNDLE_MAGIC => {}
m if m == BUNDLE_MAGIC_V2 => {
if data.len() < 8 {
return Err(AreevError::Format("truncated bundle meta segment".into()));
}
let meta_len = u32::from_le_bytes(data[4..8].try_into().unwrap()) as usize;
i = 8;
if i.checked_add(meta_len).is_none_or(|end| end > data.len()) {
return Err(AreevError::Format("truncated bundle meta segment".into()));
}
let meta_json: std::collections::BTreeMap<String, String> =
serde_json::from_slice(&data[i..i + meta_len]).map_err(|e| {
AreevError::Format(format!("malformed bundle meta segment: {e}"))
})?;
meta_entries = meta_json.into_iter().collect();
i += meta_len;
}
_ => return Err(AreevError::Format("not a MGB1/MGB2 bundle".into())),
}
while i < data.len() {
if i + 1 + 8 + 32 + 4 > data.len() {
return Err(AreevError::Format("truncated bundle record".into()));
}
let op = data[i] as i64;
i += 1;
let hlc = i64::from_le_bytes(data[i..i + 8].try_into().unwrap());
i += 8;
let hash = Hash::try_from_bytes(&data[i..i + 32])?;
i += 32;
let len = u32::from_le_bytes(data[i..i + 4].try_into().unwrap()) as usize;
i += 4;
if i.checked_add(len).is_none_or(|end| end > data.len()) {
return Err(AreevError::Format("truncated bundle blob".into()));
}
let blob = data[i..i + len].to_vec();
i += len;
if let Some(t) = max_hlc {
if hlc > t {
stats.skipped += 1;
continue; }
}
match op {
OP_ADD => {
if self.has_grain(&hash)? || blob.is_empty() {
stats.skipped += 1;
continue;
}
self.insert_blob(blob, hash, op, hlc)?;
stats.applied += 1;
}
OP_SUPERSEDE => {
let exists = self.has_grain(&hash)?;
let bytes: Option<Vec<u8>> = if !blob.is_empty() {
Some(blob)
} else if exists {
self.blob_by_hash(&hash)?
} else {
None
};
match bytes {
None => stats.skipped += 1,
Some(bb) => {
let mut changed = false;
let inserted = !exists;
if !exists {
self.insert_blob(bb.clone(), hash, op, hlc)?;
changed = true;
}
if let Ok(view) = deserialize_blob(&bb) {
for old in superseded_parents(&view) {
changed |= self.apply_supersede_flip(&old, &hash)?;
}
}
if changed && !inserted {
self.log_op(OP_SUPERSEDE, &hash, hlc)?;
}
if changed {
stats.applied += 1;
} else {
stats.skipped += 1;
}
}
}
}
OP_FORGET => match self.forget(&hash) {
Ok(()) => stats.applied += 1,
Err(AreevError::NotFound(_)) => stats.skipped += 1,
Err(e) => return Err(e),
},
_ => return Err(AreevError::Format(format!("unknown bundle op {op}"))),
}
}
if max_hlc.is_none() {
let mut anon_changed = false;
for (key, value) in &meta_entries {
if self.apply_bundle_meta_row(key, value)? {
stats.meta_applied += 1;
anon_changed |= key.starts_with(anon_gate::ANON_PREFIX);
} else {
stats.meta_skipped += 1;
}
}
if anon_changed {
self.reload_anon_policies()?;
}
} else {
stats.meta_skipped += meta_entries.len();
}
Ok(stats)
}
fn apply_bundle_meta_row(&mut self, key: &str, incoming: &str) -> Result<bool> {
if !REPLICABLE_META_PREFIXES.iter().any(|p| key.starts_with(p)) {
return Ok(false);
}
let local = self.meta_get(key)?;
let registry_row = key.starts_with("qry:") || key.starts_with("tpl:");
match local {
None => {
self.meta_put(key, incoming)?;
Ok(true)
}
Some(_) if !registry_row => Ok(false),
Some(local_raw) => {
let updated_at = |raw: &str| {
serde_json::from_str::<serde_json::Value>(raw)
.ok()
.and_then(|v| v.get("updated_at").and_then(serde_json::Value::as_u64))
.unwrap_or(0)
};
let mut incoming_json = match serde_json::from_str::<serde_json::Value>(incoming) {
Ok(v) => v,
Err(_) => return Ok(false),
};
if updated_at(incoming) <= updated_at(&local_raw) {
return Ok(false);
}
let local_last_run = serde_json::from_str::<serde_json::Value>(&local_raw)
.ok()
.and_then(|v| v.get("last_run_at").cloned())
.filter(|v| !v.is_null());
if let (Some(obj), Some(lr)) = (incoming_json.as_object_mut(), local_last_run) {
obj.insert("last_run_at".into(), lr);
}
self.meta_put(key, &incoming_json.to_string())?;
Ok(true)
}
}
}
}
impl Drop for Areev {
fn drop(&mut self) {
let _ = self.telemetry_flush();
}
}
mod anon_gate;
mod blobcrypt;
pub mod asyncdb;
pub mod memory_tool;
pub mod migrate;
pub mod telemetry;
pub use asyncdb::AsyncAreev;
pub use telemetry::{
AccessStat, BudgetStat, QueryStat, RecallEvent, RecallLogEntry, Telemetry, TelemetryMode,
};
fn fts_delta(preps: &[GrainPrep]) -> (i64, i64) {
let docs = preps.iter().filter(|p| !p.tokens.is_empty()).count() as i64;
(docs, preps.iter().map(|p| p.doc_len).sum())
}
fn bump_ns_registry(db: &dyn Db, ns_id: i64, k: i64) -> Result<()> {
db.execute_hot("INSERT OR IGNORE INTO ns_reg(ns, n) VALUES (?1, 0)", vec![pi(ns_id)])?;
db.execute_hot("UPDATE ns_reg SET n = n + ?2 WHERE ns = ?1", vec![pi(ns_id), pi(k)])?;
Ok(())
}
fn drop_ns_registry(db: &dyn Db, ns_id: i64, k: i64) -> Result<()> {
db.execute_hot("UPDATE ns_reg SET n = n - ?2 WHERE ns = ?1", vec![pi(ns_id), pi(k)])?;
db.execute_hot("DELETE FROM ns_reg WHERE ns = ?1 AND n <= 0", vec![pi(ns_id)])?;
Ok(())
}
fn insert_prepped(
db: &dyn Db,
preps: &[GrainPrep],
first_seq: i64,
first_op: i64,
hlc0: i64,
) -> Result<()> {
let mut ns_counts: Vec<(i64, i64)> = Vec::new();
for pr in preps {
match ns_counts.iter_mut().find(|(ns, _)| *ns == pr.ns_id) {
Some((_, k)) => *k += 1,
None => ns_counts.push((pr.ns_id, 1)),
}
}
for (ns, k) in &ns_counts {
bump_ns_registry(db, *ns, *k)?;
}
for (i, pr) in preps.iter().enumerate() {
let seq = first_seq + i as i64;
db.execute_hot(
"INSERT INTO grains(seq,hash,ns,gtype,created_at,s,p,o,vf,vt,svf,svt,superseded_by,supersedes,text,blob)
VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,NULL,NULL,NULL,?12,?13)",
vec![
pi(seq),
pb(pr.hash.as_bytes().to_vec()),
pi(pr.ns_id),
pi(pr.gtype),
pi(pr.created),
opt_i(pr.s),
opt_i(pr.p),
opt_i(pr.o),
opt_i(pr.vf),
opt_i(pr.vt),
pi(pr.created),
match &pr.text { Some(t) => pt(t), None => Value::Null },
pb(pr.blob.clone()),
],
)?;
if !pr.tokens.is_empty() {
for (term, tf) in &pr.tokens {
db.execute_hot(
"INSERT INTO fts_post(term,seq,ns,tf) VALUES (?1,?2,?3,?4)",
vec![pi(*term), pi(seq), pi(pr.ns_id), pi(*tf)],
)?;
}
db.execute_hot(
"INSERT OR REPLACE INTO fts_doc(seq,len) VALUES (?1,?2)",
vec![pi(seq), pi(pr.doc_len)],
)?;
}
if let (Some(s), Some(p), Some(o)) = (pr.s, pr.p, pr.o) {
db.execute_hot(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(s), pi(p), pi(o), pi(seq)],
)?;
if pr.osp {
db.execute_hot(
"INSERT INTO osp(ns,o,s,p,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(o), pi(s), pi(p), pi(seq)],
)?;
}
db.execute_hot(
"INSERT OR REPLACE INTO entity_latest(ns,s,p,o,seq,hash) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(pr.ns_id), pi(s), pi(p), pi(o), pi(seq), pb(pr.hash.as_bytes().to_vec())],
)?;
db.execute_hot(
"DELETE FROM heads WHERE ns=?1 AND s=?2 AND p=?3",
vec![pi(pr.ns_id), pi(s), pi(p)],
)?;
db.execute_hot(
"INSERT INTO heads(ns,s,p,seq,hash,created_at) VALUES (?1,?2,?3,?4,?5,?6)",
vec![pi(pr.ns_id), pi(s), pi(p), pi(seq), pb(pr.hash.as_bytes().to_vec()), pi(pr.created)],
)?;
}
if let Some(s) = pr.subject_only {
db.execute_hot(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,NULL,NULL,?3,1)",
vec![pi(pr.ns_id), pi(s), pi(seq)],
)?;
}
for (ls, lp, lo) in &pr.links {
db.execute_hot(
"INSERT INTO triples(ns,s,p,o,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(*ls), pi(*lp), pi(*lo), pi(seq)],
)?;
db.execute_hot(
"INSERT INTO osp(ns,o,s,p,seq,cur) VALUES (?1,?2,?3,?4,?5,1)",
vec![pi(pr.ns_id), pi(*lo), pi(*ls), pi(*lp), pi(seq)],
)?;
}
if let Some(sess) = pr.session {
db.execute_hot(
"INSERT INTO thread_idx(ns,session,seq) VALUES (?1,?2,?3)",
vec![pi(pr.ns_id), pi(sess), pi(seq)],
)?;
}
if let Some(run) = pr.run {
db.execute_hot(
"INSERT INTO run_idx(ns,run,seq) VALUES (?1,?2,?3)",
vec![pi(pr.ns_id), pi(run), pi(seq)],
)?;
}
if pr.corpus_export {
db.execute_hot("INSERT INTO corpus_idx(seq) VALUES (?1)", vec![pi(seq)])?;
}
if let Some(ref parent) = pr.parent {
db.execute_hot(
"INSERT INTO prov_idx(ns,parent,seq) VALUES (?1,?2,?3)",
vec![pi(pr.ns_id), pb(parent.clone()), pi(seq)],
)?;
}
if let Some(ref emb) = pr.embedding {
db.execute_hot(
"INSERT INTO embeddings(seq, vec) VALUES (?1, vector32(?2))",
vec![pi(seq), pt(&vec_to_json(emb))],
)?;
}
db.execute_hot(
"INSERT INTO oplog(op_seq,hlc,op,hash) VALUES (?1,?2,?3,?4)",
vec![pi(first_op + i as i64), pi(hlc0 + i as i64), pi(OP_ADD), pb(pr.hash.as_bytes().to_vec())],
)?;
}
Ok(())
}
fn superseded_parents(view: &DeserializedGrain) -> Vec<Hash> {
let mut out = Vec::new();
if let Some(df) = view.get_str("derived_from") {
if let Ok(h) = Hash::from_hex(df) {
out.push(h);
}
}
if let Some(arr) = view
.fields
.get("context")
.and_then(|c| c.get("merge_parents"))
.and_then(|v| v.as_array())
{
for p in arr {
if let Some(h) = p.as_str().and_then(|s| Hash::from_hex(s).ok()) {
if !out.contains(&h) {
out.push(h);
}
}
}
}
out
}
pub trait AddableDyn {
fn serialize_dyn(&self) -> Result<(Vec<u8>, Hash)>;
}
impl<G: Grain + 'static> AddableDyn for G {
fn serialize_dyn(&self) -> Result<(Vec<u8>, Hash)> {
serialize_grain(self)
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
fn replication_replay_accepts_a_namespace_a_local_write_would_refuse() {
use areev_core::format::serialize::serialize_grain;
use areev_core::types::Fact;
let dir = TempDir::new().unwrap();
let mut m = Areev::open(dir.path().join("m.db").to_str().unwrap()).unwrap();
for legacy in ["age, build_messagesnt:harness", "org.*"] {
let mut f = Fact::new("evalset:9409", "mg:eval_run", "{}");
f.common.namespace = Some(legacy.to_string());
f.common.created_at = Some(1_700_000_000_000);
let (blob, hash) = serialize_grain(&f).unwrap();
assert!(
m.add(&f).is_err(),
"{legacy:?} must be unwritable to a local author"
);
m.insert_blob(blob, hash, OP_ADD, 1)
.unwrap_or_else(|e| panic!("replay of {legacy:?} must apply: {e}"));
assert_eq!(
m.recall(legacy, "evalset:9409", None, 8).unwrap().len(),
1,
"{legacy:?} must stay readable"
);
assert!(m.namespaces().unwrap().iter().any(|(n, _)| n == legacy));
}
m.forget_subject("age, build_messagesnt:harness", "evalset:9409")
.expect("erasure must reach a grandfathered unspellable namespace");
let err = m
.forget_subject("org.*", "evalset:9409")
.unwrap_err()
.to_string();
assert!(err.starts_with("VAL-E001"), "{err}");
assert!(err.contains("exact namespace"), "{err}");
}
#[test]
fn meta_rows_round_trip_and_scan_by_prefix() {
let dir = TempDir::new().unwrap();
let m = Areev::open(dir.path().join("m.db").to_str().unwrap()).unwrap();
m.meta_put("qry:brief", "{\"body\":\"a\"}").unwrap();
m.meta_put("qry:wide", "{\"body\":\"b\"}").unwrap();
m.meta_put("tpl:card", "{\"source\":\"c\"}").unwrap();
let mut queries = m.meta_scan("qry:").unwrap();
queries.sort();
assert_eq!(
queries,
vec![
("brief".to_string(), "{\"body\":\"a\"}".to_string()),
("wide".to_string(), "{\"body\":\"b\"}".to_string()),
],
"scan returns this prefix's rows with the prefix stripped"
);
assert_eq!(m.meta_scan("tpl:").unwrap().len(), 1);
m.meta_put("qry:brief", "{\"body\":\"z\"}").unwrap();
assert_eq!(m.meta_scan("qry:").unwrap().len(), 2);
m.meta_delete("qry:brief").unwrap();
assert_eq!(m.meta_scan("qry:").unwrap().len(), 1);
m.meta_delete("qry:brief").unwrap();
}
#[test]
fn meta_scan_treats_like_wildcards_as_literals() {
let dir = TempDir::new().unwrap();
let m = Areev::open(dir.path().join("m.db").to_str().unwrap()).unwrap();
m.meta_put("a_b:mine", "1").unwrap();
m.meta_put("axb:theirs", "2").unwrap();
m.meta_put("pre%fix:mine", "3").unwrap();
m.meta_put("preXfix:theirs", "4").unwrap();
assert_eq!(
m.meta_scan("a_b:").unwrap(),
vec![("mine".to_string(), "1".to_string())],
"`_` must not match an arbitrary character"
);
assert_eq!(
m.meta_scan("pre%fix:").unwrap(),
vec![("mine".to_string(), "3".to_string())],
"`%` must not match an arbitrary run"
);
}
#[test]
fn hex32_roundtrips_to_64_chars() {
assert_eq!(hex32(&[0u8; 32]), "0".repeat(64));
assert_eq!(hex32(&[0xffu8; 32]), "f".repeat(64));
let mut key = [0u8; 32];
for (i, b) in key.iter_mut().enumerate() {
*b = i as u8;
}
let hexed = hex32(&key);
assert_eq!(hexed.len(), 64);
assert_eq!(&hexed[..2], "00");
assert_eq!(&hexed[62..], "1f"); assert_eq!(hex::decode(&hexed).unwrap(), key.to_vec());
}
#[test]
fn seq_csv_formats_integer_lists() {
assert_eq!(seq_csv(&[]), "");
assert_eq!(seq_csv(&[42]), "42");
assert_eq!(seq_csv(&[1, 2, 3]), "1,2,3");
assert_eq!(seq_csv(&[-1, 0, 7]), "-1,0,7");
}
#[test]
fn vec_to_json_renders_float_arrays() {
assert_eq!(vec_to_json(&[]), "[]");
assert_eq!(vec_to_json(&[1.0]), "[1]");
assert_eq!(vec_to_json(&[1.0, 2.5, -3.25]), "[1,2.5,-3.25]");
}
#[test]
fn value_bridge_roundtrips() {
assert_eq!(v_i64(&pi(42)), Some(42));
assert_eq!(v_i64(&pt("x")), None);
assert_eq!(v_blob(&pb(vec![1, 2, 3])), Some(vec![1, 2, 3]));
assert_eq!(v_blob(&pi(1)), None);
assert_eq!(v_i64(&opt_i(Some(7))), Some(7));
assert!(matches!(opt_i(None), Value::Null));
assert!((v_f64(&Value::Real(1.5)).unwrap() - 1.5).abs() < 1e-12);
assert!((v_f64(&pi(3)).unwrap() - 3.0).abs() < 1e-12);
assert_eq!(v_f64(&pt("x")), None);
}
fn kdf_line(salt: &[u8], m: u32, t: u32, p: u32) -> String {
format!("v1 argon2id {} {m} {t} {p}", hex::encode(salt))
}
#[test]
fn parse_kdf_sidecar_accepts_valid() {
let salt = [7u8; KDF_SALT_LEN];
let text = kdf_line(&salt, KDF_M_COST, KDF_T_COST, KDF_P_COST);
let (got_salt, m, t, p) = parse_kdf_sidecar(&text, "x.kdf").unwrap();
assert_eq!(got_salt, salt);
assert_eq!((m, t, p), (KDF_M_COST, KDF_T_COST, KDF_P_COST));
assert!(parse_kdf_sidecar(&format!(" {text}\n"), "x.kdf").is_ok());
let salt_hex = hex::encode([1u8; KDF_SALT_LEN]);
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt_hex} 8 1 1"), "x").is_ok());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt_hex} 1048576 16 16"), "x").is_ok());
}
#[test]
fn parse_kdf_sidecar_rejects_malformed() {
let salt = hex::encode([1u8; KDF_SALT_LEN]);
assert!(parse_kdf_sidecar("v1 argon2id", "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} 1 2 3 4"), "x").is_err());
assert!(parse_kdf_sidecar(&format!("v2 argon2id {salt} 19456 2 1"), "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 scrypt {salt} 19456 2 1"), "x").is_err());
assert!(parse_kdf_sidecar("v1 argon2id zzzz 19456 2 1", "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} m 2 1"), "x").is_err());
}
#[test]
fn parse_kdf_sidecar_rejects_wrong_salt_length() {
let short = hex::encode([1u8; 8]);
assert!(parse_kdf_sidecar(&format!("v1 argon2id {short} 19456 2 1"), "x").is_err());
let long = hex::encode([1u8; 32]);
assert!(parse_kdf_sidecar(&format!("v1 argon2id {long} 19456 2 1"), "x").is_err());
}
#[test]
fn parse_kdf_sidecar_rejects_out_of_range_params() {
let salt = hex::encode([1u8; KDF_SALT_LEN]);
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} 7 2 1"), "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} 1048577 2 1"), "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} 19456 0 1"), "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} 19456 17 1"), "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} 19456 2 0"), "x").is_err());
assert!(parse_kdf_sidecar(&format!("v1 argon2id {salt} 19456 2 17"), "x").is_err());
}
#[test]
fn derive_key_rejects_empty_or_whitespace_passphrase() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("k.db");
let path = path.to_str().unwrap();
assert!(Areev::derive_key_for(path, "").is_err());
assert!(Areev::derive_key_for(path, " ").is_err());
assert!(Areev::derive_key_for(path, "\t\n ").is_err());
assert!(!std::path::Path::new(&format!("{path}.kdf")).exists());
}
#[test]
fn derive_key_is_deterministic_for_same_salt() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("k.db");
let path = path.to_str().unwrap();
let k1 = Areev::derive_key_for(path, "correct horse battery staple").unwrap();
let k2 = Areev::derive_key_for(path, "correct horse battery staple").unwrap();
assert_eq!(*k1, *k2);
let k3 = Areev::derive_key_for(path, "a different passphrase").unwrap();
assert_ne!(*k1, *k3);
}
#[test]
fn derive_key_differs_across_salts() {
let dir = TempDir::new().unwrap();
let p1 = dir.path().join("a.db");
let p2 = dir.path().join("b.db");
let (p1, p2) = (p1.to_str().unwrap(), p2.to_str().unwrap());
let k1 = Areev::derive_key_for(p1, "same-pass").unwrap();
let k2 = Areev::derive_key_for(p2, "same-pass").unwrap();
assert_ne!(*k1, *k2);
assert!(std::path::Path::new(&format!("{p1}.kdf")).exists());
assert!(std::path::Path::new(&format!("{p2}.kdf")).exists());
}
fn rrf_fuse(legs: &[&[i64]]) -> Vec<(i64, f64)> {
let mut scores: HashMap<i64, f64> = HashMap::new();
for leg in legs {
for (rank, seq) in leg.iter().enumerate() {
*scores.entry(*seq).or_insert(0.0) += 1.0 / (RRF_K0 + rank as f64);
}
}
let mut ranked: Vec<(i64, f64)> = scores.into_iter().collect();
ranked.sort_by(|a, b| {
b.1.partial_cmp(&a.1)
.unwrap_or(std::cmp::Ordering::Equal)
.then(b.0.cmp(&a.0))
});
ranked
}
#[test]
fn rrf_k0_constant_is_pinned() {
assert!((RRF_K0 - 60.0).abs() < f64::EPSILON);
}
#[test]
fn rrf_single_leg_preserves_rank_order() {
let leg = [10i64, 20, 30];
let fused = rrf_fuse(&[&leg]);
assert_eq!(fused.iter().map(|(s, _)| *s).collect::<Vec<_>>(), vec![10, 20, 30]);
assert!((fused[0].1 - 1.0 / 60.0).abs() < 1e-12);
assert!((fused[1].1 - 1.0 / 61.0).abs() < 1e-12);
assert!((fused[2].1 - 1.0 / 62.0).abs() < 1e-12);
assert!(fused[0].1 > fused[1].1 && fused[1].1 > fused[2].1);
}
#[test]
fn rrf_rewards_agreement_across_legs() {
let a = [1i64, 2];
let b = [1i64, 3];
let fused = rrf_fuse(&[&a, &b]);
assert_eq!(fused[0].0, 1);
let top = fused.iter().find(|(s, _)| *s == 1).unwrap().1;
assert!((top - 2.0 / 60.0).abs() < 1e-12);
let two = fused.iter().find(|(s, _)| *s == 2).unwrap().1;
assert!(top > two);
}
#[test]
fn rrf_breaks_ties_by_seq_desc() {
let l1 = [5i64];
let l2 = [9i64];
let fused = rrf_fuse(&[&l1, &l2]);
assert!((fused[0].1 - fused[1].1).abs() < 1e-12);
assert_eq!(fused[0].0, 9);
assert_eq!(fused[1].0, 5);
}
#[test]
fn english_expander_substitutes_synonyms() {
let ex = EnglishExpander::default();
let v = ex.expand("cell");
assert!(v.contains(&"mobile".to_string()));
assert!(v.contains(&"phone".to_string()));
assert!(!v.contains(&"cell".to_string()));
}
#[test]
fn english_expander_stems_and_is_bounded() {
let ex = EnglishExpander::new(4);
assert!(ex.expand("cars").contains(&"car".to_string()));
assert!(ex.expand("").is_empty());
assert!(ex.expand("cell phone email car").len() <= 4);
}
fn open_tmp() -> (Areev, TempDir) {
let dir = TempDir::new().unwrap();
let path = dir.path().join("unit.db");
let db = Areev::open(path.to_str().unwrap()).unwrap();
(db, dir)
}
#[test]
fn next_hlc_is_strictly_monotonic_within_one_ms() {
let (mut db, _d) = open_tmp();
db.hlc_last = (now_ms() + 1_000_000) << 16;
let a = db.next_hlc();
let b = db.next_hlc();
let c = db.next_hlc();
assert_eq!(b, a + 1);
assert_eq!(c, b + 1);
}
#[test]
fn next_hlc_tracks_wall_clock_when_it_advances() {
let (mut db, _d) = open_tmp();
let before = now_ms();
db.hlc_last = 0;
let first = db.next_hlc();
let after = now_ms();
assert!(first >> 16 >= before && first >> 16 <= after);
}
#[test]
fn next_hlc_never_repeats_over_many_calls() {
let (mut db, _d) = open_tmp();
let mut last = db.next_hlc();
for _ in 0..5000 {
let n = db.next_hlc();
assert!(n > last, "HLC must strictly increase: {n} !> {last}");
last = n;
}
}
#[test]
fn term_id_interns_and_reverse_scans() {
let (mut db, _d) = open_tmp();
let a = db.term_id("alice").unwrap();
let b = db.term_id("bob").unwrap();
assert_ne!(a, b);
assert_eq!(db.term_id("alice").unwrap(), a);
assert_eq!(db.term_lookup("alice").unwrap(), Some(a));
assert_eq!(db.term_lookup("bob").unwrap(), Some(b));
assert_eq!(db.term_lookup("nobody").unwrap(), None);
assert_eq!(db.term_str(a).unwrap().as_deref(), Some("alice"));
assert_eq!(db.term_str(b).unwrap().as_deref(), Some("bob"));
assert_eq!(db.term_str(999_999).unwrap(), None);
}
#[test]
fn term_ids_persist_and_continue_across_reopen() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("dict.db");
let path = path.to_str().unwrap();
let (a, b);
{
let mut db = Areev::open(path).unwrap();
a = db.term_id("x").unwrap();
b = db.term_id("y").unwrap();
assert!(b > a);
}
{
let mut db = Areev::open(path).unwrap();
assert_eq!(db.term_lookup("x").unwrap(), Some(a));
assert_eq!(db.term_lookup("y").unwrap(), Some(b));
let c = db.term_id("z").unwrap();
assert!(c > b);
assert_eq!(db.term_str(c).unwrap().as_deref(), Some("z"));
}
}
#[test]
fn ns_registry_self_heals_when_stamp_is_missing_or_counts_drift() {
let dir = TempDir::new().unwrap();
let path = dir.path().join("heal.db");
let path = path.to_str().unwrap();
{
use areev_core::types::Fact;
let mut db = Areev::open(path).unwrap();
let mut f1 = Fact::new("john", "prefers", "window");
f1.common.namespace = Some("org.sales".into());
db.add(&f1).unwrap();
let mut f2 = Fact::new("john", "prefers", "aisle");
f2.common.namespace = Some("org.ops".into());
db.add(&f2).unwrap();
db.db.execute("DELETE FROM meta WHERE k = ?1", vec![pt(NS_REGISTRY_KEY)]).unwrap();
db.db.execute("DELETE FROM ns_reg", vec![]).unwrap();
db.db.execute("INSERT INTO ns_reg(ns, n) VALUES (99999, 7)", vec![]).unwrap();
}
{
let mut db = Areev::open(path).unwrap();
assert!(
db.open_warnings().iter().any(|w| w.contains("namespace registry")),
"the heal must be reported: {:?}",
db.open_warnings()
);
assert_eq!(
db.namespaces().unwrap(),
vec![("org.ops".to_string(), 1), ("org.sales".to_string(), 1)],
"registry rebuilt from grains, phantom row gone"
);
assert_eq!(db.recall("org.*", "john", None, 8).unwrap().len(), 2);
}
{
let db = Areev::open(path).unwrap();
assert!(
!db.open_warnings().iter().any(|w| w.contains("namespace registry")),
"a healed file must not re-heal: {:?}",
db.open_warnings()
);
}
}
}