AkurAI-CRM / crates/crm-api/src/handlers/crud.rs

Latest change 201b04d27b0773782f4a1a0558cee777ec8e3462 - Recover deployed web login and MCP group gate by Ólafur Búi Ólafsson

use crate::graph;
use crate::handlers::timeline::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::{Arc, Mutex};

fn parse_body(req: &Request) -> Result<Value, String> {
    if req.body.is_empty() {
        return Err("empty request body".into());
    }
    let text = std::str::from_utf8(&req.body).map_err(|e| format!("body not UTF-8: {e}"))?;
    akurai_json::parse(text).map_err(|e| format!("invalid JSON: {e}"))
}

fn extract_id(path: &str) -> Option<u64> {
    path.split('/')
        .next_back()
        .and_then(|s| s.parse::<u64>().ok())
}

fn collection_name(entity: &str) -> &str {
    match entity {
        "people" | "companies" | "opportunities" | "tasks" | "notes" => entity,
        _ => entity,
    }
}

fn build_key(coll_name: &str, id: u64) -> Vec<u8> {
    let mut key = format!("{coll_name}:").into_bytes();
    key.extend_from_slice(&id.to_be_bytes());
    key
}

fn authorize(req: &Request, db: &mut BTree) -> Result<String, Response> {
    graph::authorized_tenant(req, db).map_err(|error| error.response())
}

fn parse_record(bytes: &[u8]) -> Result<Value, Response> {
    akurai_json::parse(std::str::from_utf8(bytes).unwrap_or(""))
        .map_err(|_| internal_error("corrupted data"))
}

fn tenant_owned(value: &Value, tenant: &str) -> bool {
    value.get("tenantId").and_then(Value::as_str) == Some(tenant)
}

fn validate_reference(
    db: &mut BTree,
    field: &str,
    value: &Value,
    tenant: &str,
) -> Result<(), Response> {
    let collection = match field {
        "companyId" => "companies",
        "personId" => "people",
        "opportunityId" => "opportunities",
        "taskId" => "tasks",
        "noteId" => "notes",
        _ => return Ok(()),
    };
    let id = value
        .as_i64()
        .and_then(|id| u64::try_from(id).ok())
        .ok_or_else(|| bad_request(&format!("{field} must be a positive integer")))?;
    let record = db
        .get(&build_key(collection, id))
        .map_err(|e| internal_error(&format!("storage: {e}")))?
        .ok_or_else(|| bad_request(&format!("{field} references a missing record")))?;
    let record = parse_record(&record)?;
    if !tenant_owned(&record, tenant) {
        return Err(error_response(
            403,
            "related record belongs to another tenant",
        ));
    }
    Ok(())
}

fn validate_references(
    db: &mut BTree,
    fields: &[(String, Value)],
    tenant: &str,
) -> Result<(), Response> {
    for (field, value) in fields {
        validate_reference(db, field, value, tenant)?;
    }
    Ok(())
}

pub fn list_route(
    state: Arc<Mutex<CrmState>>,
    entity: &'static str,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
    Box::new(move |req: &Request| {
        let state = match state.lock() {
            Ok(s) => s,
            Err(_) => return internal_error("lock poisoned"),
        };
        let coll_name = collection_name(entity);
        let mut db = match state.db.lock() {
            Ok(db) => db,
            Err(_) => return internal_error("db lock poisoned"),
        };
        let tenant = match authorize(req, &mut db) {
            Ok(tenant) => tenant,
            Err(response) => return response,
        };
        let prefix = format!("{coll_name}:").into_bytes();
        let end = upper_bound(&prefix);
        let mut records = Vec::new();
        let entries = match db.range(&prefix, &end) {
            Ok(entries) => entries,
            Err(e) => return internal_error(&format!("storage: {e}")),
        };
        for (key, value) in entries {
            if key.ends_with(b":_counter") {
                continue;
            }
            let value = match parse_record(&value) {
                Ok(value) => value,
                Err(response) => return response,
            };
            if tenant_owned(&value, &tenant) {
                records.push(value);
            }
        }
        json_response(Value::Array(records))
    })
}

pub fn get_route(
    state: Arc<Mutex<CrmState>>,
    entity: &'static str,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
    Box::new(move |req: &Request| {
        let id = match extract_id(&req.path) {
            Some(id) => id,
            None => return bad_request("missing id"),
        };
        let state = match state.lock() {
            Ok(s) => s,
            Err(_) => return internal_error("lock poisoned"),
        };
        let coll_name = collection_name(entity);
        let mut db = match state.db.lock() {
            Ok(db) => db,
            Err(_) => return internal_error("db lock poisoned"),
        };
        let tenant = match authorize(req, &mut db) {
            Ok(tenant) => tenant,
            Err(response) => return response,
        };
        let key = build_key(coll_name, id);
        match db.get(&key) {
            Ok(Some(val)) => {
                let json = match parse_record(&val) {
                    Ok(value) => value,
                    Err(response) => return response,
                };
                if !tenant_owned(&json, &tenant) {
                    return error_response(403, "record belongs to another tenant");
                }
                json_response(json)
            }
            Ok(None) => not_found(&format!("{entity}/{id}")),
            Err(e) => internal_error(&format!("storage: {e}")),
        }
    })
}

pub fn create_route(
    state: Arc<Mutex<CrmState>>,
    entity: &'static str,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
    Box::new(move |req: &Request| {
        let body = match parse_body(req) {
            Ok(v) => v,
            Err(e) => return bad_request(&e),
        };
        let state = match state.lock() {
            Ok(s) => s,
            Err(_) => return internal_error("lock poisoned"),
        };
        let coll_name = collection_name(entity);
        let mut db = match state.db.lock() {
            Ok(db) => db,
            Err(_) => return internal_error("db lock poisoned"),
        };
        let tenant = match authorize(req, &mut db) {
            Ok(tenant) => tenant,
            Err(response) => return response,
        };
        let mut obj = match body {
            Value::Object(fields) => fields,
            _ => return bad_request("expected JSON object"),
        };
        if let Err(response) = validate_references(&mut db, &obj, &tenant) {
            return response;
        }
        let now = CrmState::now();
        let counter_key = format!("{coll_name}:_counter").into_bytes();
        let next_id = match db.get(&counter_key) {
            Ok(Some(bytes)) => String::from_utf8_lossy(&bytes)
                .parse::<u64>()
                .ok()
                .and_then(|id| id.checked_add(1))
                .ok_or_else(|| internal_error("ID space exhausted")),
            Ok(None) => Ok(1),
            Err(e) => Err(internal_error(&format!("counter: {e}"))),
        };
        let next_id = match next_id {
            Ok(id) => id,
            Err(response) => return response,
        };
        obj.push(("id".into(), Value::Int(next_id as i64)));
        obj.push(("tenantId".into(), Value::Str(tenant)));
        obj.push(("createdAt".into(), Value::Int(now)));
        obj.push(("updatedAt".into(), Value::Int(now)));
        let record = Value::Object(obj);
        if let Err(e) = db.insert(&counter_key, next_id.to_string().as_bytes()) {
            return internal_error(&format!("counter: {e}"));
        }
        if let Err(e) = db.insert(&build_key(coll_name, next_id), record.to_json().as_bytes()) {
            return internal_error(&format!("insert: {e}"));
        }
        if let Err(e) = db.commit() {
            return internal_error(&format!("commit: {e}"));
        }
        let _ = record_timeline(&mut db, coll_name, next_id, "created", None);
        Response::new(201)
            .with_header("Content-Type", "application/json")
            .with_body("application/json", record.to_json().into_bytes())
    })
}

pub fn update_route(
    state: Arc<Mutex<CrmState>>,
    entity: &'static str,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
    Box::new(move |req: &Request| {
        let id = match extract_id(&req.path) {
            Some(id) => id,
            None => return bad_request("missing id"),
        };
        let body = match parse_body(req) {
            Ok(v) => v,
            Err(e) => return bad_request(&e),
        };
        let state = match state.lock() {
            Ok(s) => s,
            Err(_) => return internal_error("lock poisoned"),
        };
        let coll_name = collection_name(entity);
        let mut db = match state.db.lock() {
            Ok(db) => db,
            Err(_) => return internal_error("db lock poisoned"),
        };
        let tenant = match authorize(req, &mut db) {
            Ok(tenant) => tenant,
            Err(response) => return response,
        };
        let key = build_key(coll_name, id);
        let existing = match db.get(&key) {
            Ok(Some(value)) => match parse_record(&value) {
                Ok(value) => value,
                Err(response) => return response,
            },
            Ok(None) => return not_found(&format!("{entity}/{id}")),
            Err(e) => return internal_error(&format!("storage: {e}")),
        };
        if !tenant_owned(&existing, &tenant) {
            return error_response(403, "record belongs to another tenant");
        }
        let update_fields = match body {
            Value::Object(fields) => fields,
            _ => return bad_request("expected JSON object"),
        };
        if update_fields
            .iter()
            .any(|(key, _)| key == "tenantId" || key == "id" || key == "createdAt")
        {
            return bad_request("id, tenantId, and createdAt are immutable");
        }
        if let Err(response) = validate_references(&mut db, &update_fields, &tenant) {
            return response;
        }
        let mut existing_fields = match existing {
            Value::Object(fields) => fields,
            _ => return internal_error("corrupted record"),
        };
        for (key, value) in update_fields {
            if let Some((_, existing)) = existing_fields.iter_mut().find(|(name, _)| name == &key) {
                *existing = value;
            } else {
                existing_fields.push((key, value));
            }
        }
        if let Some((_, value)) = existing_fields
            .iter_mut()
            .find(|(key, _)| key == "updatedAt")
        {
            *value = Value::Int(CrmState::now());
        } else {
            existing_fields.push(("updatedAt".into(), Value::Int(CrmState::now())));
        }
        let record = Value::Object(existing_fields);
        if let Err(e) = db.insert(&key, record.to_json().as_bytes()) {
            return internal_error(&format!("storage: {e}"));
        }
        if let Err(e) = db.commit() {
            return internal_error(&format!("commit: {e}"));
        }
        let _ = record_timeline(&mut db, coll_name, id, "updated", None);
        json_response(record)
    })
}

pub fn delete_route(
    state: Arc<Mutex<CrmState>>,
    entity: &'static str,
) -> Box<dyn Fn(&Request) -> Response + Send + Sync> {
    Box::new(move |req: &Request| {
        let id = match extract_id(&req.path) {
            Some(id) => id,
            None => return bad_request("missing id"),
        };
        let state = match state.lock() {
            Ok(s) => s,
            Err(_) => return internal_error("lock poisoned"),
        };
        let coll_name = collection_name(entity);
        let mut db = match state.db.lock() {
            Ok(db) => db,
            Err(_) => return internal_error("db lock poisoned"),
        };
        let tenant = match authorize(req, &mut db) {
            Ok(tenant) => tenant,
            Err(response) => return response,
        };
        let key = build_key(coll_name, id);
        let record = match db.get(&key) {
            Ok(Some(value)) => match parse_record(&value) {
                Ok(value) => value,
                Err(response) => return response,
            },
            Ok(None) => return not_found(&format!("{entity}/{id}")),
            Err(e) => return internal_error(&format!("storage: {e}")),
        };
        if !tenant_owned(&record, &tenant) {
            return error_response(403, "record belongs to another tenant");
        }
        if coll_name == "companies" {
            if let Err(e) = graph::mark_company_links_orphaned(&mut db, &tenant, id) {
                return internal_error(&e);
            }
        }
        match db.delete(&key) {
            Ok(true) => {
                if let Err(e) = db.commit() {
                    return internal_error(&format!("commit: {e}"));
                }
                let _ = record_timeline(&mut db, coll_name, id, "deleted", None);
                json_response(Value::Object(vec![
                    ("deleted".into(), Value::Bool(true)),
                    ("id".into(), Value::Int(id as i64)),
                ]))
            }
            Ok(false) => not_found(&format!("{entity}/{id}")),
            Err(e) => internal_error(&format!("storage: {e}")),
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::auth::session;
    use akurai_json::Value;
    use std::path::PathBuf;

    #[test]
    fn crud_routes_enforce_session_tenant_before_record_access() {
        let path = format!("/tmp/akurai-crm-crud-auth-{}.db", std::process::id());
        let _ = std::fs::remove_file(&path);
        let state = CrmState::new(&path, PathBuf::from("site/frontend")).unwrap();
        let session_id = {
            let mut db = state.db.lock().unwrap();
            for (id, tenant) in [(1_u64, "tenant-a"), (2_u64, "tenant-b")] {
                let record = Value::Object(vec![
                    ("id".into(), Value::Int(id as i64)),
                    ("tenantId".into(), Value::Str(tenant.into())),
                ]);
                db.insert(&build_key("companies", id), record.to_json().as_bytes())
                    .unwrap();
            }
            session::create(&mut db, "user-a", "user@example.test", "tenant-a").unwrap()
        };
        let request = format!(
            "GET /api/companies/2 HTTP/1.1\r\nX-Tenant-Id: tenant-b\r\nCookie: akurai_session={session_id}\r\n\r\n"
        );
        let response = get_route(Arc::new(Mutex::new(state)), "companies")(
            &Request::parse_head(request.as_bytes()).unwrap(),
        );
        assert_eq!(response.status, 403);
    }
}