crank/apps/admin-api/tests/integration/community_access_usage.rs

#![allow(dead_code, unused_imports)]

use super::common::*;

use std::{
    collections::BTreeMap,
    env, fmt,
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};

use async_trait::async_trait;
use axum::{Json, Router, routing::post};
use crank_core::{
    ExecutionConfig, HttpMethod, MembershipRole, OperationSecurityLevel, Protocol,
    ResponseCachePolicy, RestTarget, SecretKind, Target, ToolDescription, WorkspaceId,
};
use crank_core::{IdentityError, IdentityProvider, IdentityProviderKind, LoginOutcome};
use crank_mapping::{MappingRule, MappingSet};
use crank_registry::PostgresRegistry;
use crank_runtime::SecretCrypto;
use crank_schema::{Schema, SchemaKind};
use serde_json::{Value, json};
use serial_test::serial;
use tokio::net::TcpListener;

use admin_api::{
    app::build_app,
    auth::{AuthSettings, BootstrapAdminConfig, hash_password},
    service::{AdminService, AdminServiceBuilder, OperationPayload},
    state::AppState,
};

const DEFAULT_WORKSPACE_ID: &str = "ws_default";
const TEST_AUTH_EMAIL: &str = "owner@crank.local";
const TEST_AUTH_PASSWORD: &str = "test-password";
const TEST_PASSWORD_PEPPER: &str = "test-password-pepper";
const TEST_SESSION_SECRET: &str = "test-session-secret";
const TEST_MASTER_KEY: &str = "test-master-key";

struct TestServer {
    base_url: String,
    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
    handle: Option<tokio::task::JoinHandle<()>>,
}

impl Drop for TestServer {
    fn drop(&mut self) {
        let shutdown = self.shutdown.take();
        let handle = self.handle.take();

        tokio::task::block_in_place(|| {
            if let Some(shutdown) = shutdown {
                let _ = shutdown.send(());
            }
            if let Some(handle) = handle {
                let _ = tokio::runtime::Handle::current().block_on(handle);
            }
        });
    }
}

impl fmt::Display for TestServer {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(&self.base_url)
    }
}

impl AsRef<str> for TestServer {
    fn as_ref(&self) -> &str {
        &self.base_url
    }
}

struct RejectingIdentityProvider;

#[async_trait]
impl IdentityProvider for RejectingIdentityProvider {
    fn id(&self) -> &str {
        "rejecting-test-provider"
    }

    fn kind(&self) -> IdentityProviderKind {
        IdentityProviderKind::Password
    }

    async fn login_password(
        &self,
        _payload: crank_core::LoginPayload,
    ) -> Result<LoginOutcome, IdentityError> {
        Err(IdentityError::BadCredentials)
    }
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn rejects_workspace_access_management_in_community() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("platform_access");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let members_status = client
        .get(format!("{base_url}/members"))
        .send()
        .await
        .unwrap()
        .status();
    let create_invitation_status = client
        .post(format!("{base_url}/invitations"))
        .json(&json!({
            "email": "operator@example.com",
            "role": "operator"
        }))
        .send()
        .await
        .unwrap()
        .status();

    assert_eq!(members_status, reqwest::StatusCode::NOT_FOUND);
    assert_eq!(create_invitation_status, reqwest::StatusCode::NOT_FOUND);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn manages_agent_platform_api_keys() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("agent_platform_keys");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let created_agent = assert_success_json(
        client
            .post(format!("{base_url}/agents"))
            .json(&json!({
                "slug": "sales-routing",
                "display_name": "Sales Routing",
                "description": "Routing agent",
                "instructions": {},
                "tool_selection_policy": {}
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let agent_id = created_agent["agent_id"].as_str().unwrap().to_owned();

    let created_key = assert_success_json(
        client
            .post(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
            .json(&json!({
                "name": "sales-routing-primary",
                "scopes": ["read", "write"]
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    let key_id = created_key["api_key"]["api_key"]["id"]
        .as_str()
        .unwrap()
        .to_owned();

    let listed_keys = assert_success_json(
        client
            .get(format!("{base_url}/agents/{agent_id}/platform-api-keys"))
            .send()
            .await
            .unwrap(),
    )
    .await;

    let revoke_status = client
        .post(format!(
            "{base_url}/agents/{agent_id}/platform-api-keys/{key_id}/revoke"
        ))
        .send()
        .await
        .unwrap()
        .status();
    let delete_status = client
        .delete(format!(
            "{base_url}/agents/{agent_id}/platform-api-keys/{key_id}"
        ))
        .send()
        .await
        .unwrap()
        .status();

    assert_eq!(created_key["api_key"]["api_key"]["agent_id"], agent_id);
    assert_eq!(
        listed_keys["items"][0]["api_key"]["agent_id"],
        json!(agent_id)
    );
    assert_eq!(
        listed_keys["items"][0]["api_key"]["name"],
        "sales-routing-primary"
    );
    assert!(created_key["secret"].as_str().unwrap().starts_with("crk_"));
    assert_eq!(revoke_status, reqwest::StatusCode::NO_CONTENT);
    assert_eq!(delete_status, reqwest::StatusCode::NO_CONTENT);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn returns_community_capabilities() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("community_capabilities");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;
    let root_url = base_url
        .as_ref()
        .split("/api/admin/workspaces/")
        .next()
        .unwrap();

    let response = assert_success_json(
        client
            .get(format!("{root_url}/api/admin/capabilities"))
            .send()
            .await
            .unwrap(),
    )
    .await;

    assert_eq!(response["edition"], "community");
    assert_eq!(response["supported_protocols"], json!(["rest"]));
    assert_eq!(response["supported_security_levels"], json!(["standard"]));
    assert_eq!(
        response["machine_access_modes"],
        json!(["static_agent_key"])
    );
    assert_eq!(response["limits"]["max_workspaces"], 1);
    assert_eq!(response["limits"]["max_users_per_workspace"], 1);
    assert_eq!(response["limits"]["max_agents_per_workspace"], Value::Null);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn rejects_response_cache_for_non_get_rest_operation_create() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("community_response_cache_post_reject");
    let upstream_base_url = spawn_upstream_server().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let mut payload = test_operation_payload(&upstream_base_url, "crm_cache_post");
    payload.execution_config.response_cache = Some(ResponseCachePolicy { ttl_ms: 5_000 });

    let response = client
        .post(format!("{base_url}/operations"))
        .json(&payload)
        .send()
        .await
        .unwrap();
    let status = response.status();
    let body = response.json::<Value>().await.unwrap();

    assert_eq!(status, reqwest::StatusCode::BAD_REQUEST);
    assert_eq!(body["error"]["code"], "validation_error");
    assert_eq!(
        body["error"]["context"]["field"],
        "execution_config.response_cache"
    );
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn rejects_response_cache_for_operation_with_auth_profile() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("community_response_cache_auth_reject");
    let upstream_base_url = spawn_upstream_server().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let mut payload = test_operation_payload(&upstream_base_url, "crm_cache_auth");
    payload.target = Target::Rest(RestTarget {
        base_url: upstream_base_url,
        method: HttpMethod::Get,
        path_template: "/crm/leads".to_owned(),
        static_headers: BTreeMap::new(),
    });
    payload.execution_config.response_cache = Some(ResponseCachePolicy { ttl_ms: 5_000 });
    payload.execution_config.auth_profile_ref = Some("auth_profile_01".into());

    let response = client
        .post(format!("{base_url}/operations"))
        .json(&payload)
        .send()
        .await
        .unwrap();
    let status = response.status();
    let body = response.json::<Value>().await.unwrap();

    assert_eq!(status, reqwest::StatusCode::BAD_REQUEST);
    assert_eq!(body["error"]["code"], "validation_error");
    assert_eq!(
        body["error"]["context"]["field"],
        "execution_config.auth_profile_ref"
    );
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn exports_single_workspace_but_rejects_access_lifecycle() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("workspace_access");
    let base_url = spawn_admin_api(build_test_app(registry.clone(), storage_root)).await;
    let client = authorized_client(&base_url).await;

    let second_user_id = registry
        .upsert_bootstrap_user("operator-2@crank.local", "Operator Two", "external-hash")
        .await
        .unwrap();
    registry
        .ensure_membership(
            &WorkspaceId::new(DEFAULT_WORKSPACE_ID),
            &second_user_id,
            MembershipRole::Viewer,
        )
        .await
        .unwrap();

    let update_member_status = client
        .patch(format!("{base_url}/members/{}", second_user_id.as_str()))
        .json(&json!({ "role": "admin" }))
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(update_member_status, reqwest::StatusCode::NOT_FOUND);

    let exported = assert_success_json(
        client
            .get(format!("{base_url}/export"))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(
        exported["workspace"]["workspace"]["id"],
        DEFAULT_WORKSPACE_ID
    );
    assert!(exported.get("memberships").is_none());
    assert!(exported.get("invitations").is_none());

    let delete_member_status = client
        .delete(format!("{base_url}/members/{}", second_user_id.as_str()))
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(delete_member_status, reqwest::StatusCode::NOT_FOUND);

    let delete_workspace_status = client
        .delete(base_url.as_ref())
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(
        delete_workspace_status,
        reqwest::StatusCode::METHOD_NOT_ALLOWED
    );

    let root_url = base_url
        .as_ref()
        .split("/api/admin/workspaces/")
        .next()
        .unwrap();
    let still_available = assert_success_json(
        client
            .get(format!(
                "{root_url}/api/admin/workspaces/{DEFAULT_WORKSPACE_ID}"
            ))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(still_available["workspace"]["id"], DEFAULT_WORKSPACE_ID);
}

#[tokio::test]
#[serial]
async fn seeds_demo_assets_for_live_ui() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("demo_seed");
    let service = test_service(
        registry.clone(),
        storage_root,
        test_auth_settings(),
        test_secret_crypto(),
    );

    service.bootstrap_admin_user().await.unwrap();
    service.seed_demo_assets().await.unwrap();
    service.seed_demo_assets().await.unwrap();

    let owner = registry
        .get_auth_user_by_email(TEST_AUTH_EMAIL)
        .await
        .unwrap()
        .unwrap();
    let workspaces = service
        .list_workspaces_for_user(&owner.user.id)
        .await
        .unwrap();
    assert!(
        workspaces
            .iter()
            .any(|item| item.workspace.slug == "default")
    );
    assert_eq!(workspaces.len(), 1);

    let default_workspace_id = WorkspaceId::new(DEFAULT_WORKSPACE_ID);
    let operations = service
        .list_operations(&default_workspace_id)
        .await
        .unwrap();
    assert!(operations.len() >= 2);

    let agents = service.list_agents(&default_workspace_id).await.unwrap();
    assert!(!agents.is_empty());

    assert!(agents.iter().any(|agent| agent.key_count > 0));

    let logs = service
        .list_logs(
            &default_workspace_id,
            admin_api::service::LogsQuery {
                level: None,
                search: None,
                source: None,
                operation_id: None,
                agent_id: None,
                period: None,
                limit: Some(20),
            },
        )
        .await
        .unwrap();
    assert!(!logs.is_empty());
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn updates_profile_and_changes_password() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("settings_profile");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let root_url = base_url
        .as_ref()
        .split("/api/admin/workspaces/")
        .next()
        .unwrap()
        .to_owned();
    let client = authorized_client(&base_url).await;

    let profile = assert_success_json(
        client
            .get(format!("{root_url}/api/auth/profile"))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(profile["user"]["email"], TEST_AUTH_EMAIL);

    let updated_profile = assert_success_json(
        client
            .patch(format!("{root_url}/api/auth/profile"))
            .json(&json!({
                "display_name": "Updated Owner",
                "email": "updated-owner@crank.local"
            }))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(updated_profile["user"]["display_name"], "Updated Owner");
    assert_eq!(
        updated_profile["user"]["email"],
        "updated-owner@crank.local"
    );

    let password_status = client
        .post(format!("{root_url}/api/auth/password"))
        .json(&json!({
            "current_password": TEST_AUTH_PASSWORD,
            "new_password": "updated-password-123"
        }))
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(password_status, reqwest::StatusCode::NO_CONTENT);

    let relogin_client = reqwest::Client::builder()
        .cookie_store(true)
        .build()
        .unwrap();
    let relogin_status = relogin_client
        .post(format!("{root_url}/api/auth/login"))
        .json(&json!({
            "email": "updated-owner@crank.local",
            "password": "updated-password-123"
        }))
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(relogin_status, reqwest::StatusCode::OK);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn rejects_multi_workspace_session_switching_in_community() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("session_workspace");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let root_url = base_url
        .as_ref()
        .split("/api/admin/workspaces/")
        .next()
        .unwrap()
        .to_owned();
    let client = authorized_client(&base_url).await;

    let create_workspace_status = client
        .post(format!("{root_url}/api/admin/workspaces"))
        .json(&json!({
            "slug": "growth-lab",
            "display_name": "Growth Lab",
            "settings": {}
        }))
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(
        create_workspace_status,
        reqwest::StatusCode::METHOD_NOT_ALLOWED
    );

    let switch_status = client
        .post(format!("{root_url}/api/auth/current-workspace"))
        .json(&json!({ "workspace_id": DEFAULT_WORKSPACE_ID }))
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(switch_status, reqwest::StatusCode::NOT_FOUND);

    let session = assert_success_json(
        client
            .get(format!("{root_url}/api/auth/session"))
            .send()
            .await
            .unwrap(),
    )
    .await;
    assert_eq!(session["current_workspace_id"], DEFAULT_WORKSPACE_ID);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn exposes_logs_and_usage_from_real_test_runs() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("observability");
    let upstream_base_url = spawn_upstream_server().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let created = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload(
            &upstream_base_url,
            "crm_observability",
        ))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let operation_id = created["operation_id"].as_str().unwrap().to_owned();

    client
        .post(format!("{base_url}/operations/{operation_id}/test-runs"))
        .json(&json!({
            "version": 1,
            "input": { "email": "user@example.com" }
        }))
        .send()
        .await
        .unwrap();

    let logs = client
        .get(format!("{base_url}/logs?period=7d"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let log_id = logs["items"][0]["log"]["id"].as_str().unwrap().to_owned();
    let log_detail = client
        .get(format!("{base_url}/logs/{log_id}"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let usage = client
        .get(format!("{base_url}/usage?period=7d"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let operation_usage = client
        .get(format!(
            "{base_url}/usage/operations/{operation_id}?period=7d"
        ))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();

    assert_eq!(logs["items"][0]["log"]["source"], "admin_test_run");
    assert_eq!(logs["items"][0]["operation_name"], "crm_observability");
    assert_eq!(log_detail["log"]["status"], "ok");
    assert_eq!(usage["summary"]["rollup"]["calls_total"], 1);
    assert_eq!(usage["summary"]["rollup"]["calls_ok"], 1);
    assert_eq!(
        usage["operations"][0]["operation_name"],
        "crm_observability"
    );
    assert_eq!(operation_usage["rollup"]["calls_total"], 1);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn preserves_request_id_for_test_run_invocations() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("observability_request_id");
    let upstream_base_url = spawn_upstream_server().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let created = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload(
            &upstream_base_url,
            "crm_observability_request_id",
        ))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let operation_id = created["operation_id"].as_str().unwrap().to_owned();

    let response = client
        .post(format!("{base_url}/operations/{operation_id}/test-runs"))
        .header("x-request-id", "req_test_123")
        .json(&json!({
            "version": 1,
            "input": { "email": "user@example.com" }
        }))
        .send()
        .await
        .unwrap();

    assert_eq!(
        response.headers()["x-request-id"].to_str().unwrap(),
        "req_test_123"
    );
    response.error_for_status().unwrap();

    let logs = client
        .get(format!("{base_url}/logs?period=7d"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();

    assert_eq!(logs["items"][0]["log"]["request_id"], "req_test_123");
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn generates_request_id_for_test_run_invocations() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("observability_generated_request_id");
    let upstream_base_url = spawn_upstream_server().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let created = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload(
            &upstream_base_url,
            "crm_observability_generated_request_id",
        ))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let operation_id = created["operation_id"].as_str().unwrap().to_owned();

    let response = client
        .post(format!("{base_url}/operations/{operation_id}/test-runs"))
        .json(&json!({
            "version": 1,
            "input": { "email": "user@example.com" }
        }))
        .send()
        .await
        .unwrap();
    let request_id = response
        .headers()
        .get("x-request-id")
        .unwrap()
        .to_str()
        .unwrap()
        .to_owned();

    assert!(!request_id.is_empty());
    response.error_for_status().unwrap();

    let logs = client
        .get(format!("{base_url}/logs?period=7d"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();

    assert_eq!(logs["items"][0]["log"]["request_id"], request_id);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn returns_structured_context_for_missing_operation_usage() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("missing_operation_usage");
    let upstream_base_url = spawn_upstream_server().await;
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let created = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload(
            &upstream_base_url,
            "crm_missing_usage",
        ))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let operation_id = created["operation_id"].as_str().unwrap().to_owned();

    let response = client
        .get(format!(
            "{base_url}/usage/operations/{operation_id}?period=7d"
        ))
        .send()
        .await
        .unwrap();
    let status = response.status();
    let body = response.json::<Value>().await.unwrap();

    assert_eq!(status, reqwest::StatusCode::NOT_FOUND);
    assert_eq!(body["error"]["code"], "not_found");
    assert_eq!(
        body["error"]["message"],
        format!("usage for operation {operation_id} was not found")
    );
    assert_eq!(
        body["error"]["context"],
        json!({
            "operation_id": operation_id
        })
    );
}