mnote/rust/crates/mnote-web/src/transport/convex.rs

use crate::app::AppConfig;
use crate::context::RequestContext;
use crate::error::WebError;
use base64::engine::general_purpose::STANDARD;
use base64::Engine;
use bridge_runtime::RuntimeQueryExecutionPlan;
use serde_json::{json, Value};
use std::fs;
use std::time::Duration;

const HEADER_REQUEST_ID: &str = "x-request-id";
const HEADER_TRACE_ID: &str = "x-trace-id";
const HEADER_WORKSPACE_ID: &str = "x-mnote-workspace-id";
const HEADER_SOURCE_CHANNEL: &str = "x-mnote-source-channel";
const HEADER_SOURCE_CLIENT: &str = "x-mnote-source-client";

fn read_env_or_dotenv(key: &str) -> Option<String> {
    if let Ok(value) = std::env::var(key) {
        let trimmed = value.trim().trim_matches('"').to_string();
        if !trimmed.is_empty() {
            return Some(trimmed);
        }
    }

    let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("../../..")
        .join(".env.all");
    let content = fs::read_to_string(root).ok()?;
    for line in content.lines() {
        let line = line.trim_end_matches('\r');
        if line.starts_with('#') || line.trim().is_empty() {
            continue;
        }
        let Some((k, v)) = line.split_once('=') else {
            continue;
        };
        if k.trim() != key {
            continue;
        }
        let trimmed = v.trim().trim_matches('"').to_string();
        if !trimmed.is_empty() {
            return Some(trimmed);
        }
    }
    None
}

fn build_authorization(config: &AppConfig, context: &RequestContext) -> Result<String, WebError> {
    if let Some(authorization) = context
        .auth
        .authorization
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        return Ok(authorization.to_string());
    }

    let admin_key = config
        .convex_admin_key
        .clone()
        .or_else(|| read_env_or_dotenv("CONVEX_SELF_HOSTED_ADMIN_KEY"))
        .ok_or_else(|| {
            WebError::service_unavailable_code(
                "convex_config_missing",
                "缺少 CONVEX_SELF_HOSTED_ADMIN_KEY，且请求未携带 Authorization",
            )
            .with_context(context)
            .with_header("x-error-phase", "convex_auth")
            .with_header("x-upstream-service", "convex")
        })?;

    let identity = json!({
        "subject": config.dev_user_id,
        "issuer": "https://mnote.local/dev-auth",
        "tokenIdentifier": format!("dev-user|{}", config.dev_user_id),
        "name": config.dev_user_name,
        "email": config.dev_user_email,
    });
    let encoded = STANDARD.encode(
        serde_json::to_string(&identity)
            .map_err(|error| WebError::internal(format!("开发用户身份序列化失败: {error}")))?,
    );

    Ok(format!("Convex {admin_key}:{encoded}"))
}

fn convex_url(config: &AppConfig, context: &RequestContext) -> Result<String, WebError> {
    config
        .convex_url
        .clone()
        .or_else(|| read_env_or_dotenv("CONVEX_SELF_HOSTED_URL"))
        .or_else(|| read_env_or_dotenv("NEXT_PUBLIC_CONVEX_URL"))
        .map(|value| value.trim().trim_end_matches('/').to_string())
        .filter(|value| !value.is_empty())
        .ok_or_else(|| {
            WebError::service_unavailable_code(
                "convex_config_missing",
                "缺少 CONVEX_SELF_HOSTED_URL / NEXT_PUBLIC_CONVEX_URL",
            )
            .with_context(context)
            .with_header("x-error-phase", "convex_url")
            .with_header("x-upstream-service", "convex")
        })
}

fn load_query_fixture(
    config: &AppConfig,
    context: &RequestContext,
    plan: &RuntimeQueryExecutionPlan,
) -> Result<Option<Value>, WebError> {
    if !config.allow_dev_fixtures {
        return Ok(None);
    }

    let Ok(raw) = std::env::var("MNOTE_WEB_QUERY_FIXTURES_JSON") else {
        return Ok(None);
    };
    let trimmed = raw.trim();
    if trimmed.is_empty() {
        return Ok(None);
    }

    let fixtures: Value = serde_json::from_str(trimmed).map_err(|error| {
        WebError::internal(format!("MNOTE_WEB_QUERY_FIXTURES_JSON 非法: {error}"))
            .with_context(context)
            .with_header("x-error-phase", "fixture_parse")
    })?;

    Ok(fixtures
        .as_object()
        .and_then(|map| map.get(plan.function_name.as_str()))
        .cloned())
}

pub fn execute_convex_query_plan(
    config: &AppConfig,
    context: &RequestContext,
    plan: &RuntimeQueryExecutionPlan,
) -> Result<Value, WebError> {
    if plan.function_name.trim().is_empty() || plan.function_name.ends_with(":unknown") {
        return Err(WebError::bad_request_code(
            "transport_query_unsupported",
            format!("mnote-web transport 暂不支持 query: {}", plan.function_name),
        )
        .with_context(context)
        .with_header("x-error-phase", "plan_validation"));
    }

    if let Some(fixture) = load_query_fixture(config, context, plan)? {
        return Ok(fixture);
    }

    let payload = json!({
        "path": plan.function_name,
        "format": "convex_encoded_json",
        "args": plan.args_json,
    });

    let client = reqwest::blocking::Client::builder()
        .timeout(Duration::from_secs(20))
        .build()
        .map_err(|error| {
            WebError::internal(format!("Convex HTTP 客户端创建失败: {error}"))
                .with_context(context)
                .with_header("x-error-phase", "client_build")
                .with_header("x-upstream-service", "convex")
        })?;

    let mut request = client
        .post(format!("{}/api/query", convex_url(config, context)?))
        .header("Authorization", build_authorization(config, context)?)
        .header("Content-Type", "application/json")
        .header("Convex-Client", "mnote-web")
        .header(HEADER_REQUEST_ID, &context.trace.request_id)
        .header(HEADER_TRACE_ID, &context.trace.trace_id)
        .header(HEADER_SOURCE_CHANNEL, context.source.channel.as_str())
        .header(HEADER_SOURCE_CLIENT, context.source.client.as_str())
        .json(&payload);

    if let Some(workspace_id) = plan
        .workspace_id
        .as_deref()
        .or(context.workspace.workspace_id.as_deref())
    {
        request = request.header(HEADER_WORKSPACE_ID, workspace_id);
    }

    let response = request.send().map_err(|error| {
        let base = if error.is_timeout() {
            WebError::gateway_timeout_code("convex_timeout", format!("Convex query 超时: {error}"))
        } else {
            WebError::service_unavailable_code(
                "convex_unavailable",
                format!("Convex query 请求失败: {error}"),
            )
        };
        base.with_context(context)
            .with_header("x-error-phase", "query_send")
            .with_header("x-upstream-service", "convex")
    })?;

    let status = response.status();
    let body: Value = response.json().map_err(|error| {
        WebError::bad_gateway_code(
            "convex_bad_response",
            format!("Convex 响应解析失败: {error}"),
        )
        .with_context(context)
        .with_header("x-error-phase", "query_decode")
        .with_header("x-upstream-service", "convex")
        .with_header("x-upstream-status", status.as_u16().to_string())
    })?;
    if !status.is_success() {
        let message = body
            .get("errorMessage")
            .and_then(Value::as_str)
            .unwrap_or("Convex query 失败");
        return Err(
            WebError::bad_gateway_code("convex_upstream_error", message.to_string())
                .with_context(context)
                .with_header("x-error-phase", "query_status")
                .with_header("x-upstream-service", "convex")
                .with_header("x-upstream-status", status.as_u16().to_string()),
        );
    }

    match body.get("status").and_then(Value::as_str) {
        Some("success") => Ok(body.get("value").cloned().unwrap_or(Value::Null)),
        Some("error") => Err(WebError::bad_gateway_code(
            "convex_upstream_error",
            body.get("errorMessage")
                .and_then(Value::as_str)
                .unwrap_or("Convex 返回 error")
                .to_string(),
        )
        .with_context(context)
        .with_header("x-error-phase", "query_payload")
        .with_header("x-upstream-service", "convex")
        .with_header("x-upstream-status", status.as_u16().to_string())),
        _ => Err(WebError::bad_gateway_code(
            "convex_bad_response",
            format!("未知 Convex 响应: {body}"),
        )
        .with_context(context)
        .with_header("x-error-phase", "query_payload")
        .with_header("x-upstream-service", "convex")
        .with_header("x-upstream-status", status.as_u16().to_string())),
    }
}

pub fn execute_sidebar_dataset_query(
    config: &AppConfig,
    context: &RequestContext,
    plan: &RuntimeQueryExecutionPlan,
) -> Result<Value, WebError> {
    execute_convex_query_plan(config, context, plan)
}