Menu
Public repository / main
AkurAI-CRM / crates/crm-api/src/graph.rs
Latest change 3ca078e63808e39f59571f8562e9a829f261ae13 - Restore CSRF guard, admin-gated merge, IdP logout and login redirects by Ólafur Búi Ólafsson
use crate::auth::session;
use crate::handlers::record_timeline;
use crate::router_setup::{
bad_request, error_response, internal_error, json_response, not_found, upper_bound, CrmState,
};
use akurai_http::{Request, Response};
use akurai_json::Value;
use akurai_storage::BTree;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
pub const GRAPH_SCHEMA_VERSION: i64 = 2;
pub const MAX_GRAPH_LIMIT: usize = 100;
const LINK_COLLECTION: &str = "account_links";
const AUDIT_COLLECTION: &str = "account_graph_audit";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LinkKind {
Organization,
Workspace,
Subscription,
Opportunity,
Installation,
}
impl LinkKind {
pub fn parse(value: &str) -> Option<Self> {
match value.to_ascii_lowercase().as_str() {
"organization" | "org" => Some(Self::Organization),
"workspace" => Some(Self::Workspace),
"subscription" => Some(Self::Subscription),
"opportunity" => Some(Self::Opportunity),
"installation" => Some(Self::Installation),
_ => None,
}
}
pub fn as_str(self) -> &'static str {
match self {
Self::Organization => "organization",
Self::Workspace => "workspace",
Self::Subscription => "subscription",
Self::Opportunity => "opportunity",
Self::Installation => "installation",
}
}
fn owner(self) -> &'static str {
match self {
Self::Organization => "idp",
Self::Workspace | Self::Subscription | Self::Installation => "platform",
Self::Opportunity => "crm",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LinkStatus {
Active,
Stale,
Orphaned,
Unlinked,
Merged,
}
impl LinkStatus {
fn as_str(self) -> &'static str {
match self {
Self::Active => "active",
Self::Stale => "stale",
Self::Orphaned => "orphaned",
Self::Unlinked => "unlinked",
Self::Merged => "merged",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AccountLink {
pub id: u64,
pub tenant_id: String,
pub company_id: u64,
pub kind: LinkKind,
pub external_id: String,
pub source_owner: String,
pub status: LinkStatus,
pub created_at: i64,
pub updated_at: i64,
pub stale_at: Option<i64>,
pub unlinked_at: Option<i64>,
pub merged_into: Option<u64>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GraphError {
BadRequest(String),
Unauthorized(String),
Forbidden(String),
NotFound(String),
Conflict(String),
Storage(String),
}
impl GraphError {
pub fn response(self) -> Response {
match self {
Self::BadRequest(message) => bad_request(&message),
Self::Unauthorized(message) => error_response(401, &message),
Self::Forbidden(message) => error_response(403, &message),
Self::NotFound(message) => not_found(&message),
Self::Conflict(message) => error_response(409, &message),
Self::Storage(message) => internal_error(&message),
}
}
}
fn now() -> i64 {
CrmState::now()
}
fn text_field(value: &Value, name: &str) -> Option<String> {
value
.get(name)
.and_then(Value::as_str)
.map(ToOwned::to_owned)
}
fn int_field(value: &Value, name: &str) -> Option<u64> {
value
.get(name)
.and_then(Value::as_i64)
.and_then(|n| u64::try_from(n).ok())
}
fn object_fields(value: Value) -> Result<Vec<(String, Value)>, GraphError> {
match value {
Value::Object(fields) => Ok(fields),
_ => Err(GraphError::BadRequest("expected JSON object".into())),
}
}
fn parse_body(req: &Request) -> Result<Value, GraphError> {
if req.body.is_empty() {
return Err(GraphError::BadRequest("empty request body".into()));
}
let text = std::str::from_utf8(&req.body)
.map_err(|_| GraphError::BadRequest("body not UTF-8".into()))?;
akurai_json::parse(text).map_err(|e| GraphError::BadRequest(format!("invalid JSON: {e}")))
}
fn extract_ids(path: &str) -> Vec<u64> {
path.split('/')
.filter_map(|part| part.parse().ok())
.collect()
}
fn set_field(fields: &mut Vec<(String, Value)>, name: &str, value: Value) {
if let Some((_, existing)) = fields.iter_mut().find(|(key, _)| key == name) {
*existing = value;
} else {
fields.push((name.to_string(), value));
}
}
fn build_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_record(bytes: &[u8]) -> Result<Value, GraphError> {
akurai_json::parse(std::str::from_utf8(bytes).unwrap_or(""))
.map_err(|_| GraphError::Storage("corrupted graph record".into()))
}
fn get_record(db: &mut BTree, collection: &str, id: u64) -> Result<Option<Value>, GraphError> {
db.get(&build_key(collection, id))
.map_err(|e| GraphError::Storage(format!("storage: {e}")))?
.map(|bytes| parse_record(&bytes))
.transpose()
}
fn scan_all_records(db: &mut BTree, collection: &str) -> Result<Vec<Value>, GraphError> {
let prefix = format!("{collection}:").into_bytes();
let end = upper_bound(&prefix);
let entries = db
.range(&prefix, &end)
.map_err(|e| GraphError::Storage(format!("storage: {e}")))?;
let mut records = Vec::new();
for (key, bytes) in entries {
if key.ends_with(b":_counter") {
continue;
}
records.push(parse_record(&bytes)?);
}
Ok(records)
}
fn next_id(db: &mut BTree) -> Result<u64, GraphError> {
let key = format!("{LINK_COLLECTION}:_counter").into_bytes();
let current = db
.get(&key)
.map_err(|e| GraphError::Storage(format!("counter: {e}")))?
.and_then(|bytes| String::from_utf8(bytes).ok())
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0);
let id = current
.checked_add(1)
.ok_or_else(|| GraphError::Storage("account link ID space exhausted".into()))?;
db.insert(&key, id.to_string().as_bytes())
.map_err(|e| GraphError::Storage(format!("counter: {e}")))?;
Ok(id)
}
fn schema_key() -> &'static [u8] {
b"_schema:customer_account_graph"
}
pub fn ensure_schema(db: &mut BTree) -> Result<(), String> {
let existing = db.get(schema_key()).map_err(|e| e.to_string())?;
let stored_version = existing
.as_deref()
.and_then(|bytes| std::str::from_utf8(bytes).ok())
.and_then(|text| akurai_json::parse(text).ok())
.and_then(|value| value.get("version").and_then(Value::as_i64));
let marker_name = existing
.as_deref()
.and_then(|bytes| std::str::from_utf8(bytes).ok())
.and_then(|text| akurai_json::parse(text).ok())
.and_then(|value| {
value
.get("name")
.and_then(Value::as_str)
.map(ToOwned::to_owned)
});
if existing.is_some() && marker_name.as_deref() != Some("customer_account_graph") {
return Err("invalid customer account graph schema marker name".into());
}
if existing.is_some() && stored_version.is_none() {
return Err("invalid customer account graph schema marker".into());
}
match stored_version {
Some(version) if version > GRAPH_SCHEMA_VERSION => {
return Err(format!(
"unsupported customer account graph schema version {version}"
));
}
Some(version) if version != 1 && version != GRAPH_SCHEMA_VERSION => {
return Err(format!(
"unknown customer account graph schema version {version}"
));
}
_ => {}
}
if stored_version != Some(GRAPH_SCHEMA_VERSION) {
for value in scan_all_records(db, LINK_COLLECTION).map_err(|e| format!("{e:?}"))? {
let mut fields = object_fields(value).map_err(|e| format!("{e:?}"))?;
let kind = text_field(&Value::Object(fields.clone()), "kind")
.and_then(|kind| LinkKind::parse(&kind));
if let Some(kind) = kind {
if text_field(&Value::Object(fields.clone()), "externalId").is_none() {
if let Some(external) =
text_field(&Value::Object(fields.clone()), canonical_field(kind))
{
set_field(&mut fields, "externalId", Value::Str(external));
}
}
}
if text_field(&Value::Object(fields.clone()), "sourceOwner").is_none() {
if let Some(source) = text_field(&Value::Object(fields.clone()), "source") {
set_field(&mut fields, "sourceOwner", Value::Str(source));
}
}
fields.retain(|(key, _)| {
!matches!(
key.as_str(),
"source"
| "organizationId"
| "workspaceId"
| "subscriptionId"
| "opportunityId"
| "installationId"
)
});
if let Some(id) = int_field(&Value::Object(fields.clone()), "id") {
db.insert(
&build_key(LINK_COLLECTION, id),
Value::Object(fields).to_json().as_bytes(),
)
.map_err(|e| e.to_string())?;
}
}
}
let value = Value::Object(vec![
("name".into(), Value::Str("customer_account_graph".into())),
("version".into(), Value::Int(GRAPH_SCHEMA_VERSION)),
("createdAt".into(), Value::Int(now())),
]);
db.insert(schema_key(), value.to_json().as_bytes())
.map_err(|e| e.to_string())?;
db.commit().map_err(|e| e.to_string())?;
Ok(())
}
fn company_tenant(db: &mut BTree, company_id: u64, tenant_id: &str) -> Result<Value, GraphError> {
let company = get_record(db, "companies", company_id)?
.ok_or_else(|| GraphError::NotFound(format!("companies/{company_id}")))?;
match text_field(&company, "tenantId") {
Some(existing) if existing == tenant_id => Ok(company),
Some(_) => Err(GraphError::Forbidden(
"company belongs to another tenant".into(),
)),
None => Err(GraphError::Forbidden(
"company has no tenant ownership".into(),
)),
}
}
fn bind_or_check_company_tenant(
db: &mut BTree,
company_id: u64,
tenant_id: &str,
) -> Result<Value, GraphError> {
company_tenant(db, company_id, tenant_id)
}
fn validate_external_id(value: &str) -> Result<String, GraphError> {
let id = value.trim();
if id.is_empty() {
return Err(GraphError::BadRequest("externalId is required".into()));
}
if id.len() > 256 || id.chars().any(|c| c.is_control()) {
return Err(GraphError::BadRequest("invalid externalId".into()));
}
Ok(id.to_string())
}
fn validate_safe_input(value: &Value) -> Result<(), GraphError> {
let forbidden = [
"password",
"secret",
"token",
"credential",
"accessToken",
"clientSecret",
"billingState",
"runtimeState",
"identityState",
];
if let Value::Object(fields) = value {
for (key, nested) in fields {
if forbidden.iter().any(|bad| key.eq_ignore_ascii_case(bad)) {
return Err(GraphError::BadRequest(format!(
"{key} is not CRM-owned graph data"
)));
}
validate_safe_input(nested)?;
}
} else if let Value::Array(items) = value {
for item in items {
validate_safe_input(item)?;
}
}
Ok(())
}
fn canonical_field(kind: LinkKind) -> &'static str {
match kind {
LinkKind::Organization => "organizationId",
LinkKind::Workspace => "workspaceId",
LinkKind::Subscription => "subscriptionId",
LinkKind::Opportunity => "opportunityId",
LinkKind::Installation => "installationId",
}
}
fn link_to_value(link: &AccountLink) -> Value {
let mut fields = vec![
("id".into(), Value::Int(link.id as i64)),
("tenantId".into(), Value::Str(link.tenant_id.clone())),
("companyId".into(), Value::Int(link.company_id as i64)),
("kind".into(), Value::Str(link.kind.as_str().into())),
("externalId".into(), Value::Str(link.external_id.clone())),
("sourceOwner".into(), Value::Str(link.source_owner.clone())),
("status".into(), Value::Str(link.status.as_str().into())),
("createdAt".into(), Value::Int(link.created_at)),
("updatedAt".into(), Value::Int(link.updated_at)),
];
if let Some(value) = link.stale_at {
fields.push(("staleAt".into(), Value::Int(value)));
}
if let Some(value) = link.unlinked_at {
fields.push(("unlinkedAt".into(), Value::Int(value)));
}
if let Some(value) = link.merged_into {
fields.push(("mergedInto".into(), Value::Int(value as i64)));
}
Value::Object(fields)
}
fn link_to_response_value(link: &AccountLink) -> Value {
let mut fields = match link_to_value(link) {
Value::Object(fields) => fields,
_ => unreachable!("link serialization is always an object"),
};
fields.push((
canonical_field(link.kind).into(),
Value::Str(link.external_id.clone()),
));
fields.push(("source".into(), Value::Str(link.source_owner.clone())));
Value::Object(fields)
}
fn value_to_link(value: &Value) -> Result<AccountLink, GraphError> {
let id =
int_field(value, "id").ok_or_else(|| GraphError::Storage("graph link has no id".into()))?;
let kind = text_field(value, "kind")
.and_then(|s| LinkKind::parse(&s))
.ok_or_else(|| GraphError::Storage("graph link has invalid kind".into()))?;
let status = match text_field(value, "status").as_deref().unwrap_or("active") {
"active" => LinkStatus::Active,
"stale" => LinkStatus::Stale,
"orphaned" => LinkStatus::Orphaned,
"unlinked" => LinkStatus::Unlinked,
"merged" => LinkStatus::Merged,
_ => return Err(GraphError::Storage("graph link has invalid status".into())),
};
Ok(AccountLink {
id,
tenant_id: text_field(value, "tenantId")
.ok_or_else(|| GraphError::Storage("graph link has no tenant".into()))?,
company_id: int_field(value, "companyId")
.ok_or_else(|| GraphError::Storage("graph link has no company".into()))?,
kind,
external_id: text_field(value, "externalId")
.ok_or_else(|| GraphError::Storage("graph link has no external id".into()))?,
source_owner: text_field(value, "sourceOwner").unwrap_or_else(|| kind.owner().into()),
status,
created_at: value.get("createdAt").and_then(Value::as_i64).unwrap_or(0),
updated_at: value.get("updatedAt").and_then(Value::as_i64).unwrap_or(0),
stale_at: value.get("staleAt").and_then(Value::as_i64),
unlinked_at: value.get("unlinkedAt").and_then(Value::as_i64),
merged_into: int_field(value, "mergedInto"),
})
}
fn read_links_page(
db: &mut BTree,
tenant_id: &str,
company_id: Option<u64>,
limit: usize,
) -> Result<(Vec<AccountLink>, bool), GraphError> {
let values = scan_all_records(db, LINK_COLLECTION)?;
let mut links = Vec::new();
for value in values {
let link = value_to_link(&value)?;
if link.tenant_id == tenant_id && company_id.is_none_or(|id| id == link.company_id) {
links.push(link);
}
}
let has_more = links.len() > limit;
links.truncate(limit);
Ok((links, has_more))
}
fn unique_key(tenant_id: &str, kind: LinkKind, external_id: &str) -> Vec<u8> {
format!(
"account_link_unique:{tenant_id}:{}:{external_id}",
kind.as_str()
)
.into_bytes()
}
fn audit(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
action: &str,
details: Value,
) -> Result<(), GraphError> {
let ts = now();
let key = format!(
"{AUDIT_COLLECTION}:{tenant_id}:{company_id}:{ts}:{}",
random_suffix()
)
.into_bytes();
let value = Value::Object(vec![
("tenantId".into(), Value::Str(tenant_id.into())),
("companyId".into(), Value::Int(company_id as i64)),
("action".into(), Value::Str(action.into())),
("at".into(), Value::Int(ts)),
("details".into(), details),
]);
db.insert(&key, value.to_json().as_bytes())
.map_err(|e| GraphError::Storage(format!("audit: {e}")))?;
Ok(())
}
fn random_suffix() -> u64 {
static AUDIT_SEQUENCE: AtomicU64 = AtomicU64::new(0);
let sequence = AUDIT_SEQUENCE.fetch_add(1, Ordering::Relaxed);
(now() as u64)
.wrapping_mul(1_000_003)
.wrapping_add(sequence)
}
pub fn link_company_resource(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
kind: LinkKind,
external_id: &str,
source_owner: Option<&str>,
) -> Result<Value, GraphError> {
let external_id = validate_external_id(external_id)?;
let owner = source_owner
.unwrap_or(kind.owner())
.trim()
.to_ascii_lowercase();
if owner != kind.owner() {
return Err(GraphError::BadRequest(format!(
"{kind:?} links must be owned by {}",
kind.owner()
)));
}
bind_or_check_company_tenant(db, company_id, tenant_id)?;
let unique = unique_key(tenant_id, kind, &external_id);
if let Some(bytes) = db
.get(&unique)
.map_err(|e| GraphError::Storage(format!("uniqueness: {e}")))?
{
if let Ok(existing_id) = String::from_utf8(bytes)
.ok()
.and_then(|s| s.parse::<u64>().ok())
.ok_or(())
{
if let Some(existing) = get_record(db, LINK_COLLECTION, existing_id)? {
let existing_link = value_to_link(&existing)?;
if matches!(existing_link.status, LinkStatus::Active) {
if existing_link.company_id == company_id {
return Err(GraphError::Conflict("duplicate active account link".into()));
}
return Err(GraphError::Conflict(
"resource is already linked to another company".into(),
));
}
}
}
}
let id = next_id(db)?;
let timestamp = now();
let link = AccountLink {
id,
tenant_id: tenant_id.into(),
company_id,
kind,
external_id,
source_owner: owner,
status: LinkStatus::Active,
created_at: timestamp,
updated_at: timestamp,
stale_at: None,
unlinked_at: None,
merged_into: None,
};
let value = link_to_value(&link);
db.insert(&build_key(LINK_COLLECTION, id), value.to_json().as_bytes())
.map_err(|e| GraphError::Storage(format!("link: {e}")))?;
db.insert(&unique, id.to_string().as_bytes())
.map_err(|e| GraphError::Storage(format!("uniqueness: {e}")))?;
audit(db, tenant_id, company_id, "link", value.clone())?;
record_timeline(db, "companies", company_id, "account_linked", None)
.map_err(GraphError::Storage)?;
db.commit()
.map_err(|e| GraphError::Storage(format!("commit: {e}")))?;
Ok(link_to_response_value(&link))
}
pub fn unlink_company_resource(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
link_id: u64,
) -> Result<Value, GraphError> {
company_tenant(db, company_id, tenant_id)?;
let value = get_record(db, LINK_COLLECTION, link_id)?
.ok_or_else(|| GraphError::NotFound(format!("account link {link_id}")))?;
let link = value_to_link(&value)?;
if link.tenant_id != tenant_id || link.company_id != company_id {
return Err(GraphError::Forbidden(
"account link belongs to another tenant or company".into(),
));
}
if !matches!(link.status, LinkStatus::Active | LinkStatus::Stale) {
return Err(GraphError::Conflict(
"account link is already closed".into(),
));
}
let timestamp = now();
let mut fields = object_fields(value.clone())?;
set_field(
&mut fields,
"status",
Value::Str(LinkStatus::Unlinked.as_str().into()),
);
set_field(&mut fields, "unlinkedAt", Value::Int(timestamp));
set_field(&mut fields, "updatedAt", Value::Int(timestamp));
let updated = Value::Object(fields);
db.insert(
&build_key(LINK_COLLECTION, link_id),
updated.to_json().as_bytes(),
)
.map_err(|e| GraphError::Storage(format!("unlink: {e}")))?;
let _ = db.delete(&unique_key(tenant_id, link.kind, &link.external_id));
audit(db, tenant_id, company_id, "unlink", updated.clone())?;
record_timeline(db, "companies", company_id, "account_unlinked", None)
.map_err(GraphError::Storage)?;
db.commit()
.map_err(|e| GraphError::Storage(format!("commit: {e}")))?;
Ok(link_to_response_value(&value_to_link(&updated)?))
}
pub fn mark_link_stale(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
link_id: u64,
) -> Result<Value, GraphError> {
let value = get_record(db, LINK_COLLECTION, link_id)?
.ok_or_else(|| GraphError::NotFound(format!("account link {link_id}")))?;
let link = value_to_link(&value)?;
if link.tenant_id != tenant_id || link.company_id != company_id {
return Err(GraphError::Forbidden(
"account link belongs to another tenant or company".into(),
));
}
if link.status != LinkStatus::Active {
return Err(GraphError::Conflict(
"only active links can become stale".into(),
));
}
let timestamp = now();
let mut fields = object_fields(value.clone())?;
set_field(
&mut fields,
"status",
Value::Str(LinkStatus::Stale.as_str().into()),
);
set_field(&mut fields, "staleAt", Value::Int(timestamp));
set_field(&mut fields, "updatedAt", Value::Int(timestamp));
let updated = Value::Object(fields);
db.insert(
&build_key(LINK_COLLECTION, link_id),
updated.to_json().as_bytes(),
)
.map_err(|e| GraphError::Storage(format!("stale: {e}")))?;
let _ = db.delete(&unique_key(tenant_id, link.kind, &link.external_id));
audit(db, tenant_id, company_id, "stale", updated.clone())?;
db.commit()
.map_err(|e| GraphError::Storage(format!("commit: {e}")))?;
Ok(link_to_response_value(&value_to_link(&updated)?))
}
pub fn mark_company_links_orphaned(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
) -> Result<(), String> {
company_tenant(db, company_id, tenant_id).map_err(|e| format!("{e:?}"))?;
let values = scan_all_records(db, LINK_COLLECTION).map_err(|e| format!("{e:?}"))?;
let timestamp = now();
let mut changed = false;
for value in values {
let link = value_to_link(&value).map_err(|e| format!("{e:?}"))?;
if link.tenant_id != tenant_id
|| link.company_id != company_id
|| !matches!(link.status, LinkStatus::Active | LinkStatus::Stale)
{
continue;
}
let mut fields = object_fields(value).map_err(|e| format!("{e:?}"))?;
set_field(
&mut fields,
"status",
Value::Str(LinkStatus::Orphaned.as_str().into()),
);
set_field(&mut fields, "updatedAt", Value::Int(timestamp));
db.insert(
&build_key(LINK_COLLECTION, link.id),
Value::Object(fields).to_json().as_bytes(),
)
.map_err(|e| format!("orphan: {e}"))?;
let _ = db.delete(&unique_key(&link.tenant_id, link.kind, &link.external_id));
changed = true;
}
if changed {
db.commit().map_err(|e| format!("orphan commit: {e}"))?;
}
Ok(())
}
fn related_records(
db: &mut BTree,
collection: &str,
tenant_id: &str,
company_id: u64,
limit: usize,
) -> Result<(Vec<Value>, bool), GraphError> {
let mut records = scan_all_records(db, collection)?
.into_iter()
.filter(|value| {
int_field(value, "companyId") == Some(company_id)
&& text_field(value, "tenantId").as_deref() == Some(tenant_id)
})
.collect::<Vec<_>>();
let has_more = records.len() > limit;
records.truncate(limit);
Ok((records, has_more))
}
pub fn get_company_graph(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
limit: usize,
) -> Result<Value, GraphError> {
let company = company_tenant(db, company_id, tenant_id)?;
let limit = limit.clamp(1, MAX_GRAPH_LIMIT);
let (links, links_more) = read_links_page(db, tenant_id, Some(company_id), limit)?;
let (contacts, contacts_more) = related_records(db, "people", tenant_id, company_id, limit)?;
let (opportunities, opportunities_more) =
related_records(db, "opportunities", tenant_id, company_id, limit)?;
let (notes, notes_more) = related_records(db, "notes", tenant_id, company_id, limit)?;
let (tasks, tasks_more) = related_records(db, "tasks", tenant_id, company_id, limit)?;
let mut stale = Vec::new();
let mut orphaned = Vec::new();
for link in &links {
match link.status {
LinkStatus::Stale => stale.push(link_to_response_value(link)),
LinkStatus::Orphaned => orphaned.push(link_to_response_value(link)),
_ => {}
}
}
let (history, history_more) = read_audit(db, tenant_id, company_id, limit)?;
let truncated = links_more
|| contacts_more
|| opportunities_more
|| notes_more
|| tasks_more
|| history_more;
Ok(Value::Object(vec![
("company".into(), company),
("contacts".into(), Value::Array(contacts)),
("opportunities".into(), Value::Array(opportunities)),
("notes".into(), Value::Array(notes)),
("tasks".into(), Value::Array(tasks)),
(
"links".into(),
Value::Array(links.iter().map(link_to_response_value).collect()),
),
("staleLinks".into(), Value::Array(stale)),
("orphanedLinks".into(), Value::Array(orphaned)),
("mergeHistory".into(), Value::Array(history)),
("bounded".into(), Value::Bool(true)),
("limit".into(), Value::Int(limit as i64)),
("truncated".into(), Value::Bool(truncated)),
("complete".into(), Value::Bool(!truncated)),
]))
}
fn read_audit(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
limit: usize,
) -> Result<(Vec<Value>, bool), GraphError> {
let prefix = format!("{AUDIT_COLLECTION}:{tenant_id}:{company_id}:").into_bytes();
let end = upper_bound(&prefix);
let entries = db
.range(&prefix, &end)
.map_err(|e| GraphError::Storage(format!("audit: {e}")))?;
let mut records = entries
.into_iter()
.rev()
.map(|(_, bytes)| parse_record(&bytes))
.collect::<Result<Vec<_>, _>>()?;
let has_more = records.len() > limit;
records.truncate(limit);
Ok((records, has_more))
}
pub fn list_links(
db: &mut BTree,
tenant_id: &str,
company_id: u64,
status: Option<&str>,
limit: usize,
) -> Result<Value, GraphError> {
company_tenant(db, company_id, tenant_id)?;
let limit = limit.clamp(1, MAX_GRAPH_LIMIT);
let (links, _) = read_links_page(db, tenant_id, Some(company_id), usize::MAX)?;
let filtered = links
.into_iter()
.filter(|link| status.is_none_or(|wanted| link.status.as_str() == wanted))
.collect::<Vec<_>>();
let truncated = filtered.len() > limit;
let values = filtered
.into_iter()
.take(limit)
.map(|link| link_to_response_value(&link))
.collect();
Ok(Value::Object(vec![
("links".into(), Value::Array(values)),
("bounded".into(), Value::Bool(true)),
("limit".into(), Value::Int(limit as i64)),
("truncated".into(), Value::Bool(truncated)),
("complete".into(), Value::Bool(!truncated)),
]))
}
pub fn list_orphaned_links(
db: &mut BTree,
tenant_id: &str,
limit: usize,
) -> Result<Value, GraphError> {
let limit = limit.clamp(1, MAX_GRAPH_LIMIT);
let links = scan_all_records(db, LINK_COLLECTION)?
.into_iter()
.map(|value| value_to_link(&value))
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.filter(|link| link.tenant_id == tenant_id && link.status == LinkStatus::Orphaned)
.collect::<Vec<_>>();
let truncated = links.len() > limit;
let values = links
.into_iter()
.take(limit)
.map(|link| link_to_response_value(&link))
.collect();
Ok(Value::Object(vec![
("links".into(), Value::Array(values)),
("bounded".into(), Value::Bool(true)),
("limit".into(), Value::Int(limit as i64)),
("truncated".into(), Value::Bool(truncated)),
("complete".into(), Value::Bool(!truncated)),
]))
}
pub fn merge_companies(
db: &mut BTree,
tenant_id: &str,
source_id: u64,
target_id: u64,
) -> Result<Value, GraphError> {
if source_id == target_id {
return Err(GraphError::BadRequest(
"source and target companies must differ".into(),
));
}
let source = bind_or_check_company_tenant(db, source_id, tenant_id)?;
let target = company_tenant(db, target_id, tenant_id)?;
if text_field(&source, "status").as_deref() == Some("merged") {
return Err(GraphError::Conflict(
"source company is already merged".into(),
));
}
if text_field(&target, "mergedInto").is_some()
|| text_field(&target, "status").as_deref() == Some("merged")
{
return Err(GraphError::Conflict(
"target company is already merged".into(),
));
}
let timestamp = now();
let source_links = scan_all_records(db, LINK_COLLECTION)?
.into_iter()
.map(|value| value_to_link(&value))
.collect::<Result<Vec<_>, _>>()?
.into_iter()
.filter(|link| link.tenant_id == tenant_id && link.company_id == source_id)
.collect::<Vec<_>>();
let mut moved = Vec::new();
let mut duplicates = Vec::new();
for link in source_links {
let unique = unique_key(tenant_id, link.kind, &link.external_id);
let conflicting = if matches!(link.status, LinkStatus::Active | LinkStatus::Stale) {
db.get(&unique)
.ok()
.flatten()
.and_then(|bytes| String::from_utf8(bytes).ok())
.and_then(|s| s.parse::<u64>().ok())
.filter(|id| *id != link.id)
} else {
None
};
if let Some(conflicting) = conflicting {
let mut fields = object_fields(link_to_value(&link))?;
set_field(
&mut fields,
"status",
Value::Str(LinkStatus::Merged.as_str().into()),
);
set_field(&mut fields, "companyId", Value::Int(target_id as i64));
set_field(&mut fields, "mergedInto", Value::Int(conflicting as i64));
set_field(&mut fields, "updatedAt", Value::Int(timestamp));
db.insert(
&build_key(LINK_COLLECTION, link.id),
Value::Object(fields).to_json().as_bytes(),
)
.map_err(|e| GraphError::Storage(format!("merge duplicate: {e}")))?;
duplicates.push(link.id);
continue;
}
let mut fields = object_fields(link_to_value(&link))?;
set_field(&mut fields, "companyId", Value::Int(target_id as i64));
set_field(&mut fields, "updatedAt", Value::Int(timestamp));
db.insert(
&build_key(LINK_COLLECTION, link.id),
Value::Object(fields).to_json().as_bytes(),
)
.map_err(|e| GraphError::Storage(format!("merge link: {e}")))?;
if link.status == LinkStatus::Active {
db.insert(&unique, link.id.to_string().as_bytes())
.map_err(|e| GraphError::Storage(format!("merge uniqueness: {e}")))?;
}
moved.push(link.id);
}
let mut moved_records = Vec::new();
for collection in ["people", "opportunities", "notes", "tasks"] {
let count =
move_company_records(db, collection, source_id, target_id, tenant_id, timestamp)?;
moved_records.push(Value::Object(vec![
("collection".into(), Value::Str(collection.into())),
("count".into(), Value::Int(count as i64)),
]));
}
let mut source_fields = object_fields(source)?;
set_field(&mut source_fields, "status", Value::Str("merged".into()));
set_field(
&mut source_fields,
"mergedInto",
Value::Int(target_id as i64),
);
set_field(&mut source_fields, "mergedAt", Value::Int(timestamp));
set_field(&mut source_fields, "updatedAt", Value::Int(timestamp));
db.insert(
&build_key("companies", source_id),
Value::Object(source_fields).to_json().as_bytes(),
)
.map_err(|e| GraphError::Storage(format!("merge company: {e}")))?;
let history = Value::Object(vec![
("sourceCompanyId".into(), Value::Int(source_id as i64)),
("targetCompanyId".into(), Value::Int(target_id as i64)),
(
"movedLinkIds".into(),
Value::Array(moved.iter().map(|id| Value::Int(*id as i64)).collect()),
),
(
"duplicateLinkIds".into(),
Value::Array(duplicates.iter().map(|id| Value::Int(*id as i64)).collect()),
),
("movedRecords".into(), Value::Array(moved_records.clone())),
("complete".into(), Value::Bool(true)),
("mergedAt".into(), Value::Int(timestamp)),
]);
audit(db, tenant_id, target_id, "merge_target", history.clone())?;
audit(db, tenant_id, source_id, "merge_source", history.clone())?;
record_timeline(db, "companies", source_id, "merged", None).map_err(GraphError::Storage)?;
record_timeline(db, "companies", target_id, "merge_received", None)
.map_err(GraphError::Storage)?;
db.commit()
.map_err(|e| GraphError::Storage(format!("merge commit: {e}")))?;
Ok(Value::Object(vec![
("merged".into(), Value::Bool(true)),
("sourceCompanyId".into(), Value::Int(source_id as i64)),
("targetCompanyId".into(), Value::Int(target_id as i64)),
(
"movedLinkIds".into(),
Value::Array(moved.into_iter().map(|id| Value::Int(id as i64)).collect()),
),
(
"duplicateLinkIds".into(),
Value::Array(
duplicates
.into_iter()
.map(|id| Value::Int(id as i64))
.collect(),
),
),
("movedRecords".into(), Value::Array(moved_records)),
("complete".into(), Value::Bool(true)),
("history".into(), history),
]))
}
fn move_company_records(
db: &mut BTree,
collection: &str,
source_id: u64,
target_id: u64,
tenant_id: &str,
timestamp: i64,
) -> Result<usize, GraphError> {
let values = scan_all_records(db, collection)?;
let mut moved = 0;
for value in values {
if int_field(&value, "companyId") != Some(source_id)
|| text_field(&value, "tenantId").is_some_and(|t| t != tenant_id)
{
continue;
}
let id = int_field(&value, "id")
.ok_or_else(|| GraphError::Storage(format!("{collection} record has no id")))?;
let mut fields = object_fields(value)?;
set_field(&mut fields, "companyId", Value::Int(target_id as i64));
set_field(&mut fields, "tenantId", Value::Str(tenant_id.into()));
set_field(&mut fields, "updatedAt", Value::Int(timestamp));
db.insert(
&build_key(collection, id),
Value::Object(fields).to_json().as_bytes(),
)
.map_err(|e| GraphError::Storage(format!("merge {collection}: {e}")))?;
moved += 1;
}
Ok(moved)
}
fn parse_limit(req: &Request) -> usize {
req.query
.as_deref()
.unwrap_or("")
.split('&')
.find_map(|part| {
let (key, value) = part.split_once('=')?;
(key == "limit")
.then(|| value.parse::<usize>().ok())
.flatten()
})
.unwrap_or(50)
.clamp(1, MAX_GRAPH_LIMIT)
}
pub fn authorized_tenant(req: &Request, db: &mut BTree) -> Result<String, GraphError> {
authorized_user(req, db).map(|user| user.tenant_id)
}
pub const ADMIN_REQUIRED: &str = "this action requires an AkurAI platform administrator";
/// Tenant of a session whose organization role is owner or admin.
pub fn authorized_admin(req: &Request, db: &mut BTree) -> Result<String, GraphError> {
let user = authorized_user(req, db)?;
if !user.role.is_admin() {
return Err(GraphError::Forbidden(ADMIN_REQUIRED.into()));
}
Ok(user.tenant_id)
}
fn authorized_user(req: &Request, db: &mut BTree) -> Result<session::User, GraphError> {
let session_id = session::extract_session_id(req)
.ok_or_else(|| GraphError::Unauthorized("authenticated session is required".into()))?;
let user = session::get_user(db, &session_id)
.ok_or_else(|| GraphError::Unauthorized("invalid or expired session".into()))?;
if user.sub.is_empty() || user.tenant_id.is_empty() {
return Err(GraphError::Forbidden(
"session has no authorized tenant".into(),
));
}
if let Some(header_tenant) = req
.header("X-Tenant-Id")
.or_else(|| req.header("X-Tenant-ID"))
.map(str::trim)
.filter(|value| !value.is_empty())
{
if header_tenant != user.tenant_id {
return Err(GraphError::Forbidden(
"tenant header does not match session".into(),
));
}
}
Ok(user)
}
pub fn authorize_request(
req: &Request,
db: &mut BTree,
company_id: u64,
) -> Result<String, GraphError> {
let tenant = authorized_tenant(req, db)?;
company_tenant(db, company_id, &tenant)?;
Ok(tenant)
}
pub fn graph_route(state: Arc<Mutex<CrmState>>) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
Box::new(move |req| {
let company_id = extract_ids(&req.path)
.first()
.copied()
.ok_or_else(|| GraphError::BadRequest("missing company id".into()));
let company_id = match company_id {
Ok(id) => id,
Err(e) => return e.response(),
};
let state = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let mut db = match state.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let tenant = match authorize_request(req, &mut db, company_id) {
Ok(t) => t,
Err(e) => return e.response(),
};
match get_company_graph(&mut db, &tenant, company_id, parse_limit(req)) {
Ok(value) => json_response(value),
Err(e) => e.response(),
}
})
}
fn body_link_spec(body: &Value) -> Option<(LinkKind, String)> {
if let Some(kind) = text_field(body, "kind").and_then(|value| LinkKind::parse(&value)) {
let id =
text_field(body, "externalId").or_else(|| text_field(body, canonical_field(kind)))?;
return Some((kind, id));
}
for kind in [
LinkKind::Organization,
LinkKind::Workspace,
LinkKind::Subscription,
LinkKind::Opportunity,
LinkKind::Installation,
] {
if let Some(id) = text_field(body, canonical_field(kind)) {
return Some((kind, id));
}
}
None
}
pub fn links_route(state: Arc<Mutex<CrmState>>) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
Box::new(move |req| {
let company_id = match extract_ids(&req.path).first().copied() {
Some(id) => id,
None => return bad_request("missing company id"),
};
let state = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let mut db = match state.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let tenant = match authorize_request(req, &mut db, company_id) {
Ok(t) => t,
Err(e) => return e.response(),
};
match list_links(
&mut db,
&tenant,
company_id,
query_value(req, "status").as_deref(),
parse_limit(req),
) {
Ok(value) => json_response(value),
Err(e) => e.response(),
}
})
}
pub fn link_create_route(
state: Arc<Mutex<CrmState>>,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
Box::new(move |req| {
let company_id = match extract_ids(&req.path).first().copied() {
Some(id) => id,
None => return bad_request("missing company id"),
};
let body = match parse_body(req) {
Ok(value) => value,
Err(e) => return e.response(),
};
if let Err(e) = validate_safe_input(&body) {
return e.response();
}
let (kind, external_id) = match body_link_spec(&body) {
Some(spec) => spec,
None => {
return bad_request("kind and externalId (or a canonical resource ID) are required")
}
};
let state = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let mut db = match state.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let tenant = match authorize_request(req, &mut db, company_id) {
Ok(t) => t,
Err(e) => return e.response(),
};
let source_owner = text_field(&body, "sourceOwner").or_else(|| text_field(&body, "source"));
match link_company_resource(
&mut db,
&tenant,
company_id,
kind,
&external_id,
source_owner.as_deref(),
) {
Ok(value) => Response::new(201)
.with_header("Content-Type", "application/json")
.with_body("application/json", value.to_json().into_bytes()),
Err(e) => e.response(),
}
})
}
pub fn link_action_route(
state: Arc<Mutex<CrmState>>,
action: &'static str,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
Box::new(move |req| {
let ids = extract_ids(&req.path);
let company_id = ids.first().copied().unwrap_or(0);
let link_id = ids.get(1).copied().unwrap_or(0);
if company_id == 0 || link_id == 0 {
return bad_request("missing company or link id");
}
let state = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let mut db = match state.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let tenant = match authorize_request(req, &mut db, company_id) {
Ok(t) => t,
Err(e) => return e.response(),
};
let result = match action {
"unlink" => unlink_company_resource(&mut db, &tenant, company_id, link_id),
"stale" => mark_link_stale(&mut db, &tenant, company_id, link_id),
_ => Err(GraphError::BadRequest("unsupported link action".into())),
};
match result {
Ok(value) => json_response(value),
Err(e) => e.response(),
}
})
}
pub fn merge_route(state: Arc<Mutex<CrmState>>) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
Box::new(move |req| {
let source_id = match extract_ids(&req.path).first().copied() {
Some(id) => id,
None => return bad_request("missing source company id"),
};
let body = match parse_body(req) {
Ok(value) => value,
Err(e) => return e.response(),
};
let target_id = match int_field(&body, "targetCompanyId") {
Some(id) => id,
None => return bad_request("targetCompanyId is required"),
};
let state = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let mut db = match state.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let tenant = match authorized_admin(req, &mut db) {
Ok(t) => t,
Err(e) => return e.response(),
};
for company_id in [source_id, target_id] {
if let Err(e) = company_tenant(&mut db, company_id, &tenant) {
return e.response();
}
}
match merge_companies(&mut db, &tenant, source_id, target_id) {
Ok(value) => json_response(value),
Err(e) => e.response(),
}
})
}
pub fn orphaned_route(
state: Arc<Mutex<CrmState>>,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
Box::new(move |req| {
let state = match state.lock() {
Ok(s) => s,
Err(_) => return internal_error("lock poisoned"),
};
let mut db = match state.db.lock() {
Ok(db) => db,
Err(_) => return internal_error("db lock poisoned"),
};
let tenant = match authorized_tenant(req, &mut db) {
Ok(t) => t,
Err(e) => return e.response(),
};
match list_orphaned_links(&mut db, &tenant, parse_limit(req)) {
Ok(value) => json_response(value),
Err(e) => e.response(),
}
})
}
fn query_value(req: &Request, wanted: &str) -> Option<String> {
req.query
.as_deref()
.unwrap_or("")
.split('&')
.find_map(|part| {
let (key, value) = part.split_once('=')?;
(key == wanted).then(|| value.to_string())
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::Path;
fn db(name: &str) -> BTree {
let path = format!("/tmp/akurai-crm-graph-{name}-{}.db", std::process::id());
let _ = std::fs::remove_file(Path::new(&path));
BTree::open(&path).unwrap()
}
fn company(db: &mut BTree, id: u64, tenant: &str) {
let value = Value::Object(vec![
("id".into(), Value::Int(id as i64)),
("name".into(), Value::Str(format!("Company {id}"))),
("tenantId".into(), Value::Str(tenant.into())),
]);
db.insert(&build_key("companies", id), value.to_json().as_bytes())
.unwrap();
}
#[test]
fn links_are_tenant_scoped_and_unique() {
let mut db = db("unique");
company(&mut db, 1, "t1");
company(&mut db, 2, "t2");
let first =
link_company_resource(&mut db, "t1", 1, LinkKind::Organization, "org_1", None).unwrap();
assert_eq!(text_field(&first, "status").as_deref(), Some("active"));
assert!(matches!(
link_company_resource(&mut db, "t1", 1, LinkKind::Organization, "org_1", None),
Err(GraphError::Conflict(_))
));
assert!(
link_company_resource(&mut db, "t2", 2, LinkKind::Organization, "org_1", None).is_ok()
);
}
#[test]
fn merge_preserves_history_and_marks_source() {
let mut db = db("merge");
company(&mut db, 1, "t1");
company(&mut db, 2, "t1");
link_company_resource(&mut db, "t1", 1, LinkKind::Workspace, "ws_1", None).unwrap();
let merged = merge_companies(&mut db, "t1", 1, 2).unwrap();
assert_eq!(merged.get("merged").and_then(Value::as_bool), Some(true));
let source = get_record(&mut db, "companies", 1).unwrap().unwrap();
assert_eq!(text_field(&source, "status").as_deref(), Some("merged"));
let graph = get_company_graph(&mut db, "t1", 2, 10).unwrap();
assert_eq!(
graph.get("links").and_then(|v| match v {
Value::Array(items) => Some(items.len()),
_ => None,
}),
Some(1)
);
}
#[test]
fn stale_and_orphaned_states_remain_visible() {
let mut db = db("stale");
company(&mut db, 1, "t1");
let link = link_company_resource(&mut db, "t1", 1, LinkKind::Installation, "inst_1", None)
.unwrap();
let link_id = int_field(&link, "id").unwrap();
mark_link_stale(&mut db, "t1", 1, link_id).unwrap();
let links = list_links(&mut db, "t1", 1, Some("stale"), 10).unwrap();
assert!(links.to_json().contains("stale"));
mark_company_links_orphaned(&mut db, "t1", 1).unwrap();
assert!(list_orphaned_links(&mut db, "t1", 10)
.unwrap()
.to_json()
.contains("orphaned"));
}
#[test]
fn merge_moves_all_link_and_activity_records_without_truncation() {
let mut db = db("large-merge");
company(&mut db, 1, "t1");
company(&mut db, 2, "t1");
for id in 1..=1_201_u64 {
let link = Value::Object(vec![
("id".into(), Value::Int(id as i64)),
("tenantId".into(), Value::Str("t1".into())),
("companyId".into(), Value::Int(1)),
("kind".into(), Value::Str("organization".into())),
("externalId".into(), Value::Str(format!("org-{id}"))),
("sourceOwner".into(), Value::Str("idp".into())),
("status".into(), Value::Str("active".into())),
("createdAt".into(), Value::Int(1)),
("updatedAt".into(), Value::Int(1)),
]);
db.insert(&build_key(LINK_COLLECTION, id), link.to_json().as_bytes())
.unwrap();
}
for collection in ["people", "opportunities", "notes", "tasks"] {
let value = Value::Object(vec![
("id".into(), Value::Int(1)),
("tenantId".into(), Value::Str("t1".into())),
("companyId".into(), Value::Int(1)),
]);
db.insert(&build_key(collection, 1), value.to_json().as_bytes())
.unwrap();
}
let merged = merge_companies(&mut db, "t1", 1, 2).unwrap();
assert_eq!(merged.get("complete").and_then(Value::as_bool), Some(true));
assert_eq!(
merged.get("movedLinkIds").and_then(|value| match value {
Value::Array(items) => Some(items.len()),
_ => None,
}),
Some(1_201)
);
for collection in ["people", "opportunities", "notes", "tasks"] {
let record = get_record(&mut db, collection, 1).unwrap().unwrap();
assert_eq!(int_field(&record, "companyId"), Some(2));
assert_eq!(text_field(&record, "tenantId").as_deref(), Some("t1"));
}
}
#[test]
fn schema_marker_migrates_legacy_response_fields() {
let mut db = db("schema-migration");
let marker = Value::Object(vec![
("name".into(), Value::Str("customer_account_graph".into())),
("version".into(), Value::Int(1)),
]);
db.insert(schema_key(), marker.to_json().as_bytes())
.unwrap();
let legacy = Value::Object(vec![
("id".into(), Value::Int(1)),
("tenantId".into(), Value::Str("t1".into())),
("companyId".into(), Value::Int(1)),
("kind".into(), Value::Str("organization".into())),
("organizationId".into(), Value::Str("org-1".into())),
("source".into(), Value::Str("idp".into())),
("status".into(), Value::Str("active".into())),
]);
db.insert(&build_key(LINK_COLLECTION, 1), legacy.to_json().as_bytes())
.unwrap();
ensure_schema(&mut db).unwrap();
let migrated = get_record(&mut db, LINK_COLLECTION, 1).unwrap().unwrap();
assert_eq!(
text_field(&migrated, "externalId").as_deref(),
Some("org-1")
);
assert_eq!(text_field(&migrated, "sourceOwner").as_deref(), Some("idp"));
assert!(migrated.get("organizationId").is_none());
assert!(migrated.get("source").is_none());
}
#[test]
fn cross_tenant_reads_and_unlinks_are_forbidden() {
let mut db = db("authorization");
company(&mut db, 1, "tenant-a");
let link = link_company_resource(
&mut db,
"tenant-a",
1,
LinkKind::Subscription,
"sub_1",
None,
)
.unwrap();
let link_id = int_field(&link, "id").unwrap();
assert!(matches!(
get_company_graph(&mut db, "tenant-b", 1, 10),
Err(GraphError::Forbidden(_))
));
assert!(matches!(
unlink_company_resource(&mut db, "tenant-b", 1, link_id),
Err(GraphError::Forbidden(_))
));
}
#[test]
fn nested_graph_route_reads_company_id_and_tenant_header() {
let path = format!("/tmp/akurai-crm-graph-route-{}.db", std::process::id());
let _ = std::fs::remove_file(&path);
let state = CrmState::new(&path, std::path::PathBuf::from("site/frontend")).unwrap();
let session_id = {
let mut db = match state.db.lock() {
Ok(db) => db,
Err(_) => panic!("db lock poisoned"),
};
company(&mut db, 1, "tenant-a");
session::create(&mut db, "user-a", "user@example.test", "tenant-a").unwrap()
};
let request = format!(
"GET /api/companies/1/graph HTTP/1.1\r\nX-Tenant-Id: tenant-a\r\nCookie: akurai_session={session_id}\r\n\r\n"
);
let req = Request::parse_head(request.as_bytes()).unwrap();
let response = graph_route(Arc::new(Mutex::new(state)))(&req);
assert_eq!(response.status, 200);
}
#[test]
fn graph_header_cannot_override_session_tenant() {
let path = format!("/tmp/akurai-crm-graph-auth-{}.db", std::process::id());
let _ = std::fs::remove_file(&path);
let state = CrmState::new(&path, std::path::PathBuf::from("site/frontend")).unwrap();
let session_id = {
let mut db = state.db.lock().unwrap();
company(&mut db, 1, "tenant-a");
session::create(&mut db, "user-a", "user@example.test", "tenant-a").unwrap()
};
let request = format!(
"GET /api/companies/1/graph HTTP/1.1\r\nX-Tenant-Id: tenant-b\r\nCookie: akurai_session={session_id}\r\n\r\n"
);
let response = graph_route(Arc::new(Mutex::new(state)))(
&Request::parse_head(request.as_bytes()).unwrap(),
);
assert_eq!(response.status, 403);
}
fn merge_as(role: session::Role, name: &str) -> Response {
let path = crate::auth::test_db_path(name);
let state = CrmState::new(&path, std::path::PathBuf::from("site/frontend")).unwrap();
let session_id = {
let mut db = state.db.lock().unwrap();
company(&mut db, 1, "tenant-a");
company(&mut db, 2, "tenant-a");
session::create_with_role(&mut db, "user-a", "u@x", "tenant-a", role).unwrap()
};
let head = format!(
"POST /api/companies/1/merge HTTP/1.1\r\nCookie: akurai_session={session_id}\r\n\r\n"
);
let mut req = Request::parse_head(head.as_bytes()).unwrap();
req.body = br#"{"targetCompanyId":2}"#.to_vec();
merge_route(Arc::new(Mutex::new(state)))(&req)
}
#[test]
fn merge_requires_admin_role() {
let response = merge_as(session::Role::Member, "merge-member");
assert_eq!(response.status, 403);
assert!(String::from_utf8_lossy(&response.body).contains(ADMIN_REQUIRED));
}
#[test]
fn merge_is_allowed_for_owner() {
let response = merge_as(session::Role::Owner, "merge-owner");
assert_eq!(
response.status,
200,
"{}",
String::from_utf8_lossy(&response.body)
);
}
}