mod app;
mod catalog;
mod jsonrpc;
mod session;

use std::{env, net::SocketAddr, time::Duration};

use crank_registry::PostgresRegistry;
use tokio::net::TcpListener;
use tracing::info;

use crate::app::build_app;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt()
        .with_env_filter(
            env::var("CRANK_LOG_LEVEL")
                .unwrap_or_else(|_| "mcp_server=info,tower_http=info".into()),
        )
        .init();

    let database_url = env::var("CRANK_DATABASE_URL")?;
    let bind_addr = env::var("CRANK_MCP_BIND").unwrap_or_else(|_| "0.0.0.0:3002".into());
    let public_base_url = env::var("CRANK_PUBLIC_BASE_URL").ok();
    let refresh_interval = env::var("CRANK_MCP_REFRESH_MS")
        .ok()
        .and_then(|value| value.parse::<u64>().ok())
        .map(Duration::from_millis)
        .unwrap_or_else(|| Duration::from_secs(5));
    let socket_addr: SocketAddr = bind_addr.parse()?;
    let registry = PostgresRegistry::connect(&database_url).await?;
    let app = build_app(registry, refresh_interval, public_base_url);
    let listener = TcpListener::bind(socket_addr).await?;

    info!("mcp-server listening on {}", socket_addr);

    axum::serve(listener, app).await?;

    Ok(())
}

#[cfg(test)]
mod tests {
    use std::{
        collections::BTreeMap,
        env,
        time::{Duration, SystemTime, UNIX_EPOCH},
    };

    use axum::{Json, Router, http::header, routing::post};
    use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
    use crank_adapter_grpc::test_support as grpc_test_support;
    use crank_core::{
        Agent, AgentId, AgentOperationBinding, AgentStatus, AgentVersion, DescriptorId,
        ExecutionConfig, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod, Operation,
        OperationId, OperationStatus, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope,
        PlatformApiKeyStatus, Protocol, RestTarget, Target, ToolDescription, WorkspaceId,
    };
    use crank_mapping::{MappingRule, MappingSet};
    use crank_registry::{
        CreateAgentRequest, CreatePlatformApiKeyRequest, ListInvocationLogsQuery, PostgresRegistry,
        PublishAgentRequest, PublishRequest,
    };
    use crank_schema::{Schema, SchemaKind};
    use serde_json::{Value, json};
    use sha2::{Digest, Sha256};
    use sqlx::{Executor, postgres::PgPoolOptions};
    use tokio::net::TcpListener;

    use crate::app::build_app;

    fn test_workspace_id() -> WorkspaceId {
        WorkspaceId::new("ws_default")
    }

    fn test_workspace_slug() -> &'static str {
        "default"
    }

    #[tokio::test]
    async fn initializes_lists_and_calls_published_tool_via_mcp() {
        let registry = test_registry().await;
        let upstream_base_url = spawn_upstream_server().await;
        let operation = test_operation(&upstream_base_url, "crm_create_lead");

        registry
            .create_operation(&test_workspace_id(), &operation, Some("alice"))
            .await
            .unwrap();
        registry
            .publish_operation(PublishRequest {
                workspace_id: &test_workspace_id(),
                operation_id: &operation.id,
                version: 1,
                published_at: "2026-03-26T10:00:00Z",
                published_by: Some("alice"),
            })
            .await
            .unwrap();
        publish_agent_for_operation(&registry, &operation, "sales-rest").await;
        let api_key = create_platform_api_key(
            &registry,
            "mcp-rest",
            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
        )
        .await;

        let base_url = spawn_mcp_server(build_app(
            registry.clone(),
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let mcp_url = agent_mcp_url(&base_url, "sales-rest");
        let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;

        let tools = post_jsonrpc(
            &client,
            &mcp_url,
            &api_key,
            Some(&initialized_session),
            json!({
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/list",
                "params": {}
            }),
        )
        .await;
        let call_result = post_jsonrpc(
            &client,
            &mcp_url,
            &api_key,
            Some(&initialized_session),
            json!({
                "jsonrpc": "2.0",
                "id": 3,
                "method": "tools/call",
                "params": {
                    "name": "crm_create_lead",
                    "arguments": {
                        "email": "user@example.com"
                    }
                }
            }),
        )
        .await;

        assert_eq!(tools["result"]["tools"][0]["name"], "crm_create_lead");
        assert_eq!(
            call_result["result"]["structuredContent"],
            json!({ "id": "lead_123" })
        );
        assert_eq!(call_result["result"]["isError"], false);

        let logs = registry
            .list_invocation_logs(ListInvocationLogsQuery {
                workspace_id: &test_workspace_id(),
                level: None,
                search_text: None,
                source: Some(crank_core::InvocationSource::AgentToolCall),
                operation_id: Some(&operation.id),
                agent_id: None,
                created_after: None,
                limit: 10,
            })
            .await
            .unwrap();

        assert_eq!(logs.len(), 1);
        assert_eq!(
            logs[0].log.source,
            crank_core::InvocationSource::AgentToolCall
        );
        assert_eq!(logs[0].log.status, crank_core::InvocationStatus::Ok);
        assert_eq!(logs[0].log.tool_name, "crm_create_lead");

        let keys = registry
            .list_platform_api_keys(&test_workspace_id())
            .await
            .unwrap();
        assert!(keys[0].api_key.last_used_at.is_some());
    }

    #[tokio::test]
    async fn initializes_and_calls_published_graphql_tool_via_mcp() {
        let registry = test_registry().await;
        let endpoint = spawn_graphql_server().await;
        let operation = test_graphql_operation(&endpoint, "crm_create_lead_graphql");

        registry
            .create_operation(&test_workspace_id(), &operation, Some("alice"))
            .await
            .unwrap();
        registry
            .publish_operation(PublishRequest {
                workspace_id: &test_workspace_id(),
                operation_id: &operation.id,
                version: 1,
                published_at: "2026-03-26T10:00:00Z",
                published_by: Some("alice"),
            })
            .await
            .unwrap();
        publish_agent_for_operation(&registry, &operation, "sales-graphql").await;
        let api_key = create_platform_api_key(
            &registry,
            "mcp-graphql",
            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
        )
        .await;

        let base_url = spawn_mcp_server(build_app(
            registry,
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let mcp_url = agent_mcp_url(&base_url, "sales-graphql");
        let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;

        let call_result = post_jsonrpc(
            &client,
            &mcp_url,
            &api_key,
            Some(&initialized_session),
            json!({
                "jsonrpc": "2.0",
                "id": 3,
                "method": "tools/call",
                "params": {
                    "name": "crm_create_lead_graphql",
                    "arguments": {
                        "email": "user@example.com"
                    }
                }
            }),
        )
        .await;

        assert_eq!(
            call_result["result"]["structuredContent"],
            json!({ "id": "lead_123" })
        );
        assert_eq!(call_result["result"]["isError"], false);
    }

    #[tokio::test]
    async fn initializes_and_calls_published_grpc_tool_via_mcp() {
        let registry = test_registry().await;
        let server_addr = grpc_test_support::spawn_unary_echo_server().await;
        let operation = test_grpc_operation(&server_addr, "echo_unary_grpc");

        registry
            .create_operation(&test_workspace_id(), &operation, Some("alice"))
            .await
            .unwrap();
        registry
            .publish_operation(PublishRequest {
                workspace_id: &test_workspace_id(),
                operation_id: &operation.id,
                version: 1,
                published_at: "2026-03-26T10:00:00Z",
                published_by: Some("alice"),
            })
            .await
            .unwrap();
        publish_agent_for_operation(&registry, &operation, "sales-grpc").await;
        let api_key = create_platform_api_key(
            &registry,
            "mcp-grpc",
            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
        )
        .await;

        let base_url = spawn_mcp_server(build_app(
            registry,
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let mcp_url = agent_mcp_url(&base_url, "sales-grpc");
        let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;

        let call_result = post_jsonrpc(
            &client,
            &mcp_url,
            &api_key,
            Some(&initialized_session),
            json!({
                "jsonrpc": "2.0",
                "id": 3,
                "method": "tools/call",
                "params": {
                    "name": "echo_unary_grpc",
                    "arguments": {
                        "message": "hello"
                    }
                }
            }),
        )
        .await;

        assert_eq!(
            call_result["result"]["structuredContent"],
            json!({ "message": "hello" })
        );
        assert_eq!(call_result["result"]["isError"], false);
    }

    #[tokio::test]
    async fn requires_initialized_notification_before_tool_methods() {
        let registry = test_registry().await;
        publish_agent_with_bindings(&registry, "sales-init", vec![]).await;
        let api_key =
            create_platform_api_key(&registry, "mcp-init", &[PlatformApiKeyScope::Read]).await;
        let base_url = spawn_mcp_server(build_app(
            registry,
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let mcp_url = agent_mcp_url(&base_url, "sales-init");
        let initialize_response = client
            .post(&mcp_url)
            .header(header::ACCEPT, "application/json, text/event-stream")
            .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
            .json(&json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-11-25"
                }
            }))
            .send()
            .await
            .unwrap();
        let session_id = initialize_response
            .headers()
            .get("MCP-Session-Id")
            .unwrap()
            .to_str()
            .unwrap()
            .to_owned();
        let tools_list = client
            .post(&mcp_url)
            .header(header::ACCEPT, "application/json, text/event-stream")
            .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
            .header("MCP-Session-Id", session_id)
            .header("MCP-Protocol-Version", "2025-11-25")
            .json(&json!({
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/list",
                "params": {}
            }))
            .send()
            .await
            .unwrap()
            .json::<Value>()
            .await
            .unwrap();

        assert_eq!(tools_list["error"]["code"], -32002);
    }

    #[tokio::test]
    async fn refreshes_published_tools_without_restart() {
        let registry = test_registry().await;
        let upstream_base_url = spawn_upstream_server().await;
        let base_url = spawn_mcp_server(build_app(
            registry.clone(),
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let mcp_url = agent_mcp_url(&base_url, "sales-refresh");
        let api_key = create_platform_api_key(
            &registry,
            "mcp-refresh",
            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
        )
        .await;
        let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;

        let before_publish = post_jsonrpc(
            &client,
            &mcp_url,
            &api_key,
            Some(&initialized_session),
            json!({
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/list",
                "params": {}
            }),
        )
        .await;

        let operation = test_operation(&upstream_base_url, "crm_publish_later");
        registry
            .create_operation(&test_workspace_id(), &operation, Some("alice"))
            .await
            .unwrap();
        registry
            .publish_operation(PublishRequest {
                workspace_id: &test_workspace_id(),
                operation_id: &operation.id,
                version: 1,
                published_at: "2026-03-26T10:00:00Z",
                published_by: Some("alice"),
            })
            .await
            .unwrap();
        publish_agent_for_operation(&registry, &operation, "sales-refresh").await;

        let after_publish = post_jsonrpc(
            &client,
            &mcp_url,
            &api_key,
            Some(&initialized_session),
            json!({
                "jsonrpc": "2.0",
                "id": 3,
                "method": "tools/list",
                "params": {}
            }),
        )
        .await;

        assert_eq!(before_publish["result"]["tools"], json!([]));
        assert_eq!(
            after_publish["result"]["tools"][0]["name"],
            "crm_publish_later"
        );
    }

    #[tokio::test]
    async fn lists_only_bound_tools_for_agent_context() {
        let registry = test_registry().await;
        let upstream_base_url = spawn_upstream_server().await;
        let operation_a = test_operation(&upstream_base_url, "crm_create_lead");
        let operation_b = test_operation(&upstream_base_url, "crm_update_lead");

        for operation in [&operation_a, &operation_b] {
            registry
                .create_operation(&test_workspace_id(), operation, Some("alice"))
                .await
                .unwrap();
            registry
                .publish_operation(PublishRequest {
                    workspace_id: &test_workspace_id(),
                    operation_id: &operation.id,
                    version: 1,
                    published_at: "2026-03-26T10:00:00Z",
                    published_by: Some("alice"),
                })
                .await
                .unwrap();
        }

        publish_agent_with_bindings(
            &registry,
            "sales-a",
            vec![binding_for_operation(&operation_a)],
        )
        .await;
        publish_agent_with_bindings(
            &registry,
            "sales-b",
            vec![binding_for_operation(&operation_b)],
        )
        .await;
        let api_key = create_platform_api_key(
            &registry,
            "mcp-agents",
            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
        )
        .await;

        let base_url = spawn_mcp_server(build_app(
            registry,
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let agent_a_url = agent_mcp_url(&base_url, "sales-a");
        let agent_b_url = agent_mcp_url(&base_url, "sales-b");
        let session_a = initialize_session(&client, &agent_a_url, &api_key).await;
        let session_b = initialize_session(&client, &agent_b_url, &api_key).await;

        let tools_a = post_jsonrpc(
            &client,
            &agent_a_url,
            &api_key,
            Some(&session_a),
            json!({
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/list",
                "params": {}
            }),
        )
        .await;
        let tools_b = post_jsonrpc(
            &client,
            &agent_b_url,
            &api_key,
            Some(&session_b),
            json!({
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/list",
                "params": {}
            }),
        )
        .await;

        assert_eq!(tools_a["result"]["tools"][0]["name"], "crm_create_lead");
        assert_eq!(tools_b["result"]["tools"][0]["name"], "crm_update_lead");
    }

    #[tokio::test]
    async fn rejects_initialize_without_platform_api_key() {
        let registry = test_registry().await;
        publish_agent_with_bindings(&registry, "sales-auth", vec![]).await;
        let base_url = spawn_mcp_server(build_app(
            registry,
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let response = client
            .post(agent_mcp_url(&base_url, "sales-auth"))
            .header(header::ACCEPT, "application/json, text/event-stream")
            .json(&json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-11-25"
                }
            }))
            .send()
            .await
            .unwrap();

        assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn rejects_tool_call_with_read_only_platform_api_key() {
        let registry = test_registry().await;
        let upstream_base_url = spawn_upstream_server().await;
        let operation = test_operation(&upstream_base_url, "crm_read_only");

        registry
            .create_operation(&test_workspace_id(), &operation, Some("alice"))
            .await
            .unwrap();
        registry
            .publish_operation(PublishRequest {
                workspace_id: &test_workspace_id(),
                operation_id: &operation.id,
                version: 1,
                published_at: "2026-03-26T10:00:00Z",
                published_by: Some("alice"),
            })
            .await
            .unwrap();
        publish_agent_for_operation(&registry, &operation, "sales-read-only").await;
        let api_key =
            create_platform_api_key(&registry, "mcp-read", &[PlatformApiKeyScope::Read]).await;

        let base_url = spawn_mcp_server(build_app(
            registry,
            Duration::from_millis(0),
            Some("https://crank.example.com".to_owned()),
        ))
        .await;
        let client = reqwest::Client::new();
        let mcp_url = agent_mcp_url(&base_url, "sales-read-only");
        let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
        let response = client
            .post(&mcp_url)
            .header(header::ACCEPT, "application/json, text/event-stream")
            .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
            .header("MCP-Session-Id", initialized_session)
            .header("MCP-Protocol-Version", "2025-11-25")
            .json(&json!({
                "jsonrpc": "2.0",
                "id": 3,
                "method": "tools/call",
                "params": {
                    "name": "crm_read_only",
                    "arguments": {
                        "email": "user@example.com"
                    }
                }
            }))
            .send()
            .await
            .unwrap();

        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
    }

    async fn initialize_session(client: &reqwest::Client, mcp_url: &str, api_key: &str) -> String {
        let initialize_response = client
            .post(mcp_url)
            .header(header::ACCEPT, "application/json, text/event-stream")
            .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
            .json(&json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-11-25"
                }
            }))
            .send()
            .await
            .unwrap();
        let session_id = initialize_response
            .headers()
            .get("MCP-Session-Id")
            .unwrap()
            .to_str()
            .unwrap()
            .to_owned();

        let initialized_response = client
            .post(mcp_url)
            .header(header::ACCEPT, "application/json, text/event-stream")
            .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
            .header("MCP-Session-Id", &session_id)
            .header("MCP-Protocol-Version", "2025-11-25")
            .json(&json!({
                "jsonrpc": "2.0",
                "method": "notifications/initialized",
                "params": {}
            }))
            .send()
            .await
            .unwrap();

        assert_eq!(initialized_response.status(), reqwest::StatusCode::ACCEPTED);

        session_id
    }

    async fn post_jsonrpc(
        client: &reqwest::Client,
        mcp_url: &str,
        api_key: &str,
        session_id: Option<&str>,
        payload: Value,
    ) -> Value {
        let mut request = client
            .post(mcp_url)
            .header(header::ACCEPT, "application/json, text/event-stream")
            .header(header::AUTHORIZATION, format!("Bearer {api_key}"))
            .header("MCP-Protocol-Version", "2025-11-25");

        if let Some(session_id) = session_id {
            request = request.header("MCP-Session-Id", session_id);
        }

        request
            .json(&payload)
            .send()
            .await
            .unwrap()
            .json::<Value>()
            .await
            .unwrap()
    }

    async fn create_platform_api_key(
        registry: &PostgresRegistry,
        name: &str,
        scopes: &[PlatformApiKeyScope],
    ) -> String {
        let secret = format!("crk_{}_{}", name, uuid::Uuid::now_v7().simple());
        let api_key = PlatformApiKey {
            id: PlatformApiKeyId::new(format!("pk_{name}")),
            workspace_id: test_workspace_id(),
            name: name.to_owned(),
            prefix: secret.chars().take(16).collect(),
            scopes: scopes.to_vec(),
            status: PlatformApiKeyStatus::Active,
            created_at: "2026-03-26T10:00:00Z".to_owned(),
            last_used_at: None,
        };

        registry
            .create_platform_api_key(CreatePlatformApiKeyRequest {
                api_key: &api_key,
                secret_hash: &hash_access_secret(&secret),
            })
            .await
            .unwrap();

        secret
    }

    fn hash_access_secret(secret: &str) -> String {
        let digest = Sha256::digest(secret.as_bytes());
        URL_SAFE_NO_PAD.encode(digest)
    }

    async fn spawn_upstream_server() -> String {
        let app = Router::new().route("/crm/leads", post(create_lead));
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();

        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });

        format!("http://{}", address)
    }

    async fn spawn_graphql_server() -> String {
        let app = Router::new().route("/", post(graphql_handler));
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();

        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });

        format!("http://{}", address)
    }

    async fn spawn_mcp_server(app: Router) -> String {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();

        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });

        format!("http://{}", address)
    }

    fn agent_mcp_url(base_url: &str, agent_slug: &str) -> String {
        format!("{base_url}/v1/{}/{}", test_workspace_slug(), agent_slug)
    }

    async fn publish_agent_for_operation(
        registry: &PostgresRegistry,
        operation: &Operation<Schema, MappingSet>,
        agent_slug: &str,
    ) {
        publish_agent_with_bindings(registry, agent_slug, vec![binding_for_operation(operation)])
            .await;
    }

    async fn publish_agent_with_bindings(
        registry: &PostgresRegistry,
        agent_slug: &str,
        bindings: Vec<AgentOperationBinding>,
    ) {
        let agent_id = AgentId::new(format!("agent_{agent_slug}"));
        let agent = Agent {
            id: agent_id.clone(),
            workspace_id: test_workspace_id(),
            slug: agent_slug.to_owned(),
            display_name: format!("Agent {agent_slug}"),
            description: "Curated MCP toolset".to_owned(),
            status: AgentStatus::Draft,
            current_draft_version: 1,
            latest_published_version: None,
            created_at: "2026-03-26T10:00:00Z".to_owned(),
            updated_at: "2026-03-26T10:00:00Z".to_owned(),
            published_at: None,
        };
        let version = AgentVersion {
            agent_id: agent_id.clone(),
            version: 1,
            status: AgentStatus::Draft,
            instructions: json!({}),
            tool_selection_policy: json!({}),
            created_at: "2026-03-26T10:00:00Z".to_owned(),
        };

        registry
            .create_agent(CreateAgentRequest {
                agent: &agent,
                version: &version,
                bindings: &bindings,
            })
            .await
            .unwrap();
        registry
            .publish_agent(PublishAgentRequest {
                workspace_id: &test_workspace_id(),
                agent_id: &agent_id,
                version: 1,
                published_at: "2026-03-26T10:00:00Z",
                published_by: Some("alice"),
            })
            .await
            .unwrap();
    }

    fn binding_for_operation(operation: &Operation<Schema, MappingSet>) -> AgentOperationBinding {
        AgentOperationBinding {
            operation_id: operation.id.clone(),
            operation_version: 1,
            tool_name: operation.name.clone(),
            tool_title: operation.tool_description.title.clone(),
            tool_description_override: None,
            enabled: true,
        }
    }

    async fn create_lead(Json(payload): Json<Value>) -> Json<Value> {
        Json(json!({
            "id": "lead_123",
            "email": payload["email"]
        }))
    }

    async fn graphql_handler(Json(payload): Json<Value>) -> Json<Value> {
        let email = payload
            .get("variables")
            .and_then(|variables| variables.get("email"))
            .and_then(Value::as_str)
            .unwrap_or_default();

        Json(json!({
            "data": {
                "createLead": {
                    "id": "lead_123",
                    "email": email
                }
            }
        }))
    }

    async fn test_registry() -> PostgresRegistry {
        let database_url = env::var("TEST_DATABASE_URL")
            .unwrap_or_else(|_| "postgres://crank:crank@127.0.0.1:5432/crank".to_owned());
        let admin_pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(&database_url)
            .await
            .unwrap();
        let schema = format!(
            "test_mcp_server_{}_{}",
            std::process::id(),
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );

        admin_pool
            .execute(sqlx::query(&format!("create schema {schema}")))
            .await
            .unwrap();

        PostgresRegistry::connect(&format!("{database_url}?options=-csearch_path%3D{schema}"))
            .await
            .unwrap()
    }

    fn test_operation(base_url: &str, name: &str) -> Operation<Schema, MappingSet> {
        Operation {
            id: OperationId::new(format!("op_{name}")),
            name: name.to_owned(),
            display_name: "Create Lead".to_owned(),
            category: "sales".to_owned(),
            protocol: Protocol::Rest,
            status: OperationStatus::Published,
            version: 1,
            target: Target::Rest(RestTarget {
                base_url: base_url.to_owned(),
                method: HttpMethod::Post,
                path_template: "/crm/leads".to_owned(),
                static_headers: BTreeMap::new(),
            }),
            input_schema: object_schema("email"),
            output_schema: object_schema("id"),
            input_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.mcp.email".to_owned(),
                    target: "$.request.body.email".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            output_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.response.body.id".to_owned(),
                    target: "$.output.id".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            execution_config: ExecutionConfig {
                timeout_ms: 1_000,
                retry_policy: None,
                auth_profile_ref: None,
                headers: BTreeMap::new(),
                protocol_options: None,
            },
            tool_description: ToolDescription {
                title: "Create Lead".to_owned(),
                description: "Creates a CRM lead".to_owned(),
                tags: Vec::new(),
                examples: Vec::new(),
            },
            samples: None,
            generated_draft: None,
            config_export: None,
            created_at: "2026-03-26T10:00:00Z".to_owned(),
            updated_at: "2026-03-26T10:00:00Z".to_owned(),
            published_at: Some("2026-03-26T10:00:00Z".to_owned()),
        }
    }

    fn test_graphql_operation(endpoint: &str, name: &str) -> Operation<Schema, MappingSet> {
        Operation {
            id: OperationId::new(format!("op_{name}")),
            name: name.to_owned(),
            display_name: "Create Lead GraphQL".to_owned(),
            category: "sales".to_owned(),
            protocol: Protocol::Graphql,
            status: OperationStatus::Published,
            version: 1,
            target: Target::Graphql(GraphqlTarget {
                endpoint: endpoint.to_owned(),
                operation_type: GraphqlOperationType::Mutation,
                operation_name: "CreateLead".to_owned(),
                query_template:
                    "mutation CreateLead($email: String!) { createLead(email: $email) { id email } }"
                        .to_owned(),
                response_path: "$.response.body.data.createLead".to_owned(),
            }),
            input_schema: object_schema("email"),
            output_schema: object_schema("id"),
            input_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.mcp.email".to_owned(),
                    target: "$.request.variables.email".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            output_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.response.data.id".to_owned(),
                    target: "$.output.id".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            execution_config: ExecutionConfig {
                timeout_ms: 1_000,
                retry_policy: None,
                auth_profile_ref: None,
                headers: BTreeMap::new(),
                protocol_options: None,
            },
            tool_description: ToolDescription {
                title: "Create Lead GraphQL".to_owned(),
                description: "Creates a CRM lead through GraphQL".to_owned(),
                tags: vec!["graphql".to_owned()],
                examples: Vec::new(),
            },
            samples: None,
            generated_draft: None,
            config_export: None,
            created_at: "2026-03-26T10:00:00Z".to_owned(),
            updated_at: "2026-03-26T10:00:00Z".to_owned(),
            published_at: Some("2026-03-26T10:00:00Z".to_owned()),
        }
    }

    fn test_grpc_operation(server_addr: &str, name: &str) -> Operation<Schema, MappingSet> {
        Operation {
            id: OperationId::new(format!("op_{name}")),
            name: name.to_owned(),
            display_name: "Unary Echo gRPC".to_owned(),
            category: "support".to_owned(),
            protocol: Protocol::Grpc,
            status: OperationStatus::Published,
            version: 1,
            target: Target::Grpc(GrpcTarget {
                server_addr: server_addr.to_owned(),
                package: "echo".to_owned(),
                service: "EchoService".to_owned(),
                method: "UnaryEcho".to_owned(),
                descriptor_ref: DescriptorId::new("desc_echo"),
                descriptor_set_b64: grpc_test_support::descriptor_set_b64(),
            }),
            input_schema: object_schema("message"),
            output_schema: object_schema("message"),
            input_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.mcp.message".to_owned(),
                    target: "$.request.grpc.message".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            output_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.response.data.message".to_owned(),
                    target: "$.output.message".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            execution_config: ExecutionConfig {
                timeout_ms: 1_000,
                retry_policy: None,
                auth_profile_ref: None,
                headers: BTreeMap::new(),
                protocol_options: None,
            },
            tool_description: ToolDescription {
                title: "Unary Echo gRPC".to_owned(),
                description: "Echoes a unary gRPC payload".to_owned(),
                tags: vec!["grpc".to_owned()],
                examples: Vec::new(),
            },
            samples: None,
            generated_draft: None,
            config_export: None,
            created_at: "2026-03-26T10:00:00Z".to_owned(),
            updated_at: "2026-03-26T10:00:00Z".to_owned(),
            published_at: Some("2026-03-26T10:00:00Z".to_owned()),
        }
    }

    fn object_schema(field_name: &str) -> Schema {
        Schema {
            kind: SchemaKind::Object,
            description: None,
            required: true,
            nullable: false,
            default_value: None,
            fields: BTreeMap::from([(
                field_name.to_owned(),
                Schema {
                    kind: SchemaKind::String,
                    description: None,
                    required: true,
                    nullable: false,
                    default_value: None,
                    fields: BTreeMap::new(),
                    items: None,
                    enum_values: Vec::new(),
                    variants: Vec::new(),
                },
            )]),
            items: None,
            enum_values: Vec::new(),
            variants: Vec::new(),
        }
    }
}