use std::path::PathBuf; use mcpaas_core::{ AuthConfig, AuthKind, AuthProfile, AuthProfileId, ConfigExport, ExportMode, GeneratedDraft, GeneratedDraftStatus, OperationId, OperationStatus, Protocol, SampleId, Samples, Target, }; use mcpaas_mapping::{JsonPathRoot, MappingSet, infer_mapping_from_samples}; use mcpaas_proto::{ProtoService, services_from_descriptor_set_bytes}; use mcpaas_registry::{ CreateVersionRequest, OperationSampleMetadata, OperationSummary, OperationVersionRecord, PostgresRegistry, PublishRequest, RegistryOperation, SampleKind, SaveAuthProfileRequest, SaveDescriptorMetadataRequest, SaveSampleMetadataRequest, }; use mcpaas_runtime::{PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeOperation}; use mcpaas_schema::Schema; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tracing::{info, instrument}; use uuid::Uuid; use crate::{error::ApiError, storage::LocalArtifactStorage}; #[derive(Clone)] pub struct AdminService { registry: PostgresRegistry, runtime: RuntimeExecutor, storage: LocalArtifactStorage, } #[derive(Clone, Debug, Deserialize, Serialize)] pub struct OperationPayload { pub name: String, pub display_name: String, pub protocol: Protocol, pub target: Target, pub input_schema: Schema, pub output_schema: Schema, pub input_mapping: MappingSet, pub output_mapping: MappingSet, pub execution_config: mcpaas_core::ExecutionConfig, pub tool_description: mcpaas_core::ToolDescription, } #[derive(Clone, Debug, Deserialize)] pub struct NewVersionPayload { #[serde(flatten)] pub operation: OperationPayload, #[serde(default)] pub change_note: Option, } #[derive(Clone, Debug, Deserialize)] pub struct PublishPayload { pub version: u32, } #[derive(Clone, Debug, Deserialize)] pub struct TestRunPayload { pub version: u32, pub input: Value, } #[derive(Clone, Debug, Serialize)] pub struct TestRunResult { pub ok: bool, pub request_preview: Value, pub response_preview: Value, pub errors: Vec, } #[derive(Clone, Debug, Deserialize)] pub struct AuthProfilePayload { pub name: String, pub kind: AuthKind, pub config: AuthConfig, } #[derive(Clone, Debug, Deserialize)] pub struct GenerateDraftPayload { #[serde(default)] pub sources: Vec, } #[derive(Clone, Debug, Serialize)] pub struct DraftGenerationResult { pub generated_draft: GeneratedDraft, pub input_schema: Schema, pub output_schema: Schema, pub input_mapping: MappingSet, pub output_mapping: MappingSet, } #[derive(Clone, Debug, Deserialize)] pub struct ImportQuery { #[serde(default)] pub mode: ImportMode, } #[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)] #[serde(rename_all = "snake_case")] pub enum ImportMode { #[default] Create, Upsert, } #[derive(Clone, Debug, Deserialize)] pub struct ExportQuery { pub version: Option, #[serde(default = "default_export_mode")] pub mode: ExportMode, } #[derive(Clone, Debug, Deserialize, Serialize)] pub struct YamlOperationDocument { pub format_version: String, pub kind: String, pub operation: RegistryOperation, } #[derive(Clone, Debug, Serialize)] pub struct CreatedOperationResponse { pub operation_id: String, pub version: u32, pub status: OperationStatus, } #[derive(Clone, Debug, Serialize)] pub struct PublishResponse { pub operation_id: String, pub published_version: u32, pub published_at: String, } #[derive(Clone, Debug, Serialize)] pub struct ImportResponse { pub operation_id: String, pub version: u32, pub import_mode: ImportMode, pub warnings: Vec, } #[derive(Clone, Debug, Serialize)] pub struct DescriptorUploadResponse { pub descriptor_id: String, pub version: u32, } #[derive(Clone, Debug, Serialize)] pub struct GrpcServiceSummary { pub package: String, pub service: String, pub methods: Vec, } #[derive(Clone, Debug, Serialize)] pub struct GrpcMethodSummary { pub name: String, pub kind: String, pub input_schema: Schema, pub output_schema: Schema, } impl AdminService { pub fn new(registry: PostgresRegistry, storage_root: PathBuf) -> Self { Self { registry, runtime: RuntimeExecutor::new(), storage: LocalArtifactStorage::new(storage_root), } } #[instrument(skip(self))] pub async fn list_operations(&self) -> Result, ApiError> { Ok(self.registry.list_operations().await?) } #[instrument(skip(self))] pub async fn get_operation( &self, operation_id: &OperationId, ) -> Result { self.registry .get_operation_summary(operation_id) .await? .ok_or_else(|| { ApiError::not_found(format!("operation {} was not found", operation_id.as_str())) }) } #[instrument(skip(self))] pub async fn get_operation_version( &self, operation_id: &OperationId, version: u32, ) -> Result { self.registry .get_operation_version(operation_id, version) .await? .ok_or_else(|| { ApiError::not_found(format!( "operation version {version} for {} was not found", operation_id.as_str() )) }) } #[instrument(skip(self, payload), fields(protocol = ?payload.protocol, operation_name = %payload.name))] pub async fn create_operation( &self, payload: OperationPayload, ) -> Result { self.validate_operation_payload(&payload)?; if self.find_operation_by_name(&payload.name).await?.is_some() { return Err(ApiError::conflict(format!( "operation with name {} already exists", payload.name ))); } let now = now_string()?; let operation_id = OperationId::new(new_prefixed_id("op")); let snapshot = RegistryOperation { id: operation_id.clone(), name: payload.name, display_name: payload.display_name, protocol: payload.protocol, status: OperationStatus::Draft, version: 1, target: payload.target, input_schema: payload.input_schema, output_schema: payload.output_schema, input_mapping: payload.input_mapping, output_mapping: payload.output_mapping, execution_config: payload.execution_config, tool_description: payload.tool_description, samples: Some(Samples::default()), generated_draft: None, config_export: Some(ConfigExport { format_version: "1".to_owned(), export_mode: ExportMode::Portable, }), created_at: now.clone(), updated_at: now, published_at: None, }; self.registry.create_operation(&snapshot, None).await?; info!(operation_id = %operation_id.as_str(), version = 1, "operation created"); Ok(CreatedOperationResponse { operation_id: operation_id.as_str().to_owned(), version: 1, status: OperationStatus::Draft, }) } #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), protocol = ?payload.operation.protocol))] pub async fn create_version( &self, operation_id: &OperationId, payload: NewVersionPayload, ) -> Result { self.validate_operation_payload(&payload.operation)?; let summary = self.get_operation(operation_id).await?; let now = now_string()?; let version = summary.current_draft_version + 1; let snapshot = RegistryOperation { id: operation_id.clone(), name: payload.operation.name, display_name: payload.operation.display_name, protocol: payload.operation.protocol, status: OperationStatus::Draft, version, target: payload.operation.target, input_schema: payload.operation.input_schema, output_schema: payload.operation.output_schema, input_mapping: payload.operation.input_mapping, output_mapping: payload.operation.output_mapping, execution_config: payload.operation.execution_config, tool_description: payload.operation.tool_description, samples: Some(Samples::default()), generated_draft: None, config_export: Some(ConfigExport { format_version: "1".to_owned(), export_mode: ExportMode::Portable, }), created_at: summary.created_at, updated_at: now, published_at: None, }; self.registry .create_version(CreateVersionRequest { snapshot: &snapshot, change_note: payload.change_note.as_deref(), created_by: None, }) .await?; info!(operation_id = %operation_id.as_str(), version, "operation version created"); Ok(CreatedOperationResponse { operation_id: operation_id.as_str().to_owned(), version, status: OperationStatus::Draft, }) } #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))] pub async fn publish_operation( &self, operation_id: &OperationId, version: u32, ) -> Result { let published_at = now_string()?; self.registry .publish_operation(PublishRequest { operation_id, version, published_at: &published_at, published_by: None, }) .await?; info!(operation_id = %operation_id.as_str(), version, "operation published"); Ok(PublishResponse { operation_id: operation_id.as_str().to_owned(), published_version: version, published_at, }) } #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), version = payload.version))] pub async fn run_test( &self, operation_id: &OperationId, payload: TestRunPayload, ) -> Result { let record = self .get_operation_version(operation_id, payload.version) .await?; let runtime = RuntimeOperation::from(record.snapshot.clone()); let request_preview = match build_request_preview(&record.snapshot.input_mapping, &payload.input) { Ok(preview) => preview, Err(error) => { return Ok(TestRunResult { ok: false, request_preview: Value::Null, response_preview: Value::Null, errors: vec![crate::error::runtime_test_failure(&RuntimeError::Mapping( error, ))], }); } }; match self.runtime.execute(&runtime, &payload.input).await { Ok(output) => Ok(TestRunResult { ok: true, request_preview, response_preview: output, errors: Vec::new(), }), Err(error) => Ok(TestRunResult { ok: false, request_preview, response_preview: Value::Null, errors: vec![crate::error::runtime_test_failure(&error)], }), } } #[instrument(skip(self))] pub async fn list_auth_profiles(&self) -> Result, ApiError> { Ok(self.registry.list_auth_profiles().await?) } #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("descriptor-set.bin")))] pub async fn upload_descriptor_set( &self, operation_id: &OperationId, source_name: Option<&str>, payload: &[u8], ) -> Result { let summary = self.get_operation(operation_id).await?; if summary.protocol != Protocol::Grpc { return Err(ApiError::validation( "descriptor upload is only allowed for grpc operations", )); } let services = services_from_descriptor_set_bytes(payload) .map_err(|error| ApiError::validation(error.to_string()))?; let descriptor_id = mcpaas_core::DescriptorId::new(new_prefixed_id("desc")); let storage_ref = self .storage .write_descriptor( operation_id, summary.current_draft_version, mcpaas_registry::DescriptorKind::DescriptorSet, &descriptor_id, source_name, payload, ) .await?; self.registry .save_descriptor_metadata(SaveDescriptorMetadataRequest { descriptor: &mcpaas_registry::DescriptorMetadata { id: descriptor_id.clone(), operation_id: Some(operation_id.clone()), version: Some(summary.current_draft_version), descriptor_kind: mcpaas_registry::DescriptorKind::DescriptorSet, storage_ref, source_name: source_name.map(ToOwned::to_owned), package_index: Some( serde_json::to_value(&services) .map_err(|error| ApiError::internal(error.to_string()))?, ), created_at: now_string()?, }, }) .await?; info!( operation_id = %operation_id.as_str(), descriptor_id = %descriptor_id.as_str(), version = summary.current_draft_version, "descriptor set uploaded" ); Ok(DescriptorUploadResponse { descriptor_id: descriptor_id.as_str().to_owned(), version: summary.current_draft_version, }) } #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), source_name = source_name.unwrap_or("schema.proto")))] pub async fn upload_proto_file( &self, operation_id: &OperationId, source_name: Option<&str>, payload: &[u8], ) -> Result { let summary = self.get_operation(operation_id).await?; if summary.protocol != Protocol::Grpc { return Err(ApiError::validation( "proto upload is only allowed for grpc operations", )); } let descriptor_id = mcpaas_core::DescriptorId::new(new_prefixed_id("desc")); let storage_ref = self .storage .write_descriptor( operation_id, summary.current_draft_version, mcpaas_registry::DescriptorKind::ProtoUpload, &descriptor_id, source_name, payload, ) .await?; self.registry .save_descriptor_metadata(SaveDescriptorMetadataRequest { descriptor: &mcpaas_registry::DescriptorMetadata { id: descriptor_id.clone(), operation_id: Some(operation_id.clone()), version: Some(summary.current_draft_version), descriptor_kind: mcpaas_registry::DescriptorKind::ProtoUpload, storage_ref, source_name: source_name.map(ToOwned::to_owned), package_index: None, created_at: now_string()?, }, }) .await?; info!( operation_id = %operation_id.as_str(), descriptor_id = %descriptor_id.as_str(), version = summary.current_draft_version, "proto file uploaded" ); Ok(DescriptorUploadResponse { descriptor_id: descriptor_id.as_str().to_owned(), version: summary.current_draft_version, }) } #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version))] pub async fn list_grpc_services( &self, operation_id: &OperationId, version: Option, ) -> Result, ApiError> { let summary = self.get_operation(operation_id).await?; if summary.protocol != Protocol::Grpc { return Err(ApiError::validation( "grpc services are only available for grpc operations", )); } let version = version.unwrap_or(summary.current_draft_version); let descriptor = self .registry .list_descriptor_metadata(operation_id, version) .await? .into_iter() .rev() .find(|descriptor| { descriptor.descriptor_kind == mcpaas_registry::DescriptorKind::DescriptorSet }) .ok_or_else(|| ApiError::not_found("descriptor set was not uploaded"))?; let bytes = self.storage.read_bytes(&descriptor.storage_ref).await?; let services = services_from_descriptor_set_bytes(&bytes) .map_err(|error| ApiError::validation(error.to_string()))?; services .into_iter() .map(proto_service_summary) .collect::, _>>() } #[instrument(skip(self))] pub async fn get_auth_profile( &self, auth_profile_id: &AuthProfileId, ) -> Result { self.registry .get_auth_profile(auth_profile_id) .await? .ok_or_else(|| { ApiError::not_found(format!( "auth profile {} was not found", auth_profile_id.as_str() )) }) } #[instrument(skip(self, payload), fields(auth_profile_name = %payload.name, auth_kind = ?payload.kind))] pub async fn create_auth_profile( &self, payload: AuthProfilePayload, ) -> Result { validate_auth_profile_kind(payload.kind, &payload.config)?; let now = now_string()?; let profile = AuthProfile { id: AuthProfileId::new(new_prefixed_id("auth")), name: payload.name, kind: payload.kind, config: payload.config, created_at: now.clone(), updated_at: now, }; self.registry .save_auth_profile(SaveAuthProfileRequest { profile: &profile }) .await?; info!(auth_profile_id = %profile.id.as_str(), "auth profile created"); Ok(profile) } #[instrument(skip(self), fields(operation_id = %operation_id.as_str(), version = query.version.unwrap_or_default(), mode = ?query.mode))] pub async fn export_operation( &self, operation_id: &OperationId, query: ExportQuery, ) -> Result { let version = match query.version { Some(version) => version, None => { self.get_operation(operation_id) .await? .current_draft_version } }; let record = self.get_operation_version(operation_id, version).await?; let document = YamlOperationDocument { format_version: "1".to_owned(), kind: "operation".to_owned(), operation: RegistryOperation { config_export: Some(ConfigExport { format_version: "1".to_owned(), export_mode: query.mode, }), ..record.snapshot }, }; serde_yaml::to_string(&document).map_err(|error| ApiError::internal(error.to_string())) } #[instrument(skip(self, yaml_document), fields(mode = ?query.mode))] pub async fn import_operation( &self, query: ImportQuery, yaml_document: &str, ) -> Result { let document: YamlOperationDocument = serde_yaml::from_str(yaml_document) .map_err(|error| ApiError::validation(error.to_string()))?; if document.kind != "operation" { return Err(ApiError::validation("yaml kind must be operation")); } let payload = OperationPayload { name: document.operation.name.clone(), display_name: document.operation.display_name.clone(), protocol: document.operation.protocol, target: document.operation.target.clone(), input_schema: document.operation.input_schema.clone(), output_schema: document.operation.output_schema.clone(), input_mapping: document.operation.input_mapping.clone(), output_mapping: document.operation.output_mapping.clone(), execution_config: document.operation.execution_config.clone(), tool_description: document.operation.tool_description.clone(), }; match query.mode { ImportMode::Create => { let created = self.create_operation(payload).await?; Ok(ImportResponse { operation_id: created.operation_id, version: created.version, import_mode: ImportMode::Create, warnings: Vec::new(), }) } ImportMode::Upsert => { if let Some(existing) = self .find_operation_by_name(&document.operation.name) .await? { let created = self .create_version( &existing.id, NewVersionPayload { operation: payload, change_note: Some("yaml upsert".to_owned()), }, ) .await?; let response = ImportResponse { operation_id: created.operation_id, version: created.version, import_mode: ImportMode::Upsert, warnings: Vec::new(), }; info!( operation_id = %response.operation_id, version = response.version, "operation imported by upsert" ); Ok(response) } else { let created = self.create_operation(payload).await?; let response = ImportResponse { operation_id: created.operation_id, version: created.version, import_mode: ImportMode::Upsert, warnings: Vec::new(), }; info!( operation_id = %response.operation_id, version = response.version, "operation imported by upsert" ); Ok(response) } } } } #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))] pub async fn save_json_sample( &self, operation_id: &OperationId, sample_kind: SampleKind, payload: &Value, ) -> Result { let summary = self.get_operation(operation_id).await?; let version = summary.current_draft_version; let sample_id = SampleId::new(new_prefixed_id("sample")); let now = now_string()?; let file_name = match sample_kind { SampleKind::InputJson => "input.json", SampleKind::OutputJson => "output.json", SampleKind::YamlImportSource => "source.yaml", }; let storage_ref = self .storage .write_json_sample(operation_id, version, sample_kind, &sample_id, payload) .await?; let metadata = OperationSampleMetadata { id: sample_id, operation_id: operation_id.clone(), version, sample_kind, storage_ref, content_type: "application/json".to_owned(), file_name: Some(file_name.to_owned()), created_at: now, }; self.registry .save_sample_metadata(SaveSampleMetadataRequest { sample: &metadata }) .await?; info!( operation_id = %operation_id.as_str(), sample_id = %metadata.id.as_str(), version, "json sample saved" ); Ok(metadata) } #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str()))] pub async fn generate_draft( &self, operation_id: &OperationId, payload: GenerateDraftPayload, ) -> Result { let summary = self.get_operation(operation_id).await?; let samples = self .registry .list_sample_metadata(operation_id, summary.current_draft_version) .await?; let input_sample = latest_sample_ref(&samples, SampleKind::InputJson) .ok_or_else(|| ApiError::validation("input_json sample was not found"))?; let output_sample = latest_sample_ref(&samples, SampleKind::OutputJson) .ok_or_else(|| ApiError::validation("output_json sample was not found"))?; let input_value = self.storage.read_json(&input_sample.storage_ref).await?; let output_value = self.storage.read_json(&output_sample.storage_ref).await?; let input_schema = Schema::from_json_sample(&input_value); let output_schema = Schema::from_json_sample(&output_value); let input_mapping = infer_mapping_from_samples( &input_value, JsonPathRoot::Mcp, &input_value, JsonPathRoot::RequestBody, ); let output_mapping = infer_mapping_from_samples( &output_value, JsonPathRoot::ResponseBody, &output_value, JsonPathRoot::Output, ); let source_types = if payload.sources.is_empty() { vec![ "input_json_sample".to_owned(), "output_json_sample".to_owned(), ] } else { payload.sources }; let generated_draft = GeneratedDraft { status: GeneratedDraftStatus::Available, source_types, generated_at: Some(now_string()?), input_schema_generated: true, output_schema_generated: true, input_mapping_generated: true, output_mapping_generated: true, warnings: Vec::new(), }; let result = DraftGenerationResult { generated_draft, input_schema, output_schema, input_mapping, output_mapping, }; info!(operation_id = %operation_id.as_str(), "draft generated from samples"); Ok(result) } fn validate_operation_payload(&self, payload: &OperationPayload) -> Result<(), ApiError> { validate_protocol_target(payload.protocol, &payload.target)?; payload.input_mapping.validate_paths()?; payload.output_mapping.validate_paths()?; Ok(()) } async fn find_operation_by_name( &self, name: &str, ) -> Result, ApiError> { Ok(self .registry .list_operations() .await? .into_iter() .find(|operation| operation.name == name)) } } fn build_request_preview( mapping: &MappingSet, input: &Value, ) -> Result { let mapped = mapping.apply(&json!({ "mcp": input }))?; let prepared = PreparedRequest::from_mapping_output(&mapped).map_err(|error| { mcpaas_mapping::MappingError::InvalidJsonPath { path: error.to_string(), } })?; Ok(json!({ "path": prepared.path_params, "query": prepared.query_params, "headers": prepared.headers, "grpc": prepared.grpc.unwrap_or(Value::Null), "variables": prepared.variables.unwrap_or(Value::Null), "body": prepared.body.unwrap_or(Value::Null) })) } fn validate_protocol_target(protocol: Protocol, target: &Target) -> Result<(), ApiError> { let is_match = matches!( (protocol, target), (Protocol::Rest, Target::Rest(_)) | (Protocol::Graphql, Target::Graphql(_)) | (Protocol::Grpc, Target::Grpc(_)) ); if is_match { return Ok(()); } Err(ApiError::validation("protocol and target kind must match")) } fn validate_auth_profile_kind(kind: AuthKind, config: &AuthConfig) -> Result<(), ApiError> { let is_match = matches!( (kind, config), (AuthKind::Bearer, AuthConfig::Bearer(_)) | (AuthKind::Basic, AuthConfig::Basic(_)) | (AuthKind::ApiKeyHeader, AuthConfig::ApiKeyHeader(_)) | (AuthKind::ApiKeyQuery, AuthConfig::ApiKeyQuery(_)) ); if is_match { return Ok(()); } Err(ApiError::validation("auth kind and config must match")) } fn latest_sample_ref( samples: &[OperationSampleMetadata], sample_kind: SampleKind, ) -> Option { samples .iter() .rev() .find(|sample| sample.sample_kind == sample_kind) .cloned() } fn default_export_mode() -> ExportMode { ExportMode::Portable } fn new_prefixed_id(prefix: &str) -> String { format!("{prefix}_{}", Uuid::now_v7().simple()) } fn now_string() -> Result { OffsetDateTime::now_utc() .format(&Rfc3339) .map_err(|error| ApiError::internal(error.to_string())) } fn proto_service_summary(service: ProtoService) -> Result { let methods = service .unary_methods() .map(|method| { Ok(GrpcMethodSummary { name: method.name.clone(), kind: "unary".to_owned(), input_schema: mcpaas_proto::message_to_schema(&method.input) .map_err(|error| ApiError::validation(error.to_string()))?, output_schema: mcpaas_proto::message_to_schema(&method.output) .map_err(|error| ApiError::validation(error.to_string()))?, }) }) .collect::, ApiError>>()?; Ok(GrpcServiceSummary { package: service.package, service: service.name, methods, }) }