Menu
Public repository / e3692e977de85689a08a193853852f355b252bc8
AkurAI-CRM / crates/crm-api/src/lifecycle.rs
Latest change e3692e977de85689a08a193853852f355b252bc8 - feat: add lifecycle growth analytics AKURAI-CRM-7 by Ólafur Búi Ólafsson
use crate::graph::GraphError;
use crate::router_setup::{upper_bound, CrmState};
use akurai_json::Value;
use akurai_storage::BTree;
use std::collections::{BTreeMap, BTreeSet};
pub const MAX_ANALYTICS_LIMIT: usize = 100;
const TOUCHPOINTS: &str = "lifecycle_touchpoints";
const HEALTH: &str = "customer_health_snapshots";
const EXPERIMENTS: &str = "growth_experiments";
const SCHEMA_KEY: &[u8] = b"_schema:growth_analytics";
const STAGES: [&str; 4] = ["acquisition", "activation", "retention", "expansion"];
const FORBIDDEN: [&str; 9] = [
"body", "email", "note", "password", "secret", "token", "credential", "attachment", "content",
];
fn now() -> i64 { CrmState::now() }
fn text(value: &Value, name: &str) -> Option<String> {
value.get(name).and_then(Value::as_str).map(ToOwned::to_owned)
}
fn integer(value: &Value, name: &str) -> Option<i64> { value.get(name).and_then(Value::as_i64) }
fn key(collection: &str, id: u64) -> Vec<u8> {
let mut key = format!("{collection}:").into_bytes();
key.extend_from_slice(&id.to_be_bytes());
key
}
fn parse(bytes: &[u8]) -> Result<Value, GraphError> {
akurai_json::parse(std::str::from_utf8(bytes).unwrap_or(""))
.map_err(|_| GraphError::Storage("corrupted lifecycle record".into()))
}
fn records(db: &mut BTree, collection: &str) -> Result<Vec<Value>, GraphError> {
let prefix = format!("{collection}:").into_bytes();
let entries = db.range(&prefix, &upper_bound(&prefix)).map_err(|e| GraphError::Storage(format!("storage: {e}")))?;
entries.into_iter().filter(|(key, _)| !key.ends_with(b":_counter"))
.map(|(_, bytes)| parse(&bytes)).collect()
}
fn next_id(db: &mut BTree, collection: &str) -> Result<u64, GraphError> {
let counter = format!("{collection}:_counter").into_bytes();
let id = db.get(&counter).map_err(|e| GraphError::Storage(format!("counter: {e}")))?
.and_then(|bytes| String::from_utf8(bytes).ok()).and_then(|value| value.parse::<u64>().ok())
.unwrap_or(0).checked_add(1).ok_or_else(|| GraphError::Storage("analytics ID space exhausted".into()))?;
db.insert(&counter, id.to_string().as_bytes()).map_err(|e| GraphError::Storage(format!("counter: {e}")))?;
Ok(id)
}
fn company_owned(db: &mut BTree, tenant_id: &str, company_id: u64) -> Result<(), GraphError> {
let company = db.get(&key("companies", company_id)).map_err(|e| GraphError::Storage(format!("storage: {e}")))?
.ok_or_else(|| GraphError::NotFound(format!("companies/{company_id}")))?;
if text(&parse(&company)?, "tenantId").as_deref() != Some(tenant_id) {
return Err(GraphError::Forbidden("company belongs to another tenant".into()));
}
Ok(())
}
fn validate_safe(value: &Value) -> Result<(), GraphError> {
match value {
Value::Object(fields) => for (name, nested) in fields {
if FORBIDDEN.iter().any(|forbidden| name.eq_ignore_ascii_case(forbidden)) {
return Err(GraphError::BadRequest(format!("{name} is not analytics data")));
}
validate_safe(nested)?;
},
Value::Array(values) => for value in values { validate_safe(value)?; },
_ => {}
}
Ok(())
}
fn required_text(value: &Value, name: &str, max: usize) -> Result<String, GraphError> {
let value = text(value, name).unwrap_or_default();
let value = value.trim();
if value.is_empty() || value.len() > max || value.chars().any(char::is_control) {
return Err(GraphError::BadRequest(format!("invalid {name}")));
}
Ok(value.to_string())
}
fn metric(value: &Value) -> Result<Value, GraphError> {
let name = required_text(value, "name", 80)?;
let metric_value = integer(value, "value").ok_or_else(|| GraphError::BadRequest("metric value is required".into()))?;
let sample_size = integer(value, "sampleSize").unwrap_or(1);
if !(1..=1_000_000_000).contains(&sample_size) { return Err(GraphError::BadRequest("invalid metric sampleSize".into())); }
Ok(Value::Object(vec![
("name".into(), Value::Str(name)), ("value".into(), Value::Int(metric_value)),
("sampleSize".into(), Value::Int(sample_size)),
]))
}
fn metrics(value: &Value, name: &str) -> Result<Vec<Value>, GraphError> {
let values = match value.get(name) { Some(Value::Array(values)) if !values.is_empty() && values.len() <= 20 => values, _ => return Err(GraphError::BadRequest(format!("{name} is required"))), };
values.iter().map(metric).collect()
}
fn bounded(limit: usize) -> usize { limit.clamp(1, MAX_ANALYTICS_LIMIT) }
pub fn ensure_schema(db: &mut BTree) -> Result<(), String> {
if let Some(bytes) = db.get(SCHEMA_KEY).map_err(|e| e.to_string())? {
let marker = akurai_json::parse(std::str::from_utf8(&bytes).unwrap_or("")).map_err(|_| "invalid growth analytics schema marker")?;
if text(&marker, "name").as_deref() != Some("growth_analytics") || integer(&marker, "version") != Some(1) {
return Err("unsupported growth analytics schema marker".into());
}
return Ok(());
}
db.insert(SCHEMA_KEY, Value::Object(vec![
("name".into(), Value::Str("growth_analytics".into())), ("version".into(), Value::Int(1)), ("createdAt".into(), Value::Int(now())),
]).to_json().as_bytes()).map_err(|e| e.to_string())?;
db.commit().map_err(|e| e.to_string())
}
pub fn record_touchpoint(db: &mut BTree, tenant_id: &str, company_id: u64, input: &Value) -> Result<Value, GraphError> {
validate_safe(input)?;
company_owned(db, tenant_id, company_id)?;
let source = required_text(input, "source", 120)?;
let event = required_text(input, "event", 120)?;
let stage = required_text(input, "stage", 32)?;
if !STAGES.contains(&stage.as_str()) { return Err(GraphError::BadRequest("stage must be acquisition/activation/retention/expansion".into())); }
let occurred_at = integer(input, "occurredAt").unwrap_or_else(now);
if occurred_at <= 0 || occurred_at > now() + 300 { return Err(GraphError::BadRequest("invalid occurredAt".into())); }
let mut fields = vec![
("id".into(), Value::Int(next_id(db, TOUCHPOINTS)? as i64)), ("tenantId".into(), Value::Str(tenant_id.into())),
("companyId".into(), Value::Int(company_id as i64)), ("source".into(), Value::Str(source)),
("event".into(), Value::Str(event)), ("stage".into(), Value::Str(stage)), ("occurredAt".into(), Value::Int(occurred_at)),
("recordedAt".into(), Value::Int(now())),
];
if let Some(value) = integer(input, "value") { fields.push(("value".into(), Value::Int(value))); }
let record = Value::Object(fields);
let id = integer(&record, "id").unwrap_or_default() as u64;
db.insert(&key(TOUCHPOINTS, id), record.to_json().as_bytes()).map_err(|e| GraphError::Storage(format!("touchpoint: {e}")))?;
db.commit().map_err(|e| GraphError::Storage(format!("touchpoint commit: {e}")))?;
Ok(record)
}
pub fn record_health_snapshot(db: &mut BTree, tenant_id: &str, company_id: u64, input: &Value) -> Result<Value, GraphError> {
validate_safe(input)?;
company_owned(db, tenant_id, company_id)?;
let score = integer(input, "score").ok_or_else(|| GraphError::BadRequest("score is required".into()))?;
if !(0..=100).contains(&score) { return Err(GraphError::BadRequest("score must be 0-100".into())); }
let components = match input.get("components") { Some(Value::Array(values)) if !values.is_empty() && values.len() <= 12 => values, _ => return Err(GraphError::BadRequest("components are required".into())), };
let mut facts = Vec::new();
for component in components {
let name = required_text(component, "name", 80)?;
let component_score = integer(component, "score").ok_or_else(|| GraphError::BadRequest("component score is required".into()))?;
let weight = integer(component, "weight").unwrap_or(1);
let fact_count = integer(component, "factCount").unwrap_or(0);
if !(0..=100).contains(&component_score) || !(1..=100).contains(&weight) || !(0..=1_000_000).contains(&fact_count) { return Err(GraphError::BadRequest("invalid health component".into())); }
facts.push(Value::Object(vec![
("name".into(), Value::Str(name)), ("score".into(), Value::Int(component_score)),
("weight".into(), Value::Int(weight)), ("factCount".into(), Value::Int(fact_count)),
]));
}
let record = Value::Object(vec![
("id".into(), Value::Int(next_id(db, HEALTH)? as i64)), ("tenantId".into(), Value::Str(tenant_id.into())),
("companyId".into(), Value::Int(company_id as i64)), ("score".into(), Value::Int(score)),
("components".into(), Value::Array(facts)), ("recordedAt".into(), Value::Int(now())),
]);
let id = integer(&record, "id").unwrap_or_default() as u64;
db.insert(&key(HEALTH, id), record.to_json().as_bytes()).map_err(|e| GraphError::Storage(format!("health: {e}")))?;
db.commit().map_err(|e| GraphError::Storage(format!("health commit: {e}")))?;
Ok(record)
}
pub fn create_experiment(db: &mut BTree, tenant_id: &str, input: &Value) -> Result<Value, GraphError> {
validate_safe(input)?;
let name = required_text(input, "name", 160)?;
let hypothesis = required_text(input, "hypothesis", 1_000)?;
let decision = text(input, "decision").unwrap_or_else(|| "iterate".into());
if !["scale", "iterate", "stop"].contains(&decision.as_str()) { return Err(GraphError::BadRequest("decision must be scale/iterate/stop".into())); }
let baseline = metrics(input, "baselineMetrics")?;
let success = metrics(input, "successMetrics")?;
let guardrails = metrics(input, "guardrailMetrics")?;
let measurements = match input.get("measurements") { Some(Value::Array(values)) if values.len() <= 100 => values.iter().map(metric).collect::<Result<Vec<_>, _>>()?, _ => Vec::new(), };
let record = Value::Object(vec![
("id".into(), Value::Int(next_id(db, EXPERIMENTS)? as i64)), ("tenantId".into(), Value::Str(tenant_id.into())),
("name".into(), Value::Str(name)), ("hypothesis".into(), Value::Str(hypothesis)), ("decision".into(), Value::Str(decision)),
("baselineMetrics".into(), Value::Array(baseline)), ("successMetrics".into(), Value::Array(success)),
("guardrailMetrics".into(), Value::Array(guardrails)), ("measurements".into(), Value::Array(measurements)),
("recordedAt".into(), Value::Int(now())),
]);
let id = integer(&record, "id").unwrap_or_default() as u64;
db.insert(&key(EXPERIMENTS, id), record.to_json().as_bytes()).map_err(|e| GraphError::Storage(format!("experiment: {e}")))?;
db.commit().map_err(|e| GraphError::Storage(format!("experiment commit: {e}")))?;
Ok(record)
}
fn tenant_records(db: &mut BTree, collection: &str, tenant_id: &str) -> Result<Vec<Value>, GraphError> {
records(db, collection).map(|records| records.into_iter().filter(|record| text(record, "tenantId").as_deref() == Some(tenant_id)).collect())
}
fn latest_health(db: &mut BTree, tenant_id: &str, company_id: u64) -> Result<Option<Value>, GraphError> {
Ok(tenant_records(db, HEALTH, tenant_id)?.into_iter().filter(|record| integer(record, "companyId") == Some(company_id as i64)).max_by_key(|record| integer(record, "recordedAt").unwrap_or_default()))
}
pub fn customer_health(db: &mut BTree, tenant_id: &str, company_id: u64) -> Result<Value, GraphError> {
company_owned(db, tenant_id, company_id)?;
let snapshot = latest_health(db, tenant_id, company_id)?.ok_or_else(|| GraphError::NotFound("no health snapshot for company".into()))?;
let recorded_at = integer(&snapshot, "recordedAt").unwrap_or_default();
Ok(Value::Object(vec![
("companyId".into(), Value::Int(company_id as i64)), ("health".into(), snapshot),
("freshness".into(), Value::Object(vec![("asOf".into(), Value::Int(recorded_at)), ("ageSeconds".into(), Value::Int((now() - recorded_at).max(0)))])),
("explanation".into(), Value::Str("components contain the scored facts and weights used for this snapshot".into())),
]))
}
fn aggregate_window(events: &[Value], window: i64, limit: usize) -> Value {
let since = now() - window;
let scoped = events.iter().filter(|event| integer(event, "occurredAt").unwrap_or_default() >= since).collect::<Vec<_>>();
let mut stages: BTreeMap<&str, BTreeSet<i64>> = STAGES.into_iter().map(|stage| (stage, BTreeSet::new())).collect();
let mut sources: BTreeMap<String, (usize, BTreeSet<i64>)> = BTreeMap::new();
let mut cohorts: BTreeMap<i64, BTreeSet<i64>> = BTreeMap::new();
let mut latest = 0;
for event in &scoped {
let company = integer(event, "companyId").unwrap_or_default();
let stage = text(event, "stage").unwrap_or_default();
if let Some(companies) = stages.get_mut(stage.as_str()) { companies.insert(company); }
let source = text(event, "source").unwrap_or_default();
let entry = sources.entry(source).or_default(); entry.0 += 1; entry.1.insert(company);
let occurred_at = integer(event, "occurredAt").unwrap_or_default(); latest = latest.max(occurred_at);
if stage == "activation" { cohorts.entry(occurred_at / 2_592_000 * 2_592_000).or_default().insert(company); }
}
let mut by_company: BTreeMap<i64, Vec<&&Value>> = BTreeMap::new();
for event in &scoped { by_company.entry(integer(event, "companyId").unwrap_or_default()).or_default().push(event); }
let mut stage_seconds: BTreeMap<String, (i64, usize)> = BTreeMap::new();
for timeline in by_company.values_mut() {
timeline.sort_by_key(|event| integer(event, "occurredAt").unwrap_or_default());
for pair in timeline.windows(2) {
let stage = text(pair[0], "stage").unwrap_or_default();
let elapsed = (integer(pair[1], "occurredAt").unwrap_or_default() - integer(pair[0], "occurredAt").unwrap_or_default()).max(0);
let entry = stage_seconds.entry(stage).or_default(); entry.0 += elapsed; entry.1 += 1;
}
}
let time_in_stage = stage_seconds.into_iter().map(|(stage, (seconds, sample_size))| Value::Object(vec![
("stage".into(), Value::Str(stage)), ("totalSeconds".into(), Value::Int(seconds)),
("sampleSize".into(), Value::Int(sample_size as i64)),
])).collect();
let mut previous = 0usize;
let funnel = STAGES.into_iter().map(|stage| {
let sample = stages[stage].len();
let conversion = sample.saturating_mul(1_000_000).checked_div(previous).unwrap_or(1_000_000);
previous = sample;
Value::Object(vec![("stage".into(), Value::Str(stage.into())), ("sampleSize".into(), Value::Int(sample as i64)), ("conversionPpmFromPrevious".into(), Value::Int(conversion as i64))])
}).collect();
let mut accounts = BTreeSet::new();
let source_values = sources.into_iter().map(|(source, (events, companies))| Value::Object(vec![("source".into(), Value::Str(source)), ("eventCount".into(), Value::Int(events as i64)), ("sampleSize".into(), Value::Int(companies.len() as i64))])).collect();
let drilldown = scoped.iter().take(limit).map(|event| { accounts.insert(integer(event, "companyId").unwrap_or_default()); (*event).clone() }).collect();
Value::Object(vec![
("windowSeconds".into(), Value::Int(window)), ("sampleSize".into(), Value::Int(accounts.len() as i64)),
("funnel".into(), Value::Array(funnel)), ("timeInStage".into(), Value::Array(time_in_stage)), ("sources".into(), Value::Array(source_values)),
("cohorts".into(), Value::Array(cohorts.into_iter().map(|(start, companies)| Value::Object(vec![("cohortStart".into(), Value::Int(start)), ("sampleSize".into(), Value::Int(companies.len() as i64))])).collect())),
("freshness".into(), Value::Object(vec![("generatedAt".into(), Value::Int(now())), ("latestEventAt".into(), Value::Int(latest)), ("ageSeconds".into(), Value::Int(if latest == 0 { -1 } else { (now() - latest).max(0) }))])),
("drilldown".into(), Value::Object(vec![("events".into(), Value::Array(drilldown)), ("bounded".into(), Value::Bool(true)), ("limit".into(), Value::Int(limit as i64))])),
])
}
pub fn growth_aggregates(db: &mut BTree, tenant_id: &str, limit: usize) -> Result<Value, GraphError> {
let events = tenant_records(db, TOUCHPOINTS, tenant_id)?;
let limit = bounded(limit);
Ok(Value::Object(vec![
("windows".into(), Value::Array([86_400, 604_800, 2_592_000].into_iter().map(|window| aggregate_window(&events, window, limit)).collect())),
("eventSampleSize".into(), Value::Int(events.len() as i64)), ("bounded".into(), Value::Bool(true)), ("limit".into(), Value::Int(limit as i64)),
]))
}
pub fn lifecycle_profile(db: &mut BTree, tenant_id: &str, company_id: u64, limit: usize) -> Result<Value, GraphError> {
company_owned(db, tenant_id, company_id)?;
let mut events = tenant_records(db, TOUCHPOINTS, tenant_id)?.into_iter().filter(|event| integer(event, "companyId") == Some(company_id as i64)).collect::<Vec<_>>();
events.sort_by_key(|event| integer(event, "occurredAt").unwrap_or_default());
let first_value = events.iter().find(|event| text(event, "event").as_deref() == Some("first_value")).and_then(|event| integer(event, "occurredAt"));
let activated_at = events.iter().find(|event| text(event, "stage").as_deref() == Some("activation")).and_then(|event| integer(event, "occurredAt"));
let mut stage_seconds: BTreeMap<String, i64> = BTreeMap::new();
for pair in events.windows(2) {
let elapsed = (integer(&pair[1], "occurredAt").unwrap_or_default() - integer(&pair[0], "occurredAt").unwrap_or_default()).max(0);
*stage_seconds.entry(text(&pair[0], "stage").unwrap_or_default()).or_default() += elapsed;
}
let time_in_stage = Value::Array(stage_seconds.into_iter().map(|(stage, seconds)| Value::Object(vec![
("stage".into(), Value::Str(stage)), ("seconds".into(), Value::Int(seconds)),
])).collect());
let truncated = events.len() > bounded(limit); events.truncate(bounded(limit));
Ok(Value::Object(vec![("companyId".into(), Value::Int(company_id as i64)), ("firstValueAt".into(), first_value.map(Value::Int).unwrap_or(Value::Null)), ("activatedAt".into(), activated_at.map(Value::Int).unwrap_or(Value::Null)), ("timeInStage".into(), time_in_stage), ("touchpoints".into(), Value::Array(events)), ("truncated".into(), Value::Bool(truncated)), ("complete".into(), Value::Bool(!truncated))]))
}
pub fn list_experiments(db: &mut BTree, tenant_id: &str, limit: usize) -> Result<Value, GraphError> {
let mut experiments = tenant_records(db, EXPERIMENTS, tenant_id)?; experiments.sort_by_key(|experiment| std::cmp::Reverse(integer(experiment, "recordedAt").unwrap_or_default()));
let truncated = experiments.len() > bounded(limit); experiments.truncate(bounded(limit));
Ok(Value::Object(vec![("experiments".into(), Value::Array(experiments)), ("bounded".into(), Value::Bool(true)), ("limit".into(), Value::Int(bounded(limit) as i64)), ("truncated".into(), Value::Bool(truncated)), ("complete".into(), Value::Bool(!truncated))]))
}
pub fn growth_recommendations(db: &mut BTree, tenant_id: &str, limit: usize) -> Result<Value, GraphError> {
let limit = bounded(limit);
let mut recommendations = Vec::new();
let mut latest: BTreeMap<i64, Value> = BTreeMap::new();
for health in tenant_records(db, HEALTH, tenant_id)? { let company = integer(&health, "companyId").unwrap_or_default(); if latest.get(&company).is_none_or(|old| integer(&health, "recordedAt") > integer(old, "recordedAt")) { latest.insert(company, health); } }
let health_sample_size = latest.len();
for (company_id, health) in latest { if integer(&health, "score").unwrap_or(100) < 60 { recommendations.push(Value::Object(vec![("action".into(), Value::Str("retain".into())), ("companyId".into(), Value::Int(company_id)), ("reason".into(), Value::Str("customer health score is below 60; inspect component facts".into())), ("health".into(), health)])); } }
for experiment in tenant_records(db, EXPERIMENTS, tenant_id)? { recommendations.push(Value::Object(vec![("action".into(), text(&experiment, "decision").map(Value::Str).unwrap_or_else(|| Value::Str("iterate".into()))), ("experimentId".into(), integer(&experiment, "id").map(Value::Int).unwrap_or(Value::Null)), ("reason".into(), Value::Str("experiment decision is based on its retained baseline, success and guardrail metrics".into()))])); }
let truncated = recommendations.len() > limit; recommendations.truncate(limit);
Ok(Value::Object(vec![("recommendations".into(), Value::Array(recommendations)), ("sampleSize".into(), Value::Int(health_sample_size as i64)), ("freshness".into(), Value::Object(vec![("generatedAt".into(), Value::Int(now()))])), ("bounded".into(), Value::Bool(true)), ("limit".into(), Value::Int(limit as i64)), ("truncated".into(), Value::Bool(truncated)), ("complete".into(), Value::Bool(!truncated))]))
}