crank/apps/admin-api/src/service/upstreams.rs

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<Vec<WorkspaceUpstream>, 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<WorkspaceUpstream, ApiError> {
        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()
}