Menu
Public repository / 228b5716ba0b7bcc87cce261ea6df891133a35bb
AkurAI-CRM / crates/crm-api/src/lifecycle.rs
Latest change 73231e93c61e6e1e5c151cf2295c4da7dd4fa201 - style: cargo fmt (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)),
]))
}