feat: implement admin api v1

2026-03-25 19:12:58 +03:00
parent 0f72f2a6cb
commit aef87f582f
14 changed files with 1880 additions and 57 deletions
@@ -0,0 +1,673 @@
+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_registry::{
+    CreateVersionRequest, OperationSampleMetadata, OperationSummary, OperationVersionRecord,
+    PostgresRegistry, PublishRequest, RegistryOperation, SampleKind, SaveAuthProfileRequest,
+    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 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<String>,
+}
+
+#[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<Value>,
+}
+
+#[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<String>,
+}
+
+#[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<u32>,
+    #[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<String>,
+}
+
+impl AdminService {
+    pub fn new(registry: PostgresRegistry, storage_root: PathBuf) -> Self {
+        Self {
+            registry,
+            runtime: RuntimeExecutor::new(),
+            storage: LocalArtifactStorage::new(storage_root),
+        }
+    }
+
+    pub async fn list_operations(&self) -> Result<Vec<OperationSummary>, ApiError> {
+        Ok(self.registry.list_operations().await?)
+    }
+
+    pub async fn get_operation(
+        &self,
+        operation_id: &OperationId,
+    ) -> Result<OperationSummary, ApiError> {
+        self.registry
+            .get_operation_summary(operation_id)
+            .await?
+            .ok_or_else(|| {
+                ApiError::not_found(format!("operation {} was not found", operation_id.as_str()))
+            })
+    }
+
+    pub async fn get_operation_version(
+        &self,
+        operation_id: &OperationId,
+        version: u32,
+    ) -> Result<OperationVersionRecord, ApiError> {
+        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()
+                ))
+            })
+    }
+
+    pub async fn create_operation(
+        &self,
+        payload: OperationPayload,
+    ) -> Result<CreatedOperationResponse, ApiError> {
+        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?;
+
+        Ok(CreatedOperationResponse {
+            operation_id: operation_id.as_str().to_owned(),
+            version: 1,
+            status: OperationStatus::Draft,
+        })
+    }
+
+    pub async fn create_version(
+        &self,
+        operation_id: &OperationId,
+        payload: NewVersionPayload,
+    ) -> Result<CreatedOperationResponse, ApiError> {
+        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?;
+
+        Ok(CreatedOperationResponse {
+            operation_id: operation_id.as_str().to_owned(),
+            version,
+            status: OperationStatus::Draft,
+        })
+    }
+
+    pub async fn publish_operation(
+        &self,
+        operation_id: &OperationId,
+        version: u32,
+    ) -> Result<PublishResponse, ApiError> {
+        let published_at = now_string()?;
+        self.registry
+            .publish_operation(PublishRequest {
+                operation_id,
+                version,
+                published_at: &published_at,
+                published_by: None,
+            })
+            .await?;
+
+        Ok(PublishResponse {
+            operation_id: operation_id.as_str().to_owned(),
+            published_version: version,
+            published_at,
+        })
+    }
+
+    pub async fn run_test(
+        &self,
+        operation_id: &OperationId,
+        payload: TestRunPayload,
+    ) -> Result<TestRunResult, ApiError> {
+        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)],
+            }),
+        }
+    }
+
+    pub async fn list_auth_profiles(&self) -> Result<Vec<AuthProfile>, ApiError> {
+        Ok(self.registry.list_auth_profiles().await?)
+    }
+
+    pub async fn get_auth_profile(
+        &self,
+        auth_profile_id: &AuthProfileId,
+    ) -> Result<AuthProfile, ApiError> {
+        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()
+                ))
+            })
+    }
+
+    pub async fn create_auth_profile(
+        &self,
+        payload: AuthProfilePayload,
+    ) -> Result<AuthProfile, ApiError> {
+        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?;
+
+        Ok(profile)
+    }
+
+    pub async fn export_operation(
+        &self,
+        operation_id: &OperationId,
+        query: ExportQuery,
+    ) -> Result<String, ApiError> {
+        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()))
+    }
+
+    pub async fn import_operation(
+        &self,
+        query: ImportQuery,
+        yaml_document: &str,
+    ) -> Result<ImportResponse, ApiError> {
+        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?;
+
+                    Ok(ImportResponse {
+                        operation_id: created.operation_id,
+                        version: created.version,
+                        import_mode: ImportMode::Upsert,
+                        warnings: Vec::new(),
+                    })
+                } else {
+                    let created = self.create_operation(payload).await?;
+                    Ok(ImportResponse {
+                        operation_id: created.operation_id,
+                        version: created.version,
+                        import_mode: ImportMode::Upsert,
+                        warnings: Vec::new(),
+                    })
+                }
+            }
+        }
+    }
+
+    pub async fn save_json_sample(
+        &self,
+        operation_id: &OperationId,
+        sample_kind: SampleKind,
+        payload: &Value,
+    ) -> Result<OperationSampleMetadata, ApiError> {
+        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?;
+
+        Ok(metadata)
+    }
+
+    pub async fn generate_draft(
+        &self,
+        operation_id: &OperationId,
+        payload: GenerateDraftPayload,
+    ) -> Result<DraftGenerationResult, ApiError> {
+        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(),
+        };
+
+        Ok(DraftGenerationResult {
+            generated_draft,
+            input_schema,
+            output_schema,
+            input_mapping,
+            output_mapping,
+        })
+    }
+
+    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<Option<OperationSummary>, ApiError> {
+        Ok(self
+            .registry
+            .list_operations()
+            .await?
+            .into_iter()
+            .find(|operation| operation.name == name))
+    }
+}
+
+fn build_request_preview(
+    mapping: &MappingSet,
+    input: &Value,
+) -> Result<Value, mcpaas_mapping::MappingError> {
+    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,
+        "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<OperationSampleMetadata> {
+    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<String, ApiError> {
+    OffsetDateTime::now_utc()
+        .format(&Rfc3339)
+        .map_err(|error| ApiError::internal(error.to_string()))
+}