use crate::{AtomicMutation, AtomicPrecondition, FeltDb, FlowError, StoredRow};
use serde::{de::DeserializeOwned, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
pub const LIFECYCLE_PREFIX: &str = "_feltdb.lifecycle.";
pub const LIFECYCLE_META: &str = "_feltdb.lifecycle.meta";
pub const LIFECYCLE_FORMAT_VERSION: u32 = 1;
pub fn is_lifecycle_collection(collection: &str) -> bool {
collection.starts_with(LIFECYCLE_PREFIX) || collection == LIFECYCLE_META
}
pub fn is_migrated(db: &FeltDb, store: &str) -> Result<bool, String> {
Ok(db
.get_collection_record(LIFECYCLE_META, &row_key(LIFECYCLE_META, store))
.map_err(|error| error.to_string())?
.is_some())
}
pub type LifecycleRecord = (String, String, Value);
pub trait LifecycleState: Default + Clone + Serialize + DeserializeOwned {
const STORE: &'static str;
const KINDS: &'static [&'static str];
const IMMUTABLE: &'static [&'static str] = &[];
fn records(&self) -> Vec<LifecycleRecord>;
fn from_records(records: Vec<LifecycleRecord>) -> Result<Self, String>;
}
fn collection(store: &str, kind: &str) -> String {
format!("{LIFECYCLE_PREFIX}{store}.{kind}")
}
fn row_key(capability: &str, key: &str) -> String {
format!("{capability}:{key}")
}
fn record_key<'a>(capability: &str, key: &'a str) -> &'a str {
key.strip_prefix(capability).and_then(|rest| rest.strip_prefix(':')).unwrap_or(key)
}
fn sequence(row: &StoredRow) -> Option<u64> {
row.version()
}
fn digest(value: &impl Serialize) -> Result<String, String> {
let bytes = serde_json::to_vec(value).map_err(|error| error.to_string())?;
Ok(format!("{:x}", Sha256::digest(bytes)))
}
#[derive(Clone, Default)]
pub struct Committed {
rows: BTreeMap<(String, String), (Value, Option<u64>)>,
versions: Vec<u64>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LifecycleOpen {
FeltDb { records: usize },
Fresh,
Migrated { records: usize, source: PathBuf },
}
#[derive(Clone)]
pub struct LifecycleRecords {
db: FeltDb,
store: &'static str,
}
impl LifecycleRecords {
pub fn open<S: LifecycleState>(
db: &FeltDb,
legacy: &Path,
) -> Result<(Self, S, Committed, LifecycleOpen), String> {
let records = Self { db: db.clone(), store: S::STORE };
let marker = records.marker()?;
let legacy_state = if legacy.exists() {
let bytes = std::fs::read(legacy).map_err(|error| format!("read {}: {error}", legacy.display()))?;
Some(serde_json::from_slice::<S>(&bytes).map_err(|error| {
format!("LIFECYCLE_MIGRATION_SOURCE_INVALID: {} is not a valid {} store: {error}. Nothing was imported.", legacy.display(), S::STORE)
})?)
} else {
None
};
match (marker, legacy_state) {
(Some(_), legacy_state) => {
let (state, committed) = records.load::<S>()?;
if let Some(legacy_state) = legacy_state {
if digest(&canonical(&legacy_state.records()))? != digest(&canonical(&state.records()))? {
return Err(format!(
"LIFECYCLE_SPLIT_AUTHORITY: {} is already migrated into FeltDB, but {} exists and differs from it. Refusing to choose one: remove or rename the file if FeltDB is correct, or restore the store from backup.",
S::STORE, legacy.display()
));
}
retire(legacy)?;
}
let count = committed.rows.len();
Ok((records, state, committed, LifecycleOpen::FeltDb { records: count }))
}
(None, None) => {
let (state, committed) = records.load::<S>()?;
Ok((records, state, committed, LifecycleOpen::Fresh))
}
(None, Some(source)) => {
let imported = source.records();
let rebuilt = S::from_records(imported.clone())?;
if digest(&canonical(&rebuilt.records()))? != digest(&canonical(&imported))? {
return Err(format!("LIFECYCLE_MIGRATION_SOURCE_INVALID: {} does not round-trip; nothing was imported", legacy.display()));
}
ensure_unique(&imported)?;
records.import::<S>(&source, &digest(&canonical(&imported))?)?;
let (state, committed) = records.load::<S>()?;
if digest(&canonical(&state.records()))? != digest(&canonical(&imported))? {
return Err(format!("LIFECYCLE_MIGRATION_VERIFY_FAILED: {} in FeltDB does not equal {} after import", S::STORE, legacy.display()));
}
retire(legacy)?;
Ok((records, state, committed, LifecycleOpen::Migrated { records: imported.len(), source: legacy.to_path_buf() }))
}
}
}
fn marker(&self) -> Result<Option<Value>, String> {
Ok(self
.db
.get_collection_record(LIFECYCLE_META, &row_key(LIFECYCLE_META, self.store))
.map_err(|error| error.to_string())?
.map(|row| row.value))
}
fn load<S: LifecycleState>(&self) -> Result<(S, Committed), String> {
let mut committed = Committed { versions: self.versions::<S>(), ..Committed::default() };
let mut records = Vec::new();
for kind in S::KINDS {
let capability = collection(self.store, kind);
let rows = self
.db
.query_collection(&capability, None, |row| !row.deleted)
.map_err(|error| error.to_string())?;
for mut row in rows {
row.key = record_key(&capability, &row.key).to_string();
let version = sequence(&row);
let mut value = row.value;
if let Value::Object(fields) = &mut value {
fields.remove("__version");
}
committed
.rows
.insert(((*kind).to_string(), row.key.clone()), (value.clone(), version));
records.push(((*kind).to_string(), row.key, value));
}
}
Ok((S::from_records(records)?, committed))
}
pub fn reload<S: LifecycleState>(&self) -> Result<(S, Committed), String> {
self.load::<S>()
}
fn versions<S: LifecycleState>(&self) -> Vec<u64> {
let collections: Vec<String> = S::KINDS.iter().map(|kind| collection(self.store, kind)).collect();
self.db.collection_write_versions(&collections)
}
pub fn changed_since<S: LifecycleState>(&self, committed: &Committed) -> bool {
self.versions::<S>() != committed.versions
}
fn import<S: LifecycleState>(&self, state: &S, source: &str) -> Result<(), String> {
let records = state.records();
let mut preconditions = Vec::with_capacity(records.len() + 1);
let mut mutations = Vec::with_capacity(records.len() + 1);
for (kind, key, value) in &records {
let capability = collection(self.store, kind);
preconditions.push(AtomicPrecondition { capability: capability.clone(), key: row_key(&capability, key), expected_version: None });
mutations.push(AtomicMutation { key: row_key(&capability, key), capability, value: Some(value.clone()) });
}
preconditions.push(AtomicPrecondition { capability: LIFECYCLE_META.into(), key: row_key(LIFECYCLE_META, self.store), expected_version: None });
mutations.push(AtomicMutation {
capability: LIFECYCLE_META.into(),
key: row_key(LIFECYCLE_META, self.store),
value: Some(json!({
"store": self.store,
"format_version": LIFECYCLE_FORMAT_VERSION,
"status": "complete",
"records": records.len(),
"source": source,
})),
});
self.db
.apply_atomic_transaction(
&format!("lifecycle-migrate-{}-{}", self.store, source),
None,
&preconditions,
&mutations,
Some(json!({ "kind": "lifecycle_migration", "store": self.store, "records": records.len() })),
)
.map(|_| ())
.map_err(|error| format!("LIFECYCLE_MIGRATION_FAILED: {} could not be imported: {error}", self.store))
}
pub fn commit<S: LifecycleState>(
&self,
before: &Committed,
next: &S,
operation: &str,
) -> Result<Committed, LifecycleCommitError> {
let next_records = next.records();
ensure_unique(&next_records).map_err(LifecycleCommitError::Invalid)?;
let mut after: BTreeMap<(String, String), Value> = BTreeMap::new();
for (kind, key, value) in next_records {
after.insert((kind, key), value);
}
let mut preconditions = Vec::new();
let mut mutations = Vec::new();
let mut changed = Vec::new();
for (identity, (old, version)) in &before.rows {
if S::IMMUTABLE.contains(&identity.0.as_str()) && after.get(identity) != Some(old) {
return Err(LifecycleCommitError::Invalid(format!(
"{}/{} is immutable once committed and cannot be changed or removed",
identity.0, identity.1
)));
}
if after.get(identity) != Some(old) {
let capability = collection(self.store, &identity.0);
preconditions.push(AtomicPrecondition { capability: capability.clone(), key: row_key(&capability, &identity.1), expected_version: *version });
mutations.push(AtomicMutation { key: row_key(&capability, &identity.1), capability, value: after.get(identity).cloned() });
changed.push(identity.clone());
}
}
for (identity, value) in &after {
if !before.rows.contains_key(identity) {
let capability = collection(self.store, &identity.0);
preconditions.push(AtomicPrecondition { capability: capability.clone(), key: row_key(&capability, &identity.1), expected_version: None });
mutations.push(AtomicMutation { key: row_key(&capability, &identity.1), capability, value: Some(value.clone()) });
changed.push(identity.clone());
}
}
if mutations.is_empty() {
return Ok(before.clone());
}
if self.marker().map_err(|error| LifecycleCommitError::Storage(FlowError::CapabilityError(error)))?.is_none() {
preconditions.push(AtomicPrecondition { capability: LIFECYCLE_META.into(), key: row_key(LIFECYCLE_META, self.store), expected_version: None });
mutations.push(AtomicMutation {
capability: LIFECYCLE_META.into(),
key: row_key(LIFECYCLE_META, self.store),
value: Some(json!({
"store": self.store,
"format_version": LIFECYCLE_FORMAT_VERSION,
"status": "complete",
"records": 0,
"source": "fresh",
})),
});
}
let transaction_id = format!("lifecycle-{}-{:032x}", self.store, rand::random::<u128>());
match self.db.apply_atomic_transaction(
&transaction_id,
None,
&preconditions,
&mutations,
Some(json!({ "kind": "lifecycle", "store": self.store, "operation": operation, "records": mutations.len() })),
) {
Ok(_) => {}
Err(FlowError::CapabilityError(detail)) if detail.starts_with("PRECONDITION_FAILED") => {
return Err(LifecycleCommitError::Stale(format!(
"{} changed in FeltDB since this store last read it ({detail}); nothing was applied",
self.store
)))
}
Err(FlowError::PreconditionFailed(detail)) => {
return Err(LifecycleCommitError::Stale(format!(
"{} changed in FeltDB since this store last read it ({detail:?}); nothing was applied",
self.store
)))
}
Err(error) => return Err(LifecycleCommitError::Storage(error)),
}
let mut committed = before.clone();
for identity in changed {
let capability = collection(self.store, &identity.0);
let row = self
.db
.get_collection_record(&capability, &row_key(&capability, &identity.1))
.map_err(LifecycleCommitError::Storage)?;
match (row, after.get(&identity)) {
(Some(row), Some(value)) => {
committed.rows.insert(identity, (value.clone(), sequence(&row)));
}
_ => {
committed.rows.remove(&identity);
}
}
}
committed.versions = Vec::new();
Ok(committed)
}
pub fn store(&self) -> &'static str {
self.store
}
}
#[derive(Debug)]
pub enum LifecycleCommitError {
Invalid(String),
Stale(String),
Storage(FlowError),
}
impl std::fmt::Display for LifecycleCommitError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Invalid(message) | Self::Stale(message) => formatter.write_str(message),
Self::Storage(error) => write!(formatter, "lifecycle commit failed: {error}"),
}
}
}
fn ensure_unique(records: &[LifecycleRecord]) -> Result<(), String> {
let mut seen = std::collections::BTreeSet::new();
for (kind, key, _) in records {
if !seen.insert((kind.as_str(), key.as_str())) {
return Err(format!("duplicate lifecycle record identity {kind}/{key}"));
}
}
Ok(())
}
fn canonical(records: &[LifecycleRecord]) -> Vec<LifecycleRecord> {
let mut sorted = records.to_vec();
sorted.sort_by(|left, right| (&left.0, &left.1).cmp(&(&right.0, &right.1)));
sorted
}
fn retire(legacy: &Path) -> Result<(), String> {
let mut retired = legacy.as_os_str().to_owned();
retired.push(".migrated");
let retired = PathBuf::from(retired);
if retired.exists() {
std::fs::remove_file(legacy).map_err(|error| format!("remove {}: {error}", legacy.display()))
} else {
std::fs::rename(legacy, &retired).map_err(|error| format!("retire {}: {error}", legacy.display()))
}
}
const POSITION: &str = "_lifecycle_position";
pub fn keyed<T: Serialize>(kind: &str, values: &[T], key: impl Fn(&T) -> String) -> Vec<LifecycleRecord> {
values
.iter()
.enumerate()
.map(|(position, typed)| {
let mut value = serde_json::to_value(typed).expect("lifecycle records serialize");
if let Value::Object(fields) = &mut value {
fields.insert(POSITION.into(), json!(position));
}
(kind.to_string(), key(typed), value)
})
.collect()
}
pub fn keyed_as_is<T: Serialize>(kind: &str, values: &[T], key: impl Fn(&T) -> String) -> Vec<LifecycleRecord> {
values
.iter()
.map(|typed| (kind.to_string(), key(typed), serde_json::to_value(typed).expect("lifecycle records serialize")))
.collect()
}
pub fn values_of<T: DeserializeOwned>(records: &[LifecycleRecord], kind: &str) -> Result<Vec<T>, String> {
let mut ordered: Vec<(u64, T)> = records
.iter()
.filter(|(record_kind, _, _)| record_kind == kind)
.map(|(_, key, value)| {
let mut value = value.clone();
let position = match &mut value {
Value::Object(fields) => fields.remove(POSITION).and_then(|value| value.as_u64()).unwrap_or(u64::MAX),
_ => u64::MAX,
};
serde_json::from_value(value).map(|typed| (position, typed)).map_err(|error| format!("{kind}/{key}: {error}"))
})
.collect::<Result<_, _>>()?;
ordered.sort_by_key(|(position, _)| *position);
Ok(ordered.into_iter().map(|(_, value)| value).collect())
}
#[derive(Clone)]
pub struct LifecycleBinding {
durability: BindingDurability,
}
#[derive(Clone)]
enum BindingDurability {
File(PathBuf),
FeltDb(LifecycleRecords, std::sync::Arc<std::sync::Mutex<Committed>>),
}
impl LifecycleBinding {
pub fn file<S: LifecycleState>(path: impl Into<PathBuf>) -> Result<(Self, S), String> {
let path = path.into();
let state = if path.exists() {
let bytes = std::fs::read(&path).map_err(|error| error.to_string())?;
serde_json::from_slice(&bytes).map_err(|error| error.to_string())?
} else {
S::default()
};
Ok((Self { durability: BindingDurability::File(path) }, state))
}
pub fn feltdb<S: LifecycleState>(db: &FeltDb, legacy: &Path) -> Result<(Self, S, LifecycleOpen), String> {
let (records, state, committed, opened) = LifecycleRecords::open::<S>(db, legacy)?;
Ok((
Self { durability: BindingDurability::FeltDb(records, std::sync::Arc::new(std::sync::Mutex::new(committed))) },
state,
opened,
))
}
pub fn is_feltdb(&self) -> bool {
matches!(self.durability, BindingDurability::FeltDb(..))
}
pub fn refresh<S: LifecycleState>(&self, records: &mut S) -> Result<(), String> {
if let BindingDurability::FeltDb(durable, committed) = &self.durability {
let mut committed = committed.lock().map_err(|_| format!("{} store lock poisoned", S::STORE))?;
if durable.changed_since::<S>(&committed) {
let (state, reloaded) = durable.reload::<S>()?;
*records = state;
*committed = reloaded;
}
}
Ok(())
}
pub fn write<'a, S: LifecycleState>(
&self,
records: &'a std::sync::RwLock<S>,
) -> Result<std::sync::RwLockWriteGuard<'a, S>, String> {
let mut guard = records.write().map_err(|_| format!("{} store lock poisoned", S::STORE))?;
self.refresh(&mut *guard)?;
Ok(guard)
}
pub fn read<'a, S: LifecycleState>(
&self,
records: &'a std::sync::RwLock<S>,
) -> std::sync::LockResult<std::sync::RwLockReadGuard<'a, S>> {
if let BindingDurability::FeltDb(durable, committed) = &self.durability {
let stale = committed.lock().map(|committed| durable.changed_since::<S>(&committed)).unwrap_or(false);
if stale {
if let Ok(mut guard) = records.write() {
let _ = self.refresh(&mut *guard);
}
}
}
records.read()
}
pub fn persist<S: LifecycleState>(&self, guard: &mut S, next: S, operation: &str) -> Result<(), String> {
match &self.durability {
BindingDurability::File(path) => {
crate::application::write_file_durably(path, &serde_json::to_vec_pretty(&next).map_err(|error| error.to_string())?)?;
}
BindingDurability::FeltDb(durable, committed) => {
let mut committed = committed.lock().map_err(|_| format!("{} store lock poisoned", S::STORE))?;
match durable.commit(&committed, &next, operation) {
Ok(after) => *committed = after,
Err(LifecycleCommitError::Stale(message)) => {
let (state, reloaded) = durable.reload::<S>()?;
*guard = state;
*committed = reloaded;
return Err(format!("lifecycle_state_changed: {message}"));
}
Err(error) => return Err(error.to_string()),
}
}
}
*guard = next;
Ok(())
}
pub fn mutate<S: LifecycleState, T>(
&self,
records: &std::sync::RwLock<S>,
operation: &str,
change: impl FnOnce(&mut S) -> Result<T, String>,
) -> Result<T, String> {
let mut guard = self.write(records)?;
let mut next = guard.clone();
let result = change(&mut next)?;
self.persist(&mut *guard, next, operation)?;
Ok(result)
}
}