use crate::app::AppConfig; use crate::context::RequestContext; use crate::error::WebError; use crate::transport::convex::execute_convex_query_plan; use bridge_runtime::{ execute_runtime_input, execute_runtime_query, RuntimeActorWire, RuntimeBridgeContextWire, RuntimeExecutionPlan, RuntimeInput, RuntimeQueryEnvelopeWire, RuntimeQueryExecutionPlan, RuntimeSourceWire, }; use serde_json::Value; pub fn resolve_effective_workspace_id( context: &RequestContext, query_workspace_id: Option<&str>, require_workspace: bool, ) -> Result, WebError> { let query_workspace_id = query_workspace_id .map(str::trim) .filter(|value| !value.is_empty()); let header_workspace_id = context .workspace .workspace_id .as_deref() .map(str::trim) .filter(|value| !value.is_empty()); if let (Some(query_id), Some(header_id)) = (query_workspace_id, header_workspace_id) { if query_id != header_id { return Err(WebError::bad_request_code( "workspace_context_conflict", format!("workspaceId 参数与请求头中的 workspace 不一致: {query_id} != {header_id}"), ) .with_context(context) .with_header("x-error-phase", "workspace_resolve")); } } let effective_workspace_id = query_workspace_id.or(header_workspace_id); if require_workspace && effective_workspace_id.is_none() { return Err(WebError::bad_request_code( "workspace_required", "缺少 workspaceId,请在 query 或请求头中提供有效工作区", ) .with_context(context) .with_header("x-error-phase", "workspace_resolve")); } Ok(effective_workspace_id.map(ToOwned::to_owned)) } pub fn runtime_context( context: &RequestContext, effective_workspace_id: Option<&str>, ) -> RuntimeBridgeContextWire { RuntimeBridgeContextWire { deployment_id: context.workspace.deployment_id.clone(), project_id: context.workspace.project_id.clone(), workspace_id: effective_workspace_id .map(ToOwned::to_owned) .or_else(|| context.workspace.workspace_id.clone()), request_id: context.trace.request_id.clone(), trace_id: context.trace.trace_id.clone(), actor: RuntimeActorWire { actor_type: context.auth.actor_type.clone(), actor_id: context.auth.actor_id.clone(), session_id: context.auth.session_id.clone(), }, source: RuntimeSourceWire { channel: context.source.channel.clone(), client: context.source.client.clone(), }, tenant_id: context.workspace.tenant_id.clone(), auth_token: context.auth.authorization.clone(), idempotency_key: context.source.idempotency_key.clone(), validate_only: false, dry_run: false, } } pub fn build_runtime_query_plan( context: &RequestContext, effective_workspace_id: Option<&str>, query: RuntimeQueryEnvelopeWire, ) -> Result { let runtime_input = RuntimeInput::Query { context: runtime_context(context, effective_workspace_id), query, data: None, }; let RuntimeExecutionPlan::Query(plan) = execute_runtime_input(runtime_input) .map_err(|error| WebError::bad_request(error.message).with_context(context))? else { return Err(WebError::internal("runtime query 未返回 query plan").with_context(context)); }; Ok(plan) } pub async fn fetch_query_data_via_convex( config: &AppConfig, context: &RequestContext, effective_workspace_id: Option<&str>, query: RuntimeQueryEnvelopeWire, ) -> Result { let plan = build_runtime_query_plan(context, effective_workspace_id, query)?; execute_convex_query_plan(config, context, &plan).await } pub fn execute_runtime_query_against_data( context: &RequestContext, effective_workspace_id: Option<&str>, query: RuntimeQueryEnvelopeWire, data: Value, ) -> Result { execute_runtime_query(RuntimeInput::Query { context: runtime_context(context, effective_workspace_id), query, data: Some(data), }) .map_err(|error| WebError::bad_request(error.message).with_context(context)) } pub async fn execute_runtime_query_via_convex( config: &AppConfig, context: &RequestContext, effective_workspace_id: Option<&str>, query: RuntimeQueryEnvelopeWire, ) -> Result { let data = fetch_query_data_via_convex(config, context, effective_workspace_id, query.clone()).await?; execute_runtime_query_against_data(context, effective_workspace_id, query, data) }