From 0f509b1317dbfa4741b3406d6363bd343b7486fd Mon Sep 17 00:00:00 2001 From: github-ops Date: Sun, 21 Jun 2026 09:07:27 +0000 Subject: [PATCH] Split admin service feature modules --- apps/admin-api/src/service.rs | 373 +----------------------- apps/admin-api/src/service/api_keys.rs | 111 +++++++ apps/admin-api/src/service/samples.rs | 137 +++++++++ apps/admin-api/src/service/upstreams.rs | 141 +++++++++ 4 files changed, 402 insertions(+), 360 deletions(-) create mode 100644 apps/admin-api/src/service/api_keys.rs create mode 100644 apps/admin-api/src/service/samples.rs create mode 100644 apps/admin-api/src/service/upstreams.rs diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 0a83f42..3641dc8 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -4,37 +4,36 @@ use std::sync::Arc; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use crank_core::{ - AgentId, AuditSink, AuthProfile, AuthProfileId, CapabilityProfile, CommunityCapabilityProfile, - EditionCapabilities, ExecutionMode, GeneratedDraft, GeneratedDraftStatus, IdentityError, - IdentityProvider, InvocationLevel, InvocationLog, InvocationLogId, InvocationSource, - InvocationStatus, MembershipRole, NoopAuditSink, OperationId, OperationSecurityLevel, - OperationStatus, OwnerOnlyPolicyEngine, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyScope, - PlatformApiKeyStatus, PolicyEngine, ProductEdition, Protocol, ResponseCachePolicy, SampleId, - Target, ToolQualityMappingRule, ToolQualityMappingSet, ToolQualitySchemaKind, - ToolQualitySchemaNode, UsagePeriod, WizardState, WorkspaceId, + AgentId, AuditSink, AuthProfile, CapabilityProfile, CommunityCapabilityProfile, + EditionCapabilities, ExecutionMode, IdentityError, IdentityProvider, InvocationLevel, + InvocationLog, InvocationLogId, InvocationSource, InvocationStatus, MembershipRole, + NoopAuditSink, OperationId, OperationSecurityLevel, OperationStatus, OwnerOnlyPolicyEngine, + PlatformApiKeyScope, PlatformApiKeyStatus, PolicyEngine, ProductEdition, Protocol, + ResponseCachePolicy, Target, ToolQualityMappingRule, ToolQualityMappingSet, + ToolQualitySchemaKind, ToolQualitySchemaNode, UsagePeriod, WizardState, WorkspaceId, }; use crank_mapping::{JsonPathRoot, MappingRule, MappingSet, infer_mapping_from_samples}; use crank_registry::{ - AgentSummary, CreateInvocationLogRequest, CreatePlatformApiKeyRequest, ListInvocationLogsQuery, - OperationAgentRef, OperationSampleMetadata, OperationSummary, OperationUsageSummary, - PlatformApiKeyRecord, PostgresRegistry, RegistryOperation, SampleKind, - SaveSampleMetadataRequest, SaveWorkspaceUpstreamRequest, UsageBucket, WorkspaceUpstream, - WorkspaceUpstreamId, + AgentSummary, CreateInvocationLogRequest, ListInvocationLogsQuery, OperationAgentRef, + OperationSummary, OperationUsageSummary, PostgresRegistry, RegistryOperation, SampleKind, + UsageBucket, }; use crank_runtime::{PreparedRequest, ResolvedAuth, RuntimeError, RuntimeExecutor, SecretCrypto}; use crank_schema::{Schema, SchemaKind}; use serde_json::{Value, json}; use sha2::{Digest, Sha256}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; -use tracing::{info, instrument}; use uuid::Uuid; mod agents; +mod api_keys; mod auth; mod import_export; mod observability; mod operations; +mod samples; mod secrets; +mod upstreams; mod workspaces; use crate::{auth::AuthSettings, error::ApiError, storage::LocalArtifactStorage}; @@ -203,102 +202,6 @@ impl AdminService { }) } - #[instrument(skip(self))] - pub async fn list_agent_platform_api_keys( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - ) -> Result, ApiError> { - self.ensure_workspace_exists(workspace_id).await?; - self.registry - .get_agent_summary(workspace_id, agent_id) - .await? - .ok_or_else(|| { - ApiError::not_found_with_context( - format!("agent {} was not found", agent_id.as_str()), - json!({ "agent_id": agent_id.as_str() }), - ) - })?; - Ok(self - .registry - .list_platform_api_keys_for_agent(workspace_id, agent_id) - .await?) - } - - #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_name = %payload.name))] - pub async fn create_agent_platform_api_key( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - payload: PlatformApiKeyPayload, - ) -> Result { - self.ensure_workspace_exists(workspace_id).await?; - self.registry - .get_agent_summary(workspace_id, agent_id) - .await? - .ok_or_else(|| { - ApiError::not_found_with_context( - format!("agent {} was not found", agent_id.as_str()), - json!({ "agent_id": agent_id.as_str() }), - ) - })?; - - let secret = generate_access_secret("crk"); - let api_key = PlatformApiKeyRecord { - api_key: PlatformApiKey { - id: PlatformApiKeyId::new(new_prefixed_id("pk")), - workspace_id: workspace_id.clone(), - agent_id: Some(agent_id.clone()), - name: payload.name, - prefix: secret.chars().take(16).collect(), - scopes: payload.scopes, - status: PlatformApiKeyStatus::Active, - created_at: OffsetDateTime::now_utc(), - last_used_at: None, - }, - }; - - self.registry - .create_platform_api_key(CreatePlatformApiKeyRequest { - api_key: &api_key.api_key, - secret_hash: &hash_access_secret(&secret), - }) - .await?; - - Ok(CreatedPlatformApiKeyResponse { api_key, secret }) - } - - #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_id = %key_id.as_str()))] - pub async fn revoke_agent_platform_api_key( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - key_id: &PlatformApiKeyId, - ) -> Result<(), ApiError> { - self.registry - .revoke_platform_api_key_for_agent( - workspace_id, - agent_id, - key_id, - &OffsetDateTime::now_utc(), - ) - .await?; - Ok(()) - } - - #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_id = %key_id.as_str()))] - pub async fn delete_agent_platform_api_key( - &self, - workspace_id: &WorkspaceId, - agent_id: &AgentId, - key_id: &PlatformApiKeyId, - ) -> Result<(), ApiError> { - self.registry - .delete_platform_api_key_for_agent(workspace_id, agent_id, key_id) - .await?; - Ok(()) - } - pub async fn list_protocol_capabilities(&self) -> Vec { let capabilities = self.get_capabilities().await; capabilities @@ -387,214 +290,6 @@ impl AdminService { ResolvedAuth::from_profile(auth_profile, &secrets) } - #[instrument(skip(self))] - pub async fn list_workspace_upstreams( - &self, - workspace_id: &WorkspaceId, - ) -> Result, ApiError> { - self.ensure_workspace_exists(workspace_id).await?; - self.ensure_default_workspace_upstreams(workspace_id) - .await?; - Ok(self.registry.list_workspace_upstreams(workspace_id).await?) - } - - #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), upstream_name = %payload.name))] - pub async fn save_workspace_upstream( - &self, - workspace_id: &WorkspaceId, - upstream_id: Option<&WorkspaceUpstreamId>, - payload: UpstreamPayload, - ) -> Result { - self.ensure_workspace_exists(workspace_id).await?; - validate_upstream_payload(&payload)?; - - if let Some(auth_profile_id) = payload.auth_profile_id.as_deref() { - self.get_auth_profile( - workspace_id, - &AuthProfileId::new(auth_profile_id.to_owned()), - ) - .await?; - } - - let now = OffsetDateTime::now_utc(); - let existing = match upstream_id { - Some(id) => { - self.registry - .get_workspace_upstream(workspace_id, id) - .await? - } - None => None, - }; - if upstream_id.is_some() && existing.is_none() { - return Err(ApiError::not_found_with_context( - "upstream was not found", - json!({ "upstream_id": upstream_id.map(|id| id.as_str()).unwrap_or_default() }), - )); - } - let upstream = WorkspaceUpstream { - id: existing - .as_ref() - .map(|item| item.id.clone()) - .unwrap_or_else(|| WorkspaceUpstreamId::new(new_prefixed_id("upstream"))), - workspace_id: workspace_id.clone(), - name: payload.name.trim().to_owned(), - base_url: normalize_base_url(&payload.base_url), - static_headers: payload.static_headers, - auth_profile_id: payload - .auth_profile_id - .filter(|value| !value.trim().is_empty()), - created_at: existing.as_ref().map(|item| item.created_at).unwrap_or(now), - updated_at: now, - }; - - self.registry - .save_workspace_upstream(SaveWorkspaceUpstreamRequest { - upstream: &upstream, - }) - .await?; - info!(upstream_id = %upstream.id.as_str(), "workspace upstream saved"); - - Ok(upstream) - } - - async fn ensure_default_workspace_upstreams( - &self, - workspace_id: &WorkspaceId, - ) -> Result<(), ApiError> { - let existing = self.registry.list_workspace_upstreams(workspace_id).await?; - if existing.iter().any(|item| item.name == "Open Meteo") { - return Ok(()); - } - - let now = OffsetDateTime::now_utc(); - let upstream = WorkspaceUpstream { - id: WorkspaceUpstreamId::new(new_prefixed_id("upstream")), - workspace_id: workspace_id.clone(), - name: "Open Meteo".to_owned(), - base_url: "https://api.open-meteo.com".to_owned(), - static_headers: json!({}), - auth_profile_id: None, - created_at: now, - updated_at: now, - }; - self.registry - .save_workspace_upstream(SaveWorkspaceUpstreamRequest { - upstream: &upstream, - }) - .await?; - Ok(()) - } - - #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))] - pub async fn save_json_sample( - &self, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - sample_kind: SampleKind, - payload: &Value, - ) -> Result { - let summary = self.get_operation(workspace_id, operation_id).await?; - let version = summary.current_draft_version; - let sample_id = SampleId::new(new_prefixed_id("sample")); - let now = OffsetDateTime::now_utc(); - 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, - workspace_id: &WorkspaceId, - operation_id: &OperationId, - payload: GenerateDraftPayload, - ) -> Result { - let summary = self.get_operation(workspace_id, 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> { self.validate_operation_capabilities(payload.protocol, payload.security_level)?; validate_protocol_target(payload.protocol, &payload.target)?; @@ -1024,37 +719,6 @@ fn validate_protocol_target(protocol: Protocol, target: &Target) -> Result<(), A Err(ApiError::validation("protocol and target kind must match")) } -fn validate_upstream_payload(payload: &UpstreamPayload) -> Result<(), ApiError> { - if payload.name.trim().is_empty() { - return Err(ApiError::validation("upstream name is required")); - } - let base_url = normalize_base_url(&payload.base_url); - if !(base_url.starts_with("https://") || base_url.starts_with("http://")) { - return Err(ApiError::validation( - "upstream base_url must start with http:// or https://", - )); - } - if !payload.static_headers.is_object() { - return Err(ApiError::validation( - "upstream static_headers must be a JSON object", - )); - } - if let Some(headers) = payload.static_headers.as_object() { - for (key, value) in headers { - if key.trim().is_empty() || !value.is_string() { - return Err(ApiError::validation( - "upstream static_headers must contain string values", - )); - } - } - } - Ok(()) -} - -fn normalize_base_url(value: &str) -> String { - value.trim().trim_end_matches('/').to_owned() -} - fn validate_profile_display_name(value: &str) -> Result { let display_name = value.trim(); if display_name.is_empty() { @@ -1090,17 +754,6 @@ fn validate_profile_email(value: &str) -> Result { Ok(email) } -fn latest_sample_ref( - samples: &[OperationSampleMetadata], - sample_kind: SampleKind, -) -> Option { - samples - .iter() - .rev() - .find(|sample| sample.sample_kind == sample_kind) - .cloned() -} - fn demo_revops_agent_payload() -> AgentPayload { AgentPayload { slug: "revops-copilot".to_owned(), diff --git a/apps/admin-api/src/service/api_keys.rs b/apps/admin-api/src/service/api_keys.rs new file mode 100644 index 0000000..2f72a87 --- /dev/null +++ b/apps/admin-api/src/service/api_keys.rs @@ -0,0 +1,111 @@ +use crank_core::{AgentId, PlatformApiKey, PlatformApiKeyId, PlatformApiKeyStatus, WorkspaceId}; +use crank_registry::{CreatePlatformApiKeyRequest, PlatformApiKeyRecord}; +use serde_json::json; +use time::OffsetDateTime; +use tracing::instrument; + +use crate::{ + error::ApiError, + service::{ + AdminService, CreatedPlatformApiKeyResponse, PlatformApiKeyPayload, generate_access_secret, + hash_access_secret, new_prefixed_id, + }, +}; + +impl AdminService { + #[instrument(skip(self))] + pub async fn list_agent_platform_api_keys( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + ) -> Result, ApiError> { + self.ensure_workspace_exists(workspace_id).await?; + self.registry + .get_agent_summary(workspace_id, agent_id) + .await? + .ok_or_else(|| { + ApiError::not_found_with_context( + format!("agent {} was not found", agent_id.as_str()), + json!({ "agent_id": agent_id.as_str() }), + ) + })?; + Ok(self + .registry + .list_platform_api_keys_for_agent(workspace_id, agent_id) + .await?) + } + + #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_name = %payload.name))] + pub async fn create_agent_platform_api_key( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + payload: PlatformApiKeyPayload, + ) -> Result { + self.ensure_workspace_exists(workspace_id).await?; + self.registry + .get_agent_summary(workspace_id, agent_id) + .await? + .ok_or_else(|| { + ApiError::not_found_with_context( + format!("agent {} was not found", agent_id.as_str()), + json!({ "agent_id": agent_id.as_str() }), + ) + })?; + + let secret = generate_access_secret("crk"); + let api_key = PlatformApiKeyRecord { + api_key: PlatformApiKey { + id: PlatformApiKeyId::new(new_prefixed_id("pk")), + workspace_id: workspace_id.clone(), + agent_id: Some(agent_id.clone()), + name: payload.name, + prefix: secret.chars().take(16).collect(), + scopes: payload.scopes, + status: PlatformApiKeyStatus::Active, + created_at: OffsetDateTime::now_utc(), + last_used_at: None, + }, + }; + + self.registry + .create_platform_api_key(CreatePlatformApiKeyRequest { + api_key: &api_key.api_key, + secret_hash: &hash_access_secret(&secret), + }) + .await?; + + Ok(CreatedPlatformApiKeyResponse { api_key, secret }) + } + + #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_id = %key_id.as_str()))] + pub async fn revoke_agent_platform_api_key( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + key_id: &PlatformApiKeyId, + ) -> Result<(), ApiError> { + self.registry + .revoke_platform_api_key_for_agent( + workspace_id, + agent_id, + key_id, + &OffsetDateTime::now_utc(), + ) + .await?; + Ok(()) + } + + #[instrument(skip(self), fields(workspace_id = %workspace_id.as_str(), agent_id = %agent_id.as_str(), key_id = %key_id.as_str()))] + pub async fn delete_agent_platform_api_key( + &self, + workspace_id: &WorkspaceId, + agent_id: &AgentId, + key_id: &PlatformApiKeyId, + ) -> Result<(), ApiError> { + self.registry + .delete_platform_api_key_for_agent(workspace_id, agent_id, key_id) + .await?; + Ok(()) + } +} diff --git a/apps/admin-api/src/service/samples.rs b/apps/admin-api/src/service/samples.rs new file mode 100644 index 0000000..af6341d --- /dev/null +++ b/apps/admin-api/src/service/samples.rs @@ -0,0 +1,137 @@ +use crank_core::{GeneratedDraft, GeneratedDraftStatus, OperationId, SampleId, WorkspaceId}; +use crank_mapping::{JsonPathRoot, infer_mapping_from_samples}; +use crank_registry::{OperationSampleMetadata, SampleKind, SaveSampleMetadataRequest}; +use crank_schema::Schema; +use serde_json::Value; +use time::OffsetDateTime; +use tracing::{info, instrument}; + +use crate::{ + error::ApiError, + service::{ + AdminService, DraftGenerationResult, GenerateDraftPayload, new_prefixed_id, now_string, + }, +}; + +impl AdminService { + #[instrument(skip(self, payload), fields(operation_id = %operation_id.as_str(), sample_kind = ?sample_kind))] + pub async fn save_json_sample( + &self, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + sample_kind: SampleKind, + payload: &Value, + ) -> Result { + let summary = self.get_operation(workspace_id, operation_id).await?; + let version = summary.current_draft_version; + let sample_id = SampleId::new(new_prefixed_id("sample")); + let now = OffsetDateTime::now_utc(); + 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, + workspace_id: &WorkspaceId, + operation_id: &OperationId, + payload: GenerateDraftPayload, + ) -> Result { + let summary = self.get_operation(workspace_id, 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 latest_sample_ref( + samples: &[OperationSampleMetadata], + sample_kind: SampleKind, +) -> Option { + samples + .iter() + .rev() + .find(|sample| sample.sample_kind == sample_kind) + .cloned() +} diff --git a/apps/admin-api/src/service/upstreams.rs b/apps/admin-api/src/service/upstreams.rs new file mode 100644 index 0000000..799ef09 --- /dev/null +++ b/apps/admin-api/src/service/upstreams.rs @@ -0,0 +1,141 @@ +use crank_core::{AuthProfileId, WorkspaceId}; +use crank_registry::{SaveWorkspaceUpstreamRequest, WorkspaceUpstream, WorkspaceUpstreamId}; +use serde_json::json; +use time::OffsetDateTime; +use tracing::{info, instrument}; + +use crate::{ + error::ApiError, + service::{AdminService, UpstreamPayload, new_prefixed_id}, +}; + +impl AdminService { + #[instrument(skip(self))] + pub async fn list_workspace_upstreams( + &self, + workspace_id: &WorkspaceId, + ) -> Result, ApiError> { + self.ensure_workspace_exists(workspace_id).await?; + self.ensure_default_workspace_upstreams(workspace_id) + .await?; + Ok(self.registry.list_workspace_upstreams(workspace_id).await?) + } + + #[instrument(skip(self, payload), fields(workspace_id = %workspace_id.as_str(), upstream_name = %payload.name))] + pub async fn save_workspace_upstream( + &self, + workspace_id: &WorkspaceId, + upstream_id: Option<&WorkspaceUpstreamId>, + payload: UpstreamPayload, + ) -> Result { + self.ensure_workspace_exists(workspace_id).await?; + validate_upstream_payload(&payload)?; + + if let Some(auth_profile_id) = payload.auth_profile_id.as_deref() { + self.get_auth_profile( + workspace_id, + &AuthProfileId::new(auth_profile_id.to_owned()), + ) + .await?; + } + + let now = OffsetDateTime::now_utc(); + let existing = match upstream_id { + Some(id) => { + self.registry + .get_workspace_upstream(workspace_id, id) + .await? + } + None => None, + }; + if upstream_id.is_some() && existing.is_none() { + return Err(ApiError::not_found_with_context( + "upstream was not found", + json!({ "upstream_id": upstream_id.map(|id| id.as_str()).unwrap_or_default() }), + )); + } + let upstream = WorkspaceUpstream { + id: existing + .as_ref() + .map(|item| item.id.clone()) + .unwrap_or_else(|| WorkspaceUpstreamId::new(new_prefixed_id("upstream"))), + workspace_id: workspace_id.clone(), + name: payload.name.trim().to_owned(), + base_url: normalize_base_url(&payload.base_url), + static_headers: payload.static_headers, + auth_profile_id: payload + .auth_profile_id + .filter(|value| !value.trim().is_empty()), + created_at: existing.as_ref().map(|item| item.created_at).unwrap_or(now), + updated_at: now, + }; + + self.registry + .save_workspace_upstream(SaveWorkspaceUpstreamRequest { + upstream: &upstream, + }) + .await?; + info!(upstream_id = %upstream.id.as_str(), "workspace upstream saved"); + + Ok(upstream) + } + + pub(super) async fn ensure_default_workspace_upstreams( + &self, + workspace_id: &WorkspaceId, + ) -> Result<(), ApiError> { + let existing = self.registry.list_workspace_upstreams(workspace_id).await?; + if existing.iter().any(|item| item.name == "Open Meteo") { + return Ok(()); + } + + let now = OffsetDateTime::now_utc(); + let upstream = WorkspaceUpstream { + id: WorkspaceUpstreamId::new(new_prefixed_id("upstream")), + workspace_id: workspace_id.clone(), + name: "Open Meteo".to_owned(), + base_url: "https://api.open-meteo.com".to_owned(), + static_headers: json!({}), + auth_profile_id: None, + created_at: now, + updated_at: now, + }; + self.registry + .save_workspace_upstream(SaveWorkspaceUpstreamRequest { + upstream: &upstream, + }) + .await?; + Ok(()) + } +} + +fn validate_upstream_payload(payload: &UpstreamPayload) -> Result<(), ApiError> { + if payload.name.trim().is_empty() { + return Err(ApiError::validation("upstream name is required")); + } + let base_url = normalize_base_url(&payload.base_url); + if !(base_url.starts_with("https://") || base_url.starts_with("http://")) { + return Err(ApiError::validation( + "upstream base_url must start with http:// or https://", + )); + } + if !payload.static_headers.is_object() { + return Err(ApiError::validation( + "upstream static_headers must be a JSON object", + )); + } + if let Some(headers) = payload.static_headers.as_object() { + for (key, value) in headers { + if key.trim().is_empty() || !value.is_string() { + return Err(ApiError::validation( + "upstream static_headers must contain string values", + )); + } + } + } + Ok(()) +} + +fn normalize_base_url(value: &str) -> String { + value.trim().trim_end_matches('/').to_owned() +}