cedros_data/

http.rs

use std::sync::Arc;
#[cfg(feature = "cedros-login-profile")]
use std::time::Duration;

use axum::{
    extract::{Path, State},
    http::{
        header::{self, HeaderName, HeaderValue},
        HeaderMap, Method, StatusCode,
    },
    response::{IntoResponse, Response},
    routing::{get, post, put},
    Json, Router,
};
use serde::Deserialize;
use serde_json::json;
use tokio::net::TcpListener;
use tower_http::{
    cors::{AllowOrigin, CorsLayer},
    trace::TraceLayer,
};
use tracing::{error, info};

use crate::config::HttpServerConfig;
use crate::error::CedrosDataError;
use crate::http_auth::CedrosDataPermission;
#[cfg(feature = "cedros-login-profile")]
use crate::http_auth::{
    derive_permissions_url, enforce_permission_with_cedros_login, AuthzError, CedrosLoginState,
};
use crate::http_auth::{read_permission_for_collection, write_permission_for_collection};
use crate::models::{
    CollectionMode, ContractSchema, ExportSiteRequest, ImportSiteRequest, QueryEntriesRequest,
    RegisterCollectionRequest, RegisterCustomSchemaRequest, RegisterSiteRequest,
    UpsertEntryRequest, VerifyContractRequest,
};
use crate::store::CedrosData;

#[derive(Clone)]
pub(crate) struct AppState {
    pub(crate) store: CedrosData,
    #[cfg(feature = "cedros-login-profile")]
    pub(crate) cedros_auth: Option<CedrosLoginState>,
    #[cfg(feature = "storage")]
    pub(crate) storage: Option<crate::storage::StorageClient>,
}

pub async fn run_http_server(
    store: CedrosData,
    config: HttpServerConfig,
    #[cfg(feature = "storage")] storage: Option<crate::storage::StorageClient>,
) -> crate::error::Result<()> {
    store.migrate().await?;
    store.bootstrap_defaults().await?;
    let app_state = build_state(
        store,
        &config,
        #[cfg(feature = "storage")]
        storage,
    )?;
    let app = build_router(app_state, &config)?;
    let listener = TcpListener::bind(&config.bind_addr).await?;
    info!(bind_addr = %config.bind_addr, "cedros-data HTTP server listening");
    axum::serve(listener, app).await.map_err(Into::into)
}

#[cfg(feature = "cedros-login-profile")]
const CEDROS_LOGIN_CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
#[cfg(feature = "cedros-login-profile")]
const CEDROS_LOGIN_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);

fn build_state(
    store: CedrosData,
    config: &HttpServerConfig,
    #[cfg(feature = "storage")] storage: Option<crate::storage::StorageClient>,
) -> crate::error::Result<AppState> {
    validate_http_auth_configuration(config)?;

    #[cfg(feature = "cedros-login-profile")]
    {
        let cedros_auth = if config.enable_cedros_login_profile {
            let verify_url = config.cedros_login_verify_url.clone().ok_or_else(|| {
                CedrosDataError::InvalidRequest(
                    "cedros_login_verify_url is required when compatibility profile is enabled"
                        .to_string(),
                )
            })?;
            let permissions_url = derive_permissions_url(&verify_url)?;
            let client = reqwest::Client::builder()
                .connect_timeout(CEDROS_LOGIN_CONNECT_TIMEOUT)
                .timeout(CEDROS_LOGIN_REQUEST_TIMEOUT)
                .build()
                .map_err(|error| {
                    CedrosDataError::InvalidRequest(format!(
                        "failed to build cedros-login HTTP client: {error}"
                    ))
                })?;
            Some(CedrosLoginState {
                permissions_url,
                client,
            })
        } else {
            None
        };

        Ok(AppState {
            store,
            cedros_auth,
            #[cfg(feature = "storage")]
            storage,
        })
    }

    #[cfg(not(feature = "cedros-login-profile"))]
    {
        if config.enable_cedros_login_profile {
            return Err(CedrosDataError::InvalidRequest(
                "cedros-login-profile feature is not enabled".to_string(),
            ));
        }
        Ok(AppState {
            store,
            #[cfg(feature = "storage")]
            storage,
        })
    }
}

fn validate_http_auth_configuration(config: &HttpServerConfig) -> crate::error::Result<()> {
    if config.enable_cedros_login_profile || config.allow_unauthenticated_http {
        return Ok(());
    }

    Err(CedrosDataError::InvalidRequest(
        "embedded HTTP server requires authentication by default; pass --allow-unauthenticated for local development or enable --cedros-login-profile".to_string(),
    ))
}

fn build_router(state: AppState, config: &HttpServerConfig) -> crate::error::Result<Router> {
    #[allow(unused_mut)]
    let mut app = Router::new()
        .route("/migrate", post(migrate))
        .route("/site", post(register_site))
        .route("/collections", post(register_collection))
        .route("/custom-schema", post(register_custom_schema))
        .route("/entries/upsert", post(upsert_entry))
        .route("/entries/query", post(query_entries))
        .route("/site/export", get(export_site))
        .route("/import", post(import_site))
        .route("/contract/verify", post(verify_contract))
        .route("/admin/default-pages", get(default_pages))
        .route("/admin/bootstrap", post(bootstrap_site))
        .route(
            "/admin/collections",
            get(list_site_collections).post(register_site_collection),
        )
        .route("/admin/pages", get(list_site_pages))
        .route("/admin/pages/{page_key}", put(upsert_site_page))
        // Public config (no auth)
        .route("/site/config/{key}", get(crate::http_public::public_site_config));

    #[cfg(feature = "storage")]
    {
        use crate::http_storage;
        app = app
            .route("/admin/storage/config", get(http_storage::storage_config))
            .route("/admin/storage/test", post(http_storage::storage_test))
            .route("/admin/media/upload", post(http_storage::media_upload))
            .route("/admin/media/assets", get(http_storage::list_media_assets))
            .route(
                "/admin/media/assets/{asset_id}",
                get(http_storage::get_media_asset)
                    .put(http_storage::update_media_asset)
                    .delete(http_storage::delete_media_asset),
            );
    }

    let mut router = app
        .layer(TraceLayer::new_for_http())
        .with_state(Arc::new(state));

    if let Some(cors_layer) = configured_cors_layer(config)? {
        router = router.layer(cors_layer);
    }

    Ok(router)
}

fn configured_cors_layer(config: &HttpServerConfig) -> crate::error::Result<Option<CorsLayer>> {
    if config.cors_allowed_origins.is_empty() {
        return Ok(None);
    }

    let allow_origin = parse_cors_allowed_origins(&config.cors_allowed_origins)?;
    let cors_layer = CorsLayer::new()
        .allow_origin(allow_origin)
        .allow_methods([Method::GET, Method::POST, Method::PUT, Method::OPTIONS])
        .allow_headers([
            header::AUTHORIZATION,
            header::CONTENT_TYPE,
            header::ACCEPT,
            HeaderName::from_static("x-cedros-org-id"),
        ]);

    Ok(Some(cors_layer))
}

fn parse_cors_allowed_origins(origins: &[String]) -> crate::error::Result<AllowOrigin> {
    if origins.iter().any(|origin| origin == "*") {
        if origins.len() > 1 {
            return Err(CedrosDataError::InvalidRequest(
                "cors origins cannot mix '*' with explicit origins".to_string(),
            ));
        }

        return Ok(AllowOrigin::any());
    }

    let header_values = origins
        .iter()
        .map(|origin| {
            HeaderValue::from_str(origin).map_err(|error| {
                CedrosDataError::InvalidRequest(format!(
                    "invalid cors origin \"{origin}\": {error}"
                ))
            })
        })
        .collect::<crate::error::Result<Vec<_>>>()?;

    Ok(AllowOrigin::list(header_values))
}

async fn migrate(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::DataAdmin).await?;
    state.store.migrate().await?;
    Ok(Json(json!({"ok": true})))
}

async fn register_site(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<RegisterSiteRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::DataAdmin).await?;
    let site = state.store.register_site(request).await?;
    Ok(Json(json!(site)))
}

async fn register_collection(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<RegisterCollectionRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::CollectionsWrite).await?;
    let collection = state.store.register_collection(request).await?;
    Ok(Json(json!(collection)))
}

async fn register_custom_schema(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<RegisterCustomSchemaRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::SchemaWrite).await?;
    let report = state.store.register_custom_schema(request).await?;
    Ok(Json(json!(report)))
}

async fn upsert_entry(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<UpsertEntryRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(
        &state,
        &headers,
        write_permission_for_collection(&request.collection_name),
    )
    .await?;
    let entry = state.store.upsert_entry(request).await?;
    Ok(Json(json!(entry)))
}

async fn query_entries(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<QueryEntriesRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(
        &state,
        &headers,
        read_permission_for_collection(&request.collection_name),
    )
    .await?;
    let visitor_id = request.visitor_id.clone();
    let record_reads = request.entry_keys.len() == 1;
    let mut entries = state.store.query_entries(request).await?;

    if let Some(ref vid) = visitor_id {
        crate::store::metered_reads::validate_visitor_id(vid)?;
        state
            .store
            .apply_content_gating(&mut entries, vid, record_reads)
            .await?;
    }

    Ok(Json(json!(entries)))
}

async fn export_site(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::OpsWrite).await?;
    let export = state.store.export_site(ExportSiteRequest).await?;
    Ok(Json(json!(export)))
}

async fn import_site(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<ImportSiteRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::OpsWrite).await?;
    let result = state.store.import_site(request).await?;
    Ok(Json(json!(result)))
}

async fn verify_contract(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<VerifyContractRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::ContractRead).await?;
    let report = state.store.verify_contract(request).await?;
    Ok(Json(json!(report)))
}

async fn default_pages(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::PagesRead).await?;
    Ok(Json(json!(crate::defaults::default_page_templates())))
}

async fn bootstrap_site(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::OpsWrite).await?;
    let report = state.store.bootstrap_defaults().await?;
    Ok(Json(json!(report)))
}

async fn list_site_collections(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::CollectionsRead).await?;
    let collections = state.store.list_collections().await?;
    Ok(Json(json!(collections)))
}

#[derive(Debug, Deserialize)]
struct RegisterSiteCollectionRequest {
    collection_name: String,
    mode: Option<CollectionMode>,
    table_name: Option<String>,
    strict_contract: Option<ContractSchema>,
}

async fn register_site_collection(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<RegisterSiteCollectionRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::CollectionsWrite).await?;
    let collection = state
        .store
        .register_collection(RegisterCollectionRequest {
            collection_name: request.collection_name,
            mode: request.mode.unwrap_or(CollectionMode::Jsonb),
            table_name: request.table_name,
            strict_contract: request.strict_contract,
        })
        .await?;
    Ok(Json(json!(collection)))
}

async fn list_site_pages(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::PagesRead).await?;
    let pages = state.store.list_default_page_entries().await?;
    Ok(Json(json!(pages)))
}

#[derive(Debug, Deserialize)]
struct UpsertPageRequest {
    payload: serde_json::Value,
}

async fn upsert_site_page(
    Path(page_key): Path<String>,
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(request): Json<UpsertPageRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
    enforce_org_permission(&state, &headers, CedrosDataPermission::PagesWrite).await?;
    let page = state
        .store
        .upsert_default_page_entry(&page_key, request.payload)
        .await?;
    Ok(Json(json!(page)))
}

#[allow(unused_variables)]
pub(crate) async fn enforce_org_permission(
    state: &Arc<AppState>,
    headers: &HeaderMap,
    permission: CedrosDataPermission,
) -> Result<(), AppError> {
    #[cfg(feature = "cedros-login-profile")]
    if let Some(auth_state) = state.cedros_auth.as_ref() {
        enforce_permission_with_cedros_login(auth_state, headers, permission)
            .await
            .map_err(map_authz_error)?;
    }
    Ok(())
}

#[cfg(feature = "cedros-login-profile")]
fn map_authz_error(error: AuthzError) -> AppError {
    if error.status.is_server_error() {
        error!(
            status = %error.status,
            detail = %error.message,
            "authorization request failed"
        );
    }
    AppError::new(error.status, error.message)
}

pub(crate) struct AppError {
    status: StatusCode,
    message: String,
}

impl AppError {
    pub(crate) fn new(status: StatusCode, message: String) -> Self {
        Self { status, message }
    }
}

impl From<CedrosDataError> for AppError {
    fn from(error: CedrosDataError) -> Self {
        match error {
            CedrosDataError::SiteNotConfigured | CedrosDataError::CollectionNotFound(_) => {
                Self::new(StatusCode::NOT_FOUND, error.to_string())
            }
            CedrosDataError::BreakingSchemaChange(_)
            | CedrosDataError::ContractVerificationFailed(_)
            | CedrosDataError::InvalidIdentifier(_)
            | CedrosDataError::InvalidRequest(_)
            | CedrosDataError::InvalidTypeExpression(_) => {
                Self::new(StatusCode::BAD_REQUEST, error.to_string())
            }
            CedrosDataError::MissingPostgresUri => {
                Self::new(StatusCode::BAD_REQUEST, error.to_string())
            }
            CedrosDataError::Storage(_) => {
                error!(detail = %error, "storage request failed");
                Self::new(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "storage error".to_string(),
                )
            }
            CedrosDataError::Database(_)
            | CedrosDataError::Migration(_)
            | CedrosDataError::Json(_)
            | CedrosDataError::Io(_) => {
                error!(detail = %error, "cedros-data request failed");
                Self::new(
                    StatusCode::INTERNAL_SERVER_ERROR,
                    "internal server error".to_string(),
                )
            }
        }
    }
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        (self.status, Json(json!({ "error": self.message }))).into_response()
    }
}

#[cfg(test)]
mod tests;