windmill/backend/windmill-api-settings/src/lib.rs

/*
 * Author: Ruben Fiszel
 * Copyright: Windmill Labs, Inc 2023
 * This file and its contents are licensed under the AGPLv3 License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-AGPL for a copy of the license.
 */

use std::{
    collections::{BTreeSet, HashMap},
    time::Duration,
};

#[cfg(feature = "parquet")]
mod audit_logs_s3;
#[cfg(feature = "parquet")]
mod audit_logs_s3_backfill;
#[cfg(feature = "parquet")]
mod background_task;
#[cfg(all(feature = "private", feature = "enterprise"))]
mod datatable_roles_ee;
mod datatable_roles_oss;
#[cfg(feature = "private")]
mod ee;
pub mod ee_oss;
#[cfg(feature = "parquet")]
mod log_cleanup;
#[cfg(feature = "parquet")]
mod storage_usage;

use windmill_api_auth::{require_devops_role, require_super_admin, ApiAuthed};
use windmill_common::utils::HTTP_CLIENT_PERMISSIVE as HTTP_CLIENT;
use windmill_common::DB;

use ee_oss::validate_license_key;
use windmill_common::usernames::generate_instance_username_for_all_users;

use axum::{
    body::Body,
    extract::{Extension, Path, Query},
    response::Response,
    routing::{get, post},
    Json, Router,
};

use serde::{Deserialize, Serialize};
use windmill_ai::ai_cache::bump_instance_ai_config_revision;
#[cfg(feature = "enterprise")]
use windmill_common::ee_oss::{send_critical_alert, CriticalAlertKind, CriticalErrorChannel};
#[cfg(all(feature = "private", feature = "enterprise"))]
use windmill_common::secret_backend::{
    AwsSecretsManagerSettings, AzureKeyVaultSettings, SecretMigrationReport, VaultSettings,
};
use windmill_common::{
    ee_oss::{get_license_plan, LicensePlan},
    email_oss::{send_email_plain_text, SMTP_ENABLED},
    error::{self, pg_error_message, JsonResult, Result},
    get_database_url,
    global_settings::{
        ACCENT_COLOR_SETTING, AI_CONFIG_SETTING, APP_WORKSPACED_ROUTE_SETTING,
        AUTOMATE_USERNAME_CREATION_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, CUSTOM_TAGS_SETTING,
        DEFAULT_TAGS_WORKSPACES_SETTING, DISABLE_HUB_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS,
        EXTERNAL_INSTANCE_PG_SETTING,
        GITHUB_APP_WEBHOOK_BASE_URL_SETTING, HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING,
        HTTP_ROUTE_WORKSPACED_ROUTE_SETTING, HUB_ACCESSIBLE_URL_SETTING, HUB_BASE_URL_SETTING,
        INSTANCE_BANNER_SETTING, MAX_RETENTION_OVERRIDE_WORKSPACES,
        MAX_TOKEN_EXPIRATION_DAYS_SETTING, MCP_DISABLE_TOKEN_QUERY_PARAM_SETTING,
        RETENTION_PERIOD_SECS_OVERRIDES_SETTING, RUFF_CONFIG_SETTING, UNIQUE_ID_SETTING,
        WORKSPACE_FAIRNESS_DURATION_SECS_SETTING, WORKSPACE_FAIRNESS_ENABLED_SETTING,
        WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING,
        WS_BASE_URL_SETTING,
    },
    instance_config::{self, ApplyMode, InstanceConfig},
    server::Smtp,
};
use windmill_common::{
    error::to_anyhow,
    worker::{reload_custom_tags_setting, CLOUD_HOSTED},
    PgDatabase,
};

/// Unauthenticated settings routes.
///
/// Used by the extra container (LSP service) to fetch non-sensitive instance
/// configuration like the shared ruff.toml content without needing to carry
/// a credential.
pub fn unauthed_service() -> Router {
    Router::new().route("/ruff_config", get(get_ruff_config_unauthed))
}

/// Public endpoint that returns the instance-level ruff config as plain text
/// TOML. Returns an empty body when unset.
///
/// This is intentionally unauthenticated: ruff config is lint/format policy,
/// not a credential, and the extra container needs to pull it from any
/// deployment topology (docker-compose, k8s, local dev) without the extra
/// burden of shared secrets.
async fn get_ruff_config_unauthed(Extension(db): Extension<DB>) -> error::Result<Response> {
    let value = sqlx::query_scalar!(
        "SELECT value FROM global_settings WHERE name = $1",
        RUFF_CONFIG_SETTING
    )
    .fetch_optional(&db)
    .await?;

    let body = value
        .and_then(|v| v.as_str().map(|s| s.to_string()))
        .unwrap_or_default();

    Ok(Response::builder()
        .status(200)
        .header("content-type", "text/plain; charset=utf-8")
        .header("cache-control", "no-store")
        .body(Body::from(body))
        .unwrap())
}

/// The announcement banner and accent color, which every signed-in session reads on each
/// full page load.
async fn get_instance_ui(
    Extension(db): Extension<DB>,
    _authed: ApiAuthed,
) -> JsonResult<windmill_common::global_settings::InstanceUi> {
    Ok(Json(
        windmill_common::global_settings::get_instance_ui(&db).await?,
    ))
}

pub fn global_service() -> Router {
    #[warn(unused_mut)]
    let r = Router::new()
        // `/local` is the path in openapi.yaml, so every generated client (getLocal) calls it;
        // `/envs` stays for callers that found the route in the code.
        .route("/local", get(get_local_settings))
        .route("/envs", get(get_local_settings))
        .route(
            "/global/{key}",
            post(set_global_setting).get(get_global_setting),
        )
        .route("/instance_ui", get(get_instance_ui))
        .route("/list_global", get(list_global_settings))
        .route("/github_app_stale_webhooks", get(github_app_stale_webhooks))
        .route(
            "/instance_config",
            get(get_instance_config).put(set_instance_config),
        )
        .route("/instance_config/yaml", get(get_instance_config_yaml))
        .route("/test_smtp", post(test_email))
        .route("/test_license_key", post(test_license_key))
        .route("/send_stats", post(send_stats))
        .route("/get_stats", get(get_stats))
        .route(
            "/latest_key_renewal_attempt",
            get(get_latest_key_renewal_attempt),
        )
        .route("/renew_license_key", post(renew_license_key))
        .route("/offline_license_status", get(get_offline_license_status))
        .route("/instance_hash", get(get_instance_hash))
        .route("/customer_portal", post(create_customer_portal_session))
        .route("/test_critical_channels", post(test_critical_channels))
        .route("/critical_alerts", get(get_critical_alerts))
        .route(
            "/critical_alerts/{id}/acknowledge",
            post(acknowledge_critical_alert),
        )
        .route(
            "/list_custom_instance_pg_databases",
            post(list_custom_instance_pg_databases),
        )
        .route(
            "/datatable_roles",
            get(datatable_roles_oss::list_datatable_roles)
                .post(datatable_roles_oss::create_datatable_role),
        )
        .route(
            "/datatable_roles/{id}",
            post(datatable_roles_oss::update_datatable_role)
                .delete(datatable_roles_oss::delete_datatable_role),
        )
        .route(
            "/refresh_custom_instance_user_pwd",
            post(refresh_custom_instance_user_pwd),
        )
        .route(
            "/external_instance_pg/status",
            get(get_external_instance_pg_status),
        )
        .route(
            "/external_instance_pg/setup",
            post(setup_external_instance_pg),
        )
        .route(
            "/external_instance_pg/databases",
            get(list_external_instance_pg_databases),
        )
        .route(
            "/external_instance_pg/databases/{name}",
            post(create_external_instance_pg_database).delete(drop_external_instance_pg_database),
        )
        .route(
            "/setup_custom_instance_pg_database/{name}",
            post(setup_custom_instance_pg_database),
        )
        .route(
            "/drop_custom_instance_pg_database/{name}",
            post(drop_custom_instance_pg_database),
        )
        .route(
            "/critical_alerts/acknowledge_all",
            post(acknowledge_all_critical_alerts),
        )
        .route(
            "/sync_cached_resource_types",
            post(sync_cached_resource_types),
        )
        .route(
            "/restart_worker_group/{worker_group}",
            post(restart_worker_group),
        );

    // Vault/Azure KV integration routes (EE only - requires both private and enterprise features)
    #[cfg(all(feature = "private", feature = "enterprise"))]
    let r = r
        .route("/test_secret_backend", post(test_secret_backend))
        .route("/migrate_secrets_to_vault", post(migrate_secrets_to_vault))
        .route(
            "/migrate_secrets_to_database",
            post(migrate_secrets_to_database),
        )
        .route("/test_azure_kv_backend", post(test_azure_kv_backend))
        .route(
            "/migrate_secrets_to_azure_kv",
            post(migrate_secrets_to_azure_kv),
        )
        .route(
            "/migrate_secrets_from_azure_kv",
            post(migrate_secrets_from_azure_kv),
        )
        .route("/test_aws_sm_backend", post(test_aws_sm_backend))
        .route(
            "/migrate_secrets_to_aws_sm",
            post(migrate_secrets_to_aws_sm),
        )
        .route(
            "/migrate_secrets_from_aws_sm",
            post(migrate_secrets_from_aws_sm),
        );

    #[cfg(feature = "parquet")]
    {
        return r
            .route("/test_object_storage_config", post(test_s3_bucket))
            .route(
                "/object_storage_usage",
                get(get_object_storage_usage).post(compute_object_storage_usage),
            )
            .route("/run_log_cleanup", post(run_log_cleanup))
            .route("/log_cleanup_status", get(log_cleanup_status))
            .route("/audit_logs_s3_status", get(audit_logs_s3_status))
            .route("/audit_logs_s3_backfill", post(run_audit_logs_s3_backfill))
            .route(
                "/audit_logs_s3_backfill_status",
                get(audit_logs_s3_backfill_status),
            );
    }

    #[cfg(not(feature = "parquet"))]
    {
        return r;
    }
}

#[derive(Deserialize)]
pub struct TestEmail {
    pub to: String,
    pub smtp: Smtp,
}

pub async fn test_email(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(test_email): Json<TestEmail>,
) -> error::Result<String> {
    require_super_admin(&db, &authed).await?;
    if !SMTP_ENABLED {
        return Err(error::Error::Generic(
            axum::http::StatusCode::NOT_IMPLEMENTED,
            "This Windmill build was compiled without SMTP support, so no email can be sent."
                .to_string(),
        ));
    }
    let smtp = test_email.smtp;
    let to = test_email.to;

    // A connection attempt covers TCP, the TLS handshake, EHLO and authentication against a remote
    // provider; a tighter budget times out before the server ever states why it refused.
    let client_timeout = Duration::from_secs(20);
    send_email_plain_text(
        "Test email from Windmill",
        "Test email content",
        vec![to],
        smtp,
        Some(client_timeout),
    )
    .await
    // The SMTP layer already phrases its failures for an instance admin; the anyhow wrapper it
    // comes back in would bury that behind "Internal: ... @<source location>".
    .map_err(|e| match e {
        error::Error::Anyhow { error, .. } => {
            error::Error::Generic(axum::http::StatusCode::BAD_REQUEST, format!("{error:#}"))
        }
        e => e,
    })?;

    Ok("Sent test email".to_string())
}

#[cfg(feature = "parquet")]
use windmill_object_store::ObjectSettings;

#[cfg(feature = "parquet")]
use windmill_object_store::{
    build_object_store_from_settings, build_public_object_store_from_settings,
};

#[cfg(feature = "parquet")]
pub async fn test_s3_bucket(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Json(test_s3_bucket): Json<ObjectSettings>,
) -> error::Result<String> {
    use bytes::Bytes;
    use futures::StreamExt;

    // The probe executes on the API server itself and reflects the upstream response into the
    // error, so any authenticated caller could otherwise use it as an SSRF / port-scan primitive
    // against the server's network, exfiltrate its ambient credentials, or write to its local
    // disk (see validate_object_storage_test). That holds on self-hosted instances as much as on
    // Cloud, so only super admins get the unrestricted path.
    let is_super_admin = windmill_api_auth::is_super_admin_authed(&db, &authed).await?;
    let restrict = !is_super_admin;
    if restrict {
        validate_object_storage_test(&test_s3_bucket)
            .await
            .map_err(|e| match e {
                // A job token never counts as a super admin (it is capped at workspace admin), so
                // a super admin calling this route from a script is told why rather than that
                // they lack a privilege they hold.
                error::Error::NotAuthorized(msg) if authed.job_id.is_some() => {
                    error::Error::NotAuthorized(format!(
                        "{msg} A job token ($WM_TOKEN) is never treated as a super admin; call \
                         this route with a user token instead."
                    ))
                }
                e => e,
            })?;
    }

    // The restricted client re-judges every address it connects to: the checks above resolve the
    // endpoint separately from the connect, which a rebinding name answers differently.
    let client = if restrict {
        build_public_object_store_from_settings(test_s3_bucket).await?
    } else {
        build_object_store_from_settings(test_s3_bucket, Some(&db))
            .await?
            .store
    };

    let run = async {
        let mut list = client.list(Some(
            &windmill_object_store::object_store_reexports::Path::from("".to_string()),
        ));
        match list.next().await {
            Some(Err(e)) => {
                tracing::error!("error listing bucket: {e:#}");
                return Err(error::Error::internal_err(format!(
                    "Failed to list files in blob storage: {e:#}"
                )));
            }
            Some(Ok(first_file)) => tracing::info!("Listed files: {:?}", first_file),
            None => tracing::info!("No files in blob storage"),
        }

        let path = windmill_object_store::object_store_reexports::Path::from(format!(
            "/test-s3-bucket-{uuid}",
            uuid = uuid::Uuid::new_v4()
        ));
        tracing::info!("Testing blob storage at path: {path}");
        client
            .put(
                &path,
                windmill_object_store::object_store_reexports::PutPayload::from_static(b"hello"),
            )
            .await
            .map_err(|e| anyhow::anyhow!("error writing file to {path}: {e:#}"))?;
        let content = client
            .get(&path)
            .await
            .map_err(to_anyhow)?
            .bytes()
            .await
            .map_err(to_anyhow)?;
        if content != Bytes::from_static(b"hello") {
            return Err(error::Error::internal_err(
                "Failed to read back from blob storage".to_string(),
            ));
        }
        client.delete(&path).await.map_err(to_anyhow)?;
        Ok::<String, error::Error>("Tested blob storage successfully".to_string())
    };

    if restrict {
        // The object-store client is built with timeouts disabled, so a malicious endpoint could
        // otherwise hold the API server connection open indefinitely.
        tokio::time::timeout(Duration::from_secs(15), run)
            .await
            .map_err(|_| {
                error::Error::internal_err("Object storage connectivity test timed out".to_string())
            })?
    } else {
        run.await
    }
}

// Hardening for the object-storage connectivity test by an untrusted (non-super-admin) caller.
// The probe runs on the API server, so without these constraints an authenticated
// user could coerce the server into connecting to arbitrary internal endpoints (SSRF), signing
// requests with the instance role (credential exfiltration), or reading/writing the server's local
// disk (filesystem object store).
#[cfg(feature = "parquet")]
async fn validate_object_storage_test(settings: &ObjectSettings) -> error::Result<()> {
    fn non_empty(opt: &Option<String>) -> bool {
        opt.as_ref().is_some_and(|s| !s.is_empty())
    }

    // Every refusal names the way out: the resource usually works in jobs (workers reach the
    // endpoint directly), so without it the refusal reads as a broken resource.
    const ALTERNATIVE: &str =
        "Ask a super admin to run it, or test the resource from a script, which runs on a worker.";

    // Reject backends that rely on the server's identity or local filesystem, require explicit
    // credentials for the rest (so the server never falls back to its own ambient credentials), and
    // resolve the host the client will actually connect to. We derive the *effective* endpoint here
    // — mirroring build_*_from_settings: the region/account-derived default and the virtual-hosted
    // bucket prefix — rather than only validating a caller-supplied `endpoint`, so caller-controlled
    // `region`/`account_name`/`bucket` cannot smuggle an internal host past the check (e.g. an empty
    // endpoint with region = "@169.254.169.254/" otherwise resolves to the cloud metadata service).
    let effective_endpoint: Option<String> = match settings {
        ObjectSettings::Filesystem(_) => {
            return Err(error::Error::NotAuthorized(
                "Testing a local filesystem object store requires a super admin: it runs on the \
                 Windmill server and reads and writes the server's local disk. Ask a super admin \
                 to run it."
                    .to_string(),
            ));
        }
        ObjectSettings::AwsOidc(_) => {
            return Err(error::Error::NotAuthorized(format!(
                "Testing OIDC-based object storage requires a super admin: it runs on the \
                 Windmill server with the server's own identity. {ALTERNATIVE}"
            )));
        }
        ObjectSettings::S3(s3) => {
            if !(non_empty(&s3.access_key) && non_empty(&s3.secret_key)) {
                return Err(error::Error::NotAuthorized(format!(
                    "Testing S3 storage without an explicit access key and secret key requires a \
                     super admin: it runs on the Windmill server, which would use its own ambient \
                     credentials. {ALTERNATIVE}"
                )));
            }
            let region = s3
                .region
                .clone()
                .filter(|r| !r.is_empty())
                .or_else(|| std::env::var("AWS_REGION").ok().filter(|r| !r.is_empty()))
                .unwrap_or_else(|| "us-east-1".to_string());
            let raw_endpoint = s3
                .endpoint
                .clone()
                .filter(|e| !e.is_empty())
                .or_else(|| std::env::var("S3_ENDPOINT").ok().filter(|e| !e.is_empty()))
                .unwrap_or_else(|| format!("s3.{region}.amazonaws.com"));
            Some(windmill_object_store::render_endpoint(
                raw_endpoint,
                !s3.allow_http.unwrap_or(true),
                s3.port,
                s3.path_style,
                s3.bucket.clone().unwrap_or_default(),
            ))
        }
        ObjectSettings::Azure(azure) => {
            if !non_empty(&azure.access_key) {
                return Err(error::Error::NotAuthorized(format!(
                    "Testing Azure storage without an explicit access key requires a super admin: \
                     it runs on the Windmill server, which would use its own ambient credentials. \
                     {ALTERNATIVE}"
                )));
            }
            Some(
                azure
                    .endpoint
                    .clone()
                    .filter(|e| !e.is_empty())
                    .unwrap_or_else(|| format!("{}.blob.core.windows.net", azure.account_name)),
            )
        }
        ObjectSettings::Gcs(gcs) => {
            // Mirror `build_gcs_client`'s blank-key check (shared predicate): a blank/`{}` key falls
            // back to the instance's ambient credentials there, so it must be rejected here too —
            // otherwise an untrusted caller could probe with the server's identity (the very
            // SSRF/credential-exfil this function guards against).
            if windmill_object_store::gcs_service_account_key_is_blank(&gcs.service_account_key) {
                return Err(error::Error::NotAuthorized(format!(
                    "Testing GCS storage without a service account key requires a super admin: \
                     it runs on the Windmill server, which would use its own ambient credentials. \
                     {ALTERNATIVE}"
                )));
            }
            // The service-account-key JSON can override the data-plane URL (`gcs_base_url`) and the
            // OAuth token endpoint (`token_uri`); the GCS client connects to whatever they point at.
            // Validate every http(s) URL embedded in the key. When none override it, the host stays
            // the public storage.googleapis.com, so no further check is needed.
            if let Ok(serde_json::Value::Object(map)) =
                serde_json::from_str::<serde_json::Value>(&gcs.service_account_key)
            {
                for value in map.values() {
                    if let Some(url) = value.as_str() {
                        // Match how the URL parser reads the value: leading whitespace/control is
                        // ignored and the scheme is case-insensitive.
                        let url =
                            url.trim_start_matches(|c: char| c.is_whitespace() || c.is_control());
                        if strip_http_scheme(url).is_some() {
                            validate_public_endpoint(url).await?;
                        }
                    }
                }
            }
            None
        }
    };

    // Block non-public network targets (internal services, cloud metadata, loopback, ...).
    if let Some(endpoint) = effective_endpoint {
        validate_public_endpoint(&endpoint).await?;
    }
    Ok(())
}

#[cfg(feature = "parquet")]
async fn validate_public_endpoint(endpoint: &str) -> error::Result<()> {
    let host = extract_host(endpoint).ok_or_else(|| {
        error::Error::BadRequest(format!("Invalid object storage endpoint: {endpoint}"))
    })?;

    let addrs: Vec<std::net::SocketAddr> = tokio::net::lookup_host((host.as_str(), 443u16))
        .await
        .map_err(|e| {
            error::Error::BadRequest(format!(
                "Could not resolve object storage endpoint '{host}': {e}"
            ))
        })?
        .collect();

    if addrs.is_empty() {
        return Err(error::Error::BadRequest(format!(
            "Could not resolve object storage endpoint '{host}'"
        )));
    }

    // Reject if any resolved address is non-public, which also defeats the simplest DNS-rebinding
    // attempts (a name resolving to both a public and a private address).
    for addr in addrs {
        if windmill_common::ssrf::is_private_ip(&addr.ip()) {
            // The resolved address stays out of the message: it is the server's resolver's
            // answer, and this message is only ever shown to the caller being constrained.
            return Err(error::Error::NotAuthorized(format!(
                "Testing object storage at '{host}', which resolves to a private, loopback, or \
                 link-local address, requires a super admin: this test runs on the Windmill \
                 server, which is not allowed to probe internal addresses for non-super-admins. \
                 Ask a super admin to run it, or test the resource from a script, which runs on \
                 a worker."
            )));
        }
    }
    Ok(())
}

// Strip a leading `http://`/`https://` scheme case-insensitively (URL schemes are
// case-insensitive), returning the remainder when one was present.
#[cfg(feature = "parquet")]
fn strip_http_scheme(s: &str) -> Option<&str> {
    for scheme in ["https://", "http://"] {
        let b = scheme.as_bytes();
        if s.len() >= b.len() && s.as_bytes()[..b.len()].eq_ignore_ascii_case(b) {
            return Some(&s[b.len()..]);
        }
    }
    None
}

#[cfg(feature = "parquet")]
fn extract_host(endpoint: &str) -> Option<String> {
    let mut s = endpoint.trim();
    if let Some(rest) = strip_http_scheme(s) {
        s = rest;
    }
    s = s.split(['/', '?', '#', '\\']).next().unwrap_or(s);
    if let Some((_, rest)) = s.rsplit_once('@') {
        s = rest;
    }
    let host = if let Some(rest) = s.strip_prefix('[') {
        // IPv6 literal, e.g. [::1]:9000
        rest.split(']').next().unwrap_or(rest)
    } else {
        // host or host:port
        s.split(':').next().unwrap_or(s)
    }
    .trim();
    if host.is_empty() {
        None
    } else {
        Some(host.to_string())
    }
}

#[cfg(feature = "parquet")]
async fn get_object_storage_usage(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<Option<storage_usage::StorageUsageProgress>> {
    require_super_admin(&db, &authed).await?;
    Ok(Json(storage_usage::get_status(&db).await?))
}

#[cfg(feature = "parquet")]
async fn compute_object_storage_usage(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::Result<axum::http::StatusCode> {
    require_super_admin(&db, &authed).await?;
    storage_usage::try_start(&db).await?;
    storage_usage::spawn_compute(db.clone());
    Ok(axum::http::StatusCode::ACCEPTED)
}

#[cfg(feature = "parquet")]
async fn run_log_cleanup(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::Result<axum::http::StatusCode> {
    require_super_admin(&db, &authed).await?;
    log_cleanup::try_start(&db).await?;
    log_cleanup::spawn_cleanup(db.clone());
    Ok(axum::http::StatusCode::ACCEPTED)
}

#[cfg(feature = "parquet")]
async fn log_cleanup_status(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<Option<log_cleanup::LogCleanupProgress>> {
    require_super_admin(&db, &authed).await?;
    Ok(Json(log_cleanup::get_status(&db).await?))
}

#[cfg(feature = "parquet")]
async fn audit_logs_s3_status(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<Option<audit_logs_s3::AuditLogsS3ExportStatus>> {
    require_super_admin(&db, &authed).await?;
    Ok(Json(audit_logs_s3::get_status(&db).await?))
}

#[cfg(feature = "parquet")]
async fn run_audit_logs_s3_backfill(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(req): Json<audit_logs_s3_backfill::BackfillRequest>,
) -> error::Result<axum::http::StatusCode> {
    require_super_admin(&db, &authed).await?;
    if !matches!(get_license_plan().await, LicensePlan::Enterprise) {
        return Err(error::Error::BadRequest(
            "Audit log export to object storage is an Enterprise feature".to_string(),
        ));
    }
    audit_logs_s3_backfill::try_start(&db, req.from, req.to).await?;
    audit_logs_s3_backfill::spawn_backfill(db.clone(), req.from, req.to);
    Ok(axum::http::StatusCode::ACCEPTED)
}

#[cfg(feature = "parquet")]
async fn audit_logs_s3_backfill_status(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<Option<audit_logs_s3_backfill::AuditBackfillProgress>> {
    require_super_admin(&db, &authed).await?;
    Ok(Json(audit_logs_s3_backfill::get_status(&db).await?))
}

#[derive(Deserialize)]
pub struct TestKey {
    pub license_key: String,
}

pub async fn test_license_key(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(TestKey { license_key }): Json<TestKey>,
) -> error::Result<String> {
    require_super_admin(&db, &authed).await?;
    let (_, expired, _offline_meta) = validate_license_key(license_key, Some(&db)).await?;

    if expired {
        Err(error::Error::BadRequest("Expired license key".to_string()))
    } else {
        Ok("Valid license key".to_string())
    }
}

#[derive(serde::Serialize)]
pub struct InstanceHash {
    pub instance_hash: Option<String>,
}

/// Returns the live cap status for an offline license, or `null` when no
/// offline license is loaded. Used by the superadmin settings panel.
pub async fn get_offline_license_status(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<Option<windmill_common::ee_oss::OfflineCapStatus>> {
    require_super_admin(&db, &authed).await?;

    let offline = (**windmill_common::ee_oss::LICENSE_OFFLINE_METADATA.load()).clone();
    let is_offline = matches!(&offline, Some(m) if m.is_offline());

    if !is_offline {
        return Ok(Json(None));
    }

    #[cfg(feature = "enterprise")]
    let cap = windmill_common::ee_oss::enforce_offline_caps(&db)
        .await
        .map_err(|e| error::Error::internal_err(format!("enforce_offline_caps: {e:#}")))?;
    #[cfg(not(feature = "enterprise"))]
    let cap: Option<windmill_common::ee_oss::OfflineCapStatus> = None;

    Ok(Json(cap))
}

/// Returns the per-instance binding hash that goes into offline license keys.
/// Admin invokes via `curl` with their personal token when requesting a key
/// from support.
pub async fn get_instance_hash(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<InstanceHash> {
    require_super_admin(&db, &authed).await?;
    #[cfg(feature = "enterprise")]
    let hash = windmill_common::ee_oss::compute_instance_hash(&db)
        .await
        .map_err(|e| error::Error::internal_err(format!("compute_instance_hash: {e:#}")))?;
    #[cfg(not(feature = "enterprise"))]
    let hash: Option<String> = None;
    Ok(Json(InstanceHash { instance_hash: hash }))
}

pub async fn get_local_settings(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<serde_json::Value> {
    require_super_admin(&db, &authed).await?;

    let mut settings = serde_json::Map::new();
    for key in ENV_SETTINGS.iter() {
        if let Some(value) = std::env::var(key).ok() {
            settings.insert(key.to_string(), serde_json::Value::String(value));
        }
    }
    Ok(Json(serde_json::Value::Object(settings)))
}

#[derive(serde::Deserialize)]
pub struct Value {
    pub value: Option<serde_json::Value>,
}

pub async fn delete_global_setting(db: &DB, key: &str) -> error::Result<()> {
    // ducklake_user_pg_pwd and ducklake_settings were old names stored as standalone global settings.
    // Leave them for backward compatibility (CLI will try to delete them if not present in the yaml)
    if key == "ducklake_user_pg_pwd"
        || key == "ducklake_settings"
        || key == "custom_instance_pg_databases"
    {
        tracing::error!("Tried to unset global setting {}, ignored", key);
        return Ok(());
    }
    sqlx::query!("DELETE FROM global_settings WHERE name = $1", key,)
        .execute(db)
        .await?;
    tracing::info!("Unset global setting {}", key);
    Ok(())
}
/// Returns true when `key` is one of the workspace-fairness settings whose
/// writes must be gated to cloud only.
fn is_workspace_fairness_setting(key: &str) -> bool {
    matches!(
        key,
        WORKSPACE_FAIRNESS_ENABLED_SETTING
            | WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING
            | WORKSPACE_FAIRNESS_DURATION_SECS_SETTING
            | WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING
    )
}

/// Enterprise gate for the workspace-fairness settings. Workspace fairness is
/// only useful on multi-tenant clusters where one workspace can starve other
/// workspaces sharing the same worker pool, and the feature is licensed as
/// part of Enterprise. Non-EE installs are rejected at write time; the runtime
/// dispatch additionally honours the `WORKSPACE_FAIRNESS_ENABLED` toggle.
async fn workspace_fairness_settings_allowed() -> bool {
    matches!(get_license_plan().await, LicensePlan::Enterprise)
}

pub async fn set_global_setting(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Path(key): Path<String>,
    Json(value): Json<Value>,
) -> error::Result<()> {
    require_super_admin(&db, &authed).await?;
    set_global_setting_internal(&db, key, value.value.unwrap_or(serde_json::Value::Null)).await
}

pub async fn set_global_setting_internal(
    db: &DB,
    key: String,
    value: serde_json::Value,
) -> error::Result<()> {
    let should_bump_instance_ai_revision = key == AI_CONFIG_SETTING;
    let value = if key == "retention_period_secs" {
        instance_config::clamp_retention_period(value)
    } else {
        value
    };

    // EE gate for workspace-fairness settings. Workspace fairness only matters
    // on multi-tenant clusters; it is licensed as an Enterprise feature so the
    // setter rejects writes from non-EE builds. Disabling/clearing writes are
    // *always* allowed regardless of license, so an admin who downgrades from
    // EE (or imports a row from a cloned EE DB) can always turn the cap off:
    //   - `Null` / empty-string  → row delete
    //   - `Bool(false)` on `workspace_fairness_enabled` → explicit disable
    // Without the `Bool(false)` carve-out, a stale `enabled=true` row from a
    // downgrade would be impossible to flip off through the normal API/UI
    // and the runtime path (which only checks the toggle) would keep
    // throttling.
    let is_clearing_value = matches!(&value, serde_json::Value::Null)
        || matches!(&value, serde_json::Value::String(s) if s.trim().is_empty())
        || (key == WORKSPACE_FAIRNESS_ENABLED_SETTING
            && matches!(&value, serde_json::Value::Bool(false)));
    if is_workspace_fairness_setting(&key)
        && !is_clearing_value
        && !workspace_fairness_settings_allowed().await
    {
        return Err(error::Error::BadRequest(format!(
            "{} requires an Enterprise license",
            key
        )));
    }

    if key == EXTERNAL_INSTANCE_PG_SETTING {
        return windmill_common::external_instance_pg::write_external_instance_pg_setting(
            db,
            Some(&value),
        )
        .await;
    }

    run_setting_pre_write_hook(db, &key, &value).await?;

    match value {
        serde_json::Value::Null => {
            if instance_config::PROTECTED_SETTINGS.contains(&key.as_str()) {
                return Err(error::Error::BadRequest(format!(
                    "{key} is a protected setting and cannot be deleted"
                )));
            }
            delete_global_setting(db, &key).await?;
        }
        serde_json::Value::String(ref x) if x.trim().is_empty() => {
            if instance_config::PROTECTED_SETTINGS.contains(&key.as_str()) {
                return Err(error::Error::BadRequest(format!(
                    "{key} is a protected setting and cannot be set to empty"
                )));
            }
            delete_global_setting(db, &key).await?;
        }
        v => {
            sqlx::query!(
                 "INSERT INTO global_settings (name, value) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value, updated_at = now()",
                 key,
                 v
             )
             .execute(db)
             .await?;
            tracing::info!(
                "Set global setting {} to {}",
                key,
                instance_config::format_setting_value(&key, &v)
            );
        }
    };

    if should_bump_instance_ai_revision {
        bump_instance_ai_config_revision();
    }

    // Tag reads are served from an in-memory cache that this process otherwise only
    // refreshes on the next global-settings poll, so without this a refetch right after
    // the write still returns the pre-write list. The setting is already persisted at
    // this point, so a failed refresh must not be reported as a failed write — the
    // poller retries it.
    if key == CUSTOM_TAGS_SETTING {
        if let Err(e) = reload_custom_tags_setting(db).await {
            tracing::error!(error = %e, "Could not reload custom tags setting after write");
        }
    }

    Ok(())
}

/// Run side-effect hooks for specific settings before writing to DB.
/// Extracted from `set_global_setting_internal` for reuse by the bulk endpoint.
async fn run_setting_pre_write_hook(
    db: &DB,
    key: &str,
    value: &serde_json::Value,
) -> error::Result<()> {
    match key {
        // The instance AI config is written as an untyped blob through this generic
        // endpoint, so it never passes the typed check the workspace handler applies.
        // Rates that reach a cost total unbounded would make it negative or infinite.
        AI_CONFIG_SETTING => {
            windmill_ai::ai_types::validate_model_pricing_json(value)
                .map_err(error::Error::BadRequest)?;
            windmill_ai::ai_types::validate_token_maps_json(value)
                .map_err(error::Error::BadRequest)?;
        }
        AUTOMATE_USERNAME_CREATION_SETTING => {
            if value.as_bool().unwrap_or(false) {
                generate_instance_username_for_all_users(db)
                    .await
                    .map_err(|err| {
                        error::Error::internal_err(format!(
                            "Failed to generate instance wide usernames: {}",
                            err
                        ))
                    })?;
            } else {
                // Disabling is only allowed before any instance-wide username has been
                // assigned. Once usernames exist they are globally unique and are baked
                // into stored `u/<username>` identities (schedules, triggers, drafts,
                // and non-member superadmin ownership). Disabling would drop back to
                // workspace-local username uniqueness, letting a member reuse an
                // existing instance username and silently take over those identities —
                // so the setting is effectively one-way once derivation has taken
                // effect. Re-saving `false` on an already-disabled instance is a no-op
                // and stays allowed (guarded by the current-value check).
                let currently_enabled = sqlx::query_scalar!(
                    "SELECT value FROM global_settings WHERE name = $1",
                    AUTOMATE_USERNAME_CREATION_SETTING
                )
                .fetch_optional(db)
                .await?
                .and_then(|v| v.as_bool())
                .unwrap_or(true);
                if currently_enabled {
                    let usernames_exist = sqlx::query_scalar!(
                        "SELECT EXISTS(SELECT 1 FROM password WHERE username IS NOT NULL)"
                    )
                    .fetch_one(db)
                    .await?
                    .unwrap_or(false);
                    if usernames_exist {
                        return Err(error::Error::BadRequest(
                            "automate_username_creation cannot be disabled once instance-wide usernames have been assigned: existing u/<username> identities (schedules, triggers, drafts, superadmin ownership) rely on those usernames staying stable and globally unique.".to_string(),
                        ));
                    }
                }
            }
        }
        CRITICAL_ALERT_MUTE_UI_SETTING => {
            if value.as_bool().unwrap_or(false) {
                sqlx::query!("UPDATE alerts SET acknowledged = true")
                    .execute(db)
                    .await?;
            }
        }
        APP_WORKSPACED_ROUTE_SETTING => {
            let serde_json::Value::Bool(workspaced_route) = value else {
                return Err(error::Error::BadRequest(format!(
                    "{} setting Expected to be boolean",
                    APP_WORKSPACED_ROUTE_SETTING
                )));
            };

            // Cloud always scopes app custom paths by workspace_id (see
            // `custom_path_exists` in apps.rs), so duplicates across workspaces
            // are expected and this setting has no runtime effect on cloud.
            if !*workspaced_route && !*CLOUD_HOSTED {
                #[derive(Debug, Deserialize, Serialize)]
                #[allow(unused)]
                struct DuplicateApp {
                    custom_path: Option<String>,
                    path: String,
                }
                let duplicate_app = sqlx::query_as!(
                    DuplicateApp,
                    r#"
                        SELECT
                            path,
                            custom_path
                        FROM
                            app
                        WHERE
                            custom_path IN (
                                SELECT
                                    custom_path
                                FROM
                                    app
                                GROUP
                                    BY custom_path
                                HAVING COUNT(*) > 1
                            )
                        ORDER BY custom_path
                        "#
                )
                .fetch_all(db)
                .await?;

                if !duplicate_app.is_empty() {
                    tracing::error!(
                        "Cannot disable {} setting as duplicate app with custom path were found: {:?}",
                        APP_WORKSPACED_ROUTE_SETTING,
                        &duplicate_app
                    );

                    #[derive(Serialize)]
                    struct ErrorResponse {
                        error: String,
                        details: Vec<DuplicateApp>,
                    }

                    let error_response = ErrorResponse {
                        error: "Duplicate custom paths detected".to_string(),
                        details: duplicate_app,
                    };

                    return Err(error::Error::JsonErr(
                        serde_json::to_value(error_response).unwrap(),
                    ));
                }
            }
        }
        HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING => {
            // Rejected at write time rather than at boot: a mistyped origin
            // matches no request, so it would silently block the very app it
            // names with nothing but a log line to go on.
            windmill_common::global_settings::parse_allowed_origins_setting(Some(value))?;
        }
        HTTP_ROUTE_WORKSPACED_ROUTE_SETTING => {
            let serde_json::Value::Bool(workspaced_route) = value else {
                return Err(error::Error::BadRequest(format!(
                    "{} setting expected to be boolean",
                    HTTP_ROUTE_WORKSPACED_ROUTE_SETTING
                )));
            };

            // Cloud always scopes routes by workspace_id (see
            // `route_path_key_exists` in windmill-trigger-http), so duplicates
            // across workspaces are expected and this setting has no runtime
            // effect on cloud.
            if !*workspaced_route && !*CLOUD_HOSTED {
                #[derive(Debug, Deserialize, Serialize)]
                #[allow(unused)]
                struct DuplicateRoute {
                    route_path: String,
                    workspace_id: String,
                    http_method: String,
                }
                let duplicate_routes = sqlx::query_as!(
                    DuplicateRoute,
                    r#"
                        SELECT
                            route_path,
                            workspace_id,
                            http_method::TEXT AS "http_method!"
                        FROM
                            http_trigger
                        WHERE
                            workspaced_route IS FALSE
                            AND route_path_key IN (
                                SELECT
                                    route_path_key
                                FROM
                                    http_trigger
                                WHERE
                                    workspaced_route IS FALSE
                                GROUP BY
                                    route_path_key, http_method
                                HAVING COUNT(*) > 1
                            )
                        ORDER BY route_path_key
                    "#
                )
                .fetch_all(db)
                .await?;

                if !duplicate_routes.is_empty() {
                    tracing::error!(
                        "Cannot disable {} setting as duplicate http routes were found: {:?}",
                        HTTP_ROUTE_WORKSPACED_ROUTE_SETTING,
                        &duplicate_routes
                    );

                    #[derive(Serialize)]
                    struct ErrorResponse {
                        error: String,
                        details: Vec<DuplicateRoute>,
                    }

                    let error_response = ErrorResponse {
                        error: "Duplicate HTTP route paths detected".to_string(),
                        details: duplicate_routes,
                    };

                    return Err(error::Error::JsonErr(
                        serde_json::to_value(error_response).unwrap(),
                    ));
                }
            }
        }
        RETENTION_PERIOD_SECS_OVERRIDES_SETTING => {
            // Reject a malformed map at write time so it can never be persisted. A persisted bad
            // value (negative or non-integer) would fail to parse on the next server start and,
            // because the loader fails closed (skips cleanup until a known-good value is read),
            // silently disable ALL job-retention cleanup indefinitely. This shape check must stay in
            // sync with `parse_retention_overrides` in backend/src/monitor.rs.
            match value {
                // Clearing (delete row) is handled by the caller; allow it through.
                serde_json::Value::Null => {}
                serde_json::Value::String(s) if s.trim().is_empty() => {}
                serde_json::Value::Object(map) => {
                    if map.len() > MAX_RETENTION_OVERRIDE_WORKSPACES {
                        return Err(error::Error::BadRequest(format!(
                            "retention_period_secs_overrides: at most {MAX_RETENTION_OVERRIDE_WORKSPACES} per-workspace overrides are allowed, got {}",
                            map.len()
                        )));
                    }
                    for (ws, v) in map {
                        if !v.as_i64().is_some_and(|secs| secs >= 0) {
                            return Err(error::Error::BadRequest(format!(
                                "retention_period_secs_overrides: override for '{ws}' must be a non-negative integer number of seconds, got {v}"
                            )));
                        }
                    }
                }
                _ => {
                    return Err(error::Error::BadRequest(
                        "retention_period_secs_overrides must be a JSON object of {workspace_id: seconds}".to_string(),
                    ));
                }
            }
        }
        GITHUB_APP_WEBHOOK_BASE_URL_SETTING => {
            // A bad value here yields a webhook GitHub can never deliver to, and the
            // failure only shows up much later as "falling back to polling" on a
            // repository — so reject it at the boundary instead.
            match value {
                // Clearing (delete row) is handled by the caller; allow it through.
                serde_json::Value::Null => {}
                serde_json::Value::String(s) if s.trim().is_empty() => {}
                serde_json::Value::String(s) => {
                    windmill_common::global_settings::validate_webhook_base_url(s).map_err(
                        |e| {
                            error::Error::BadRequest(format!(
                                "{GITHUB_APP_WEBHOOK_BASE_URL_SETTING}: {e}"
                            ))
                        },
                    )?;
                }
                _ => {
                    return Err(error::Error::BadRequest(format!(
                        "{GITHUB_APP_WEBHOOK_BASE_URL_SETTING} must be a URL string"
                    )));
                }
            }
        }
        MAX_TOKEN_EXPIRATION_DAYS_SETTING => {
            windmill_common::global_settings::parse_max_token_expiration_days(Some(value))
                .map_err(|e| {
                    error::Error::BadRequest(format!("{MAX_TOKEN_EXPIRATION_DAYS_SETTING}: {e}"))
                })?;
        }
        INSTANCE_BANNER_SETTING => {
            match value {
                // Clearing (delete row) is handled by the caller; allow it through.
                serde_json::Value::Null => {}
                serde_json::Value::String(s) if s.trim().is_empty() => {}
                v => {
                    windmill_common::global_settings::validate_instance_banner(v).map_err(|e| {
                        error::Error::BadRequest(format!("{INSTANCE_BANNER_SETTING}: {e}"))
                    })?;
                }
            }
        }
        ACCENT_COLOR_SETTING => match value {
            serde_json::Value::Null => {}
            serde_json::Value::String(s) if s.trim().is_empty() => {}
            v => windmill_common::global_settings::validate_accent_color(v)
                .map_err(|e| error::Error::BadRequest(format!("{ACCENT_COLOR_SETTING}: {e}")))?,
        },
        _ => {}
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Bulk instance config endpoints
// ---------------------------------------------------------------------------

async fn get_instance_config(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> JsonResult<InstanceConfig> {
    require_super_admin(&db, &authed).await?;
    let config = InstanceConfig::from_db(&db)
        .await
        .map_err(|e| error::Error::internal_err(e.to_string()))?;
    Ok(Json(config))
}

async fn get_instance_config_yaml(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::Result<Response> {
    require_super_admin(&db, &authed).await?;
    let config = InstanceConfig::from_db(&db)
        .await
        .map_err(|e| error::Error::internal_err(e.to_string()))?;
    let yaml = config
        .to_sorted_yaml()
        .map_err(|e| error::Error::internal_err(e))?;
    Response::builder()
        .header("content-type", "application/yaml")
        .body(Body::from(yaml))
        .map_err(|e| error::Error::internal_err(e.to_string()))
}

async fn set_instance_config(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(desired): Json<InstanceConfig>,
) -> error::Result<()> {
    require_super_admin(&db, &authed).await?;

    let current = InstanceConfig::from_db(&db)
        .await
        .map_err(|e| error::Error::internal_err(e.to_string()))?;

    let desired_map = desired.global_settings.to_settings_map();
    if !desired_map.is_empty() {
        let current_map = current.global_settings.to_settings_map();
        let mut settings_diff =
            instance_config::diff_global_settings(&current_map, &desired_map, ApplyMode::Merge);
        let ai_config_changed = settings_diff
            .upserts
            .iter()
            .any(|(key, _)| key == AI_CONFIG_SETTING);

        // Mirror the per-key EE gate in `set_global_setting_internal`. Without
        // this, the bulk endpoint would let a non-EE superadmin persist
        // `workspace_fairness_*` rows even though the per-key API rejects them.
        // Only block *non-disabling* upserts; deletes are allowed everywhere
        // (already filtered into `settings_diff.removals`) and a
        // `workspace_fairness_enabled=false` upsert is treated as a disable,
        // so a downgraded instance can always turn the cap off via the bulk
        // YAML endpoint too.
        let upserts_touch_fairness_non_disable = settings_diff.upserts.iter().any(|(k, v)| {
            if !is_workspace_fairness_setting(k) {
                return false;
            }
            !(k == WORKSPACE_FAIRNESS_ENABLED_SETTING
                && matches!(v, serde_json::Value::Bool(false)))
        });
        if upserts_touch_fairness_non_disable && !workspace_fairness_settings_allowed().await {
            return Err(error::Error::BadRequest(
                "Workspace fairness settings require an Enterprise license".to_string(),
            ));
        }

        for (key, value) in &settings_diff.upserts {
            if key != EXTERNAL_INSTANCE_PG_SETTING {
                run_setting_pre_write_hook(&db, key, value).await?;
            }
        }
        windmill_common::external_instance_pg::write_external_instance_pg_from_diff(
            &db,
            &mut settings_diff,
        )
        .await?;

        instance_config::apply_settings_diff(&db, &settings_diff)
            .await
            .map_err(|e| error::Error::internal_err(e.to_string()))?;

        if ai_config_changed {
            bump_instance_ai_config_revision();
        }
    }

    if !desired.worker_configs.is_empty() {
        let current_wc: std::collections::BTreeMap<String, serde_json::Value> = current
            .worker_configs
            .iter()
            .map(|(k, v)| {
                (
                    k.clone(),
                    serde_json::to_value(v).expect("WorkerGroupConfig serialization cannot fail"),
                )
            })
            .collect();
        let desired_wc: std::collections::BTreeMap<String, serde_json::Value> = desired
            .worker_configs
            .iter()
            .map(|(k, v)| {
                (
                    k.clone(),
                    serde_json::to_value(v).expect("WorkerGroupConfig serialization cannot fail"),
                )
            })
            .collect();
        let configs_diff =
            instance_config::diff_worker_configs(&current_wc, &desired_wc, ApplyMode::Merge);
        instance_config::apply_configs_diff(&db, &configs_diff)
            .await
            .map_err(|e| error::Error::internal_err(e.to_string()))?;
    }

    Ok(())
}

pub async fn get_global_setting(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Path(key): Path<String>,
) -> JsonResult<serde_json::Value> {
    if !key.starts_with("default_error_handler_")
        && !key.starts_with("default_recovery_handler_")
        && !key.starts_with("default_success_handler_")
        && key != AUTOMATE_USERNAME_CREATION_SETTING
        && key != DEFAULT_TAGS_WORKSPACES_SETTING
        && key != HUB_BASE_URL_SETTING
        // `wmill hub pull` reads it from a job, and no job token clears the gate. It binds an
        // offline license only together with `license_key`, which stays gated.
        && key != UNIQUE_ID_SETTING
        && key != HUB_ACCESSIBLE_URL_SETTING
        && key != DISABLE_HUB_SETTING
        && key != EMAIL_DOMAIN_SETTING
        && key != APP_WORKSPACED_ROUTE_SETTING
        && key != HTTP_ROUTE_WORKSPACED_ROUTE_SETTING
        // The route editor shows the inherited default to whoever is editing a
        // trigger, who is usually not a superadmin. Not a secret either: any
        // browser discovers the list by reading Access-Control-Allow-Origin off
        // a response.
        && key != HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING
        && key != WS_BASE_URL_SETTING
        && key != INSTANCE_BANNER_SETTING
        // The token form reads it to stop offering expirations the server would shorten.
        && key != MAX_TOKEN_EXPIRATION_DAYS_SETTING
        // Whoever is wiring up an MCP client reads it to know whether a URL-borne token
        // would be refused, and they are usually not a superadmin. Not a secret: pointing
        // any MCP client at the instance discovers the same answer.
        && key != MCP_DISABLE_TOKEN_QUERY_PARAM_SETTING
    {
        require_super_admin(&db, &authed).await?;
    }
    let value = sqlx::query!("SELECT value FROM global_settings WHERE name = $1", key)
        .fetch_optional(&db)
        .await?
        .map(|x| x.value);

    Ok(Json(value.unwrap_or_else(|| serde_json::Value::Null)))
}

#[cfg(feature = "enterprise")]
#[derive(Deserialize, serde::Serialize)]
struct GlobalSetting {
    name: String,
    value: serde_json::Value,
}

/// Repositories whose registered webhook still points at a receiver the instance no
/// longer uses — what an admin has to re-save after changing the webhook base URL.
/// Read-only; changing the setting never moves a live hook on its own.
async fn github_app_stale_webhooks(
    Extension(_db): Extension<DB>,
    authed: ApiAuthed,
) -> JsonResult<serde_json::Value> {
    require_super_admin(&_db, &authed).await?;
    #[cfg(all(feature = "enterprise", feature = "private"))]
    {
        let stale = windmill_common::git_sync_ee::stale_webhook_repos(&_db).await?;
        return Ok(Json(serde_json::to_value(stale).map_err(|e| {
            error::Error::internal_err(format!("Failed to serialize stale webhooks: {e}"))
        })?));
    }
    #[cfg(not(all(feature = "enterprise", feature = "private")))]
    Ok(Json(serde_json::json!([])))
}

#[cfg(feature = "enterprise")]
async fn list_global_settings(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> JsonResult<Vec<GlobalSetting>> {
    require_super_admin(&db, &authed).await?;
    let settings = sqlx::query_as!(GlobalSetting, "SELECT name, value FROM global_settings")
        .fetch_all(&db)
        .await?;

    Ok(Json(settings))
}

#[cfg(not(feature = "enterprise"))]
async fn list_global_settings() -> JsonResult<String> {
    return Err(error::Error::BadRequest(
        "Listing global settings not available on community edition".to_string(),
    ));
}

pub async fn send_stats(Extension(db): Extension<DB>, authed: ApiAuthed) -> Result<String> {
    require_super_admin(&db, &authed).await?;
    windmill_common::stats_oss::send_stats(
        &HTTP_CLIENT,
        &db,
        windmill_common::stats_oss::SendStatsReason::Manual,
        false,
    )
    .await?;

    Ok("Sent stats".to_string())
}

async fn restart_worker_group(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Path(worker_group): Path<String>,
) -> error::Result<String> {
    require_devops_role(&db, &authed).await?;

    sqlx::query!(
        "INSERT INTO notify_event (channel, payload) VALUES ('restart_worker_group', $1)",
        worker_group
    )
    .execute(&db)
    .await?;

    Ok(format!(
        "Restart signal sent to worker group '{worker_group}'"
    ))
}

#[derive(serde::Serialize)]
pub struct StatsDownload {
    pub signature: String,
    pub data: String,
}

#[cfg(feature = "enterprise")]
pub async fn get_stats(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::JsonResult<StatsDownload> {
    require_super_admin(&db, &authed).await?;
    let stats = windmill_common::stats_oss::get_stats_payload(
        &db,
        &windmill_common::stats_oss::SendStatsReason::Manual,
        false,
    )
    .await?;
    let json =
        serde_json::to_string(&stats).map_err(|e| error::Error::InternalErr(e.to_string()))?;
    let signature = windmill_common::stats_oss::sign_stats(&json);
    Ok(axum::Json(StatsDownload { signature, data: json }))
}

#[cfg(not(feature = "enterprise"))]
pub async fn get_stats() -> error::JsonResult<StatsDownload> {
    Err(error::Error::BadRequest(
        "Downloading telemetry is only available on enterprise edition".to_string(),
    ))
}

#[derive(serde::Serialize)]
pub struct KeyRenewalAttempt {
    result: String,
    attempted_at: chrono::DateTime<chrono::Utc>,
}

pub async fn get_latest_key_renewal_attempt(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> JsonResult<Option<KeyRenewalAttempt>> {
    require_super_admin(&db, &authed).await?;

    let last_attempt = sqlx::query!(
        "SELECT value, created_at FROM metrics WHERE id = $1 ORDER BY created_at DESC LIMIT 1",
        "license_key_renewal"
    )
    .fetch_optional(&db)
    .await?;

    match last_attempt {
        Some(last_attempt) => {
            let last_attempt_result = serde_json::from_value::<String>(last_attempt.value)
                .map_err(|e| {
                    error::Error::internal_err(format!("Failed to parse last attempt: {}", e))
                })?;
            Ok(Json(Some(KeyRenewalAttempt {
                result: last_attempt_result,
                attempted_at: last_attempt.created_at,
            })))
        }
        None => Ok(Json(None)),
    }
}

#[cfg(feature = "enterprise")]
#[derive(Deserialize)]
pub struct LicenseQuery {
    license_key: Option<String>,
}

#[cfg(not(feature = "enterprise"))]
pub async fn renew_license_key() -> Result<String> {
    return Err(error::Error::BadRequest(
        "License key renewal not available on community edition".to_string(),
    ));
}

#[cfg(feature = "enterprise")]
pub async fn renew_license_key(
    Extension(db): Extension<DB>,
    Query(LicenseQuery { license_key }): Query<LicenseQuery>,
    authed: ApiAuthed,
) -> Result<String> {
    require_super_admin(&db, &authed).await?;
    let result = windmill_common::ee_oss::renew_license_key(
        &HTTP_CLIENT,
        &db,
        license_key,
        windmill_common::ee_oss::RenewReason::Manual,
    )
    .await;

    if result != "success" {
        return Err(error::Error::BadRequest(format!(
            "Failed to renew license key: {}",
            if result == "Unauthorized" {
                "Invalid key".to_string()
            } else {
                result
            }
        )));
    } else {
        return Ok("Renewed license key".to_string());
    }
}

#[cfg(not(feature = "enterprise"))]
pub async fn create_customer_portal_session() -> Result<String> {
    return Err(error::Error::BadRequest(
        "Customer portal is not available on community edition".to_string(),
    ));
}

#[cfg(feature = "enterprise")]
pub async fn create_customer_portal_session(
    Query(LicenseQuery { license_key }): Query<LicenseQuery>,
) -> Result<String> {
    let url =
        windmill_common::ee_oss::create_customer_portal_session(&HTTP_CLIENT, license_key).await?;

    return Ok(url);
}

#[cfg(feature = "enterprise")]
pub async fn test_critical_channels(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(test_critical_channels): Json<Vec<CriticalErrorChannel>>,
) -> Result<String> {
    require_super_admin(&db, &authed).await?;

    #[cfg(feature = "enterprise")]
    send_critical_alert(
        "Test critical error".to_string(),
        &db,
        CriticalAlertKind::CriticalError,
        Some(test_critical_channels),
    )
    .await;
    Ok("Sent test critical error".to_string())
}

#[cfg(not(feature = "enterprise"))]
pub async fn test_critical_channels() -> Result<String> {
    Ok("Critical channels require EE".to_string())
}

#[cfg(feature = "enterprise")]
pub async fn get_critical_alerts(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Query(params): Query<windmill_alerting::AlertQueryParams>,
) -> JsonResult<serde_json::Value> {
    require_devops_role(&db, &authed).await?;

    windmill_alerting::get_critical_alerts(db, params, None).await
}

#[cfg(not(feature = "enterprise"))]
pub async fn get_critical_alerts() -> error::Error {
    error::Error::NotFound("Critical Alerts require EE".to_string())
}

#[cfg(feature = "enterprise")]
pub async fn acknowledge_critical_alert(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Path(id): Path<i32>,
) -> error::Result<String> {
    require_devops_role(&db, &authed).await?;
    windmill_alerting::acknowledge_critical_alert(db, None, id).await
}

#[cfg(not(feature = "enterprise"))]
pub async fn acknowledge_critical_alert() -> error::Error {
    error::Error::NotFound("Critical Alerts require EE".to_string())
}

#[cfg(feature = "enterprise")]
pub async fn acknowledge_all_critical_alerts(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
) -> error::Result<String> {
    require_super_admin(&db, &authed).await?;

    windmill_alerting::acknowledge_all_critical_alerts(db, None).await
}

#[cfg(not(feature = "enterprise"))]
pub async fn acknowledge_all_critical_alerts() -> error::Error {
    error::Error::NotFound("Critical Alerts require EE".to_string())
}

#[derive(Deserialize, Debug, Serialize)]
struct CustomInstanceDb {
    logs: CustomInstanceDbLogs, // (Step, Message)[]
    success: bool,
    error: Option<String>,
    tag: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    used_by_workspaces: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    workspace_id: Option<String>,
}

#[derive(Deserialize, Debug, Serialize, Default)]
#[serde(default)]
struct CustomInstanceDbLogs {
    super_admin: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    database_credentials: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    valid_dbname: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    created_database: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    db_connect: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    grant_permissions: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    replication_user: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    replication_user_error: Option<String>,
    #[serde(skip_serializing_if = "String::is_empty")]
    user_connect: String,
}

async fn list_custom_instance_pg_databases(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
) -> JsonResult<HashMap<String, CustomInstanceDb>> {
    let result = sqlx::query_scalar!(
        r#"SELECT value->'databases' FROM global_settings WHERE name = 'custom_instance_pg_databases'"#,
    )
    .fetch_one(&db)
    .await?
    .ok_or_else(|| error::Error::ExecutionErr("Couldn't find custom_instance_pg_databases".to_string()))?;
    let mut result: HashMap<String, CustomInstanceDb> =
        serde_json::from_value(result).map_err(|e| {
            error::Error::ExecutionErr(format!(
                "couldn't parse custom_instance_pg_databases.databases : {}",
                e.to_string()
            ))
        })?;

    if !windmill_api_auth::is_super_admin_authed(&db, &authed).await? {
        // A fork copy's name gives away the workspace it was reserved for, so every pending fork on
        // the instance would be listed. Kept for members of that workspace, and wherever the
        // caller's workspaces use it, e.g. the fork it was finalized into.
        let reserved_visible: BTreeSet<String> = sqlx::query_scalar(
            r#"SELECT e.k FROM global_settings gs
               CROSS JOIN LATERAL jsonb_each(gs.value->'databases') AS e(k, v)
               WHERE gs.name = 'custom_instance_pg_databases' AND e.v->>'workspace_id' IS NOT NULL
                 AND (EXISTS (SELECT 1 FROM usr WHERE usr.email = $1
                                AND usr.workspace_id = e.v->>'workspace_id')
                   OR EXISTS (SELECT 1 FROM usr JOIN workspace_settings ws
                                ON ws.workspace_id = usr.workspace_id
                              CROSS JOIN LATERAL jsonb_each(
                                  CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object'
                                      THEN ws.datatable->'datatables' ELSE '{}'::jsonb END) dt
                              WHERE usr.email = $1
                                AND dt.value->'database'->>'resource_type' = 'instance'
                                AND dt.value->'database'->>'resource_path' = e.k))"#,
        )
        .bind(&authed.email)
        .fetch_all(&db)
        .await?
        .into_iter()
        .collect();
        result.retain(|dbname, entry| {
            entry.workspace_id.is_none() || reserved_visible.contains(dbname)
        });
        // Which workspace reserved a copy is still only for superadmins.
        for entry in result.values_mut() {
            entry.workspace_id = None;
        }
        return Ok(Json(result));
    }
    {
        // Enrich each database with the list of workspaces referencing it through
        // either a ducklake catalog or a datatable database whose resource_type is
        // 'instance'. Not stored in DB to avoid drift.
        let usages = sqlx::query!(
            r#"
            SELECT ws.workspace_id AS "workspace_id!", entry->'catalog'->>'resource_path' AS dbname
            FROM workspace_settings ws
            CROSS JOIN LATERAL jsonb_each(
                CASE WHEN jsonb_typeof(ws.ducklake->'ducklakes') = 'object'
                    THEN ws.ducklake->'ducklakes'
                    ELSE '{}'::jsonb END
            ) AS dl(k, entry)
            WHERE entry->'catalog'->>'resource_type' = 'instance'
            AND entry->'catalog'->>'resource_path' IS NOT NULL
            UNION ALL
            SELECT ws.workspace_id AS "workspace_id!", entry->'database'->>'resource_path' AS dbname
            FROM workspace_settings ws
            CROSS JOIN LATERAL jsonb_each(
                CASE WHEN jsonb_typeof(ws.datatable->'datatables') = 'object'
                    THEN ws.datatable->'datatables'
                    ELSE '{}'::jsonb END
            ) AS dt(k, entry)
            WHERE entry->'database'->>'resource_type' = 'instance'
            AND entry->'database'->>'resource_path' IS NOT NULL
            "#,
        )
        .fetch_all(&db)
        .await?;

        let mut by_db: HashMap<String, BTreeSet<String>> = HashMap::new();
        for row in usages {
            if let Some(dbname) = row.dbname {
                by_db.entry(dbname).or_default().insert(row.workspace_id);
            }
        }
        for (dbname, entry) in result.iter_mut() {
            if let Some(workspaces) = by_db.remove(dbname) {
                entry.used_by_workspaces = workspaces.into_iter().collect();
            }
        }
    }

    return Ok(Json(result));
}

async fn refresh_custom_instance_user_pwd(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
) -> JsonResult<()> {
    require_super_admin(&db, &authed).await?;
    windmill_common::utils::refresh_custom_instance_user_pwd(&db).await?;
    windmill_common::utils::refresh_custom_instance_replication_user_pwd(&db).await?;
    Ok(Json(()))
}

async fn get_external_instance_pg_status(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
) -> JsonResult<windmill_common::external_instance_pg::ExternalInstancePgStatus> {
    require_super_admin(&db, &authed).await?;
    Ok(Json(
        windmill_common::external_instance_pg::external_instance_pg_status(&db).await?,
    ))
}

#[derive(Deserialize)]
struct SetupExternalInstancePgBody {
    #[serde(default)]
    rotate_passwords: bool,
}

async fn setup_external_instance_pg(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Json(body): Json<SetupExternalInstancePgBody>,
) -> JsonResult<windmill_common::external_instance_pg::ExternalInstancePgSetupReport> {
    require_super_admin(&db, &authed).await?;
    let report = windmill_common::external_instance_pg::setup_external_instance_pg_unchecked(
        &db,
        body.rotate_passwords,
    )
    .await?;
    let rotated = body.rotate_passwords.to_string();
    let success = report.success.to_string();
    windmill_audit::audit_oss::audit_log(
        &db,
        &authed,
        "settings.setup_external_instance_pg",
        windmill_audit::ActionKind::Update,
        "global",
        Some(&authed.email),
        Some(
            [
                ("rotate_passwords", rotated.as_str()),
                ("success", success.as_str()),
            ]
            .into(),
        ),
    )
    .await?;
    Ok(Json(report))
}

#[derive(Serialize)]
struct ExternalInstancePgDatabase {
    #[serde(flatten)]
    status: windmill_common::instance_config::CustomInstanceDb,
    used_by_workspaces: Vec<String>,
}

async fn list_external_instance_pg_databases(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
) -> JsonResult<std::collections::BTreeMap<String, ExternalInstancePgDatabase>> {
    require_super_admin(&db, &authed).await?;
    let databases = windmill_common::external_instance_pg::external_instance_databases(&db).await?;
    let mut usages =
        windmill_common::external_instance_pg::external_instance_database_usages(&db).await?;
    Ok(Json(
        databases
            .into_iter()
            .map(|(name, status)| {
                let used_by_workspaces = usages.remove(&name).unwrap_or_default();
                (
                    name,
                    ExternalInstancePgDatabase {
                        status,
                        used_by_workspaces: used_by_workspaces.into_iter().collect(),
                    },
                )
            })
            .collect(),
    ))
}

async fn create_external_instance_pg_database(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path(dbname): Path<String>,
    Json(body): Json<SetupCustomInstanceDbBody>,
) -> JsonResult<()> {
    require_super_admin(&db, &authed).await?;
    let tag = body.tag.as_deref().unwrap_or("datatable");
    let mut tx = db.begin().await?;
    windmill_common::external_instance_pg::create_external_instance_database_unchecked(
        &db, &mut tx, &dbname, tag, None,
    )
    .await?;
    tx.commit().await?;
    windmill_audit::audit_oss::audit_log(
        &db,
        &authed,
        "settings.create_external_instance_pg_database",
        windmill_audit::ActionKind::Create,
        "global",
        Some(&authed.email),
        Some([("dbname", dbname.as_str()), ("tag", tag)].into()),
    )
    .await?;
    Ok(Json(()))
}

async fn drop_external_instance_pg_database(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path(dbname): Path<String>,
) -> JsonResult<()> {
    require_super_admin(&db, &authed).await?;
    // A data table naming a dropped database fails on every job, far from the drop that caused it.
    let mut tx = db.begin().await?;
    windmill_common::external_instance_pg::drop_external_instance_database_unchecked(
        &mut tx, &dbname, None,
    )
    .await?;
    tx.commit().await?;
    windmill_audit::audit_oss::audit_log(
        &db,
        &authed,
        "settings.drop_external_instance_pg_database",
        windmill_audit::ActionKind::Delete,
        "global",
        Some(&authed.email),
        Some([("dbname", dbname.as_str())].into()),
    )
    .await?;
    Ok(Json(()))
}

#[derive(Deserialize)]
struct SetupCustomInstanceDbBody {
    tag: Option<String>,
}

async fn setup_custom_instance_pg_database(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path(dbname): Path<String>,
    Json(body): Json<SetupCustomInstanceDbBody>,
) -> JsonResult<CustomInstanceDb> {
    // Before anything is recorded: the status written below replaces the registry entry, and with it
    // the workspace a fork copy is reserved for.
    require_super_admin(&db, &authed).await?;
    windmill_common::workspaces::ensure_instance_pg_available(&db).await?;
    // Fork cleanup checks and drops the database and its entry under this lock. Held from before
    // the setup creates the database to after its entry is written, neither lands on the other's
    // half-done state: a dropped database with its entry written back, or the reverse.
    let mut tx = db.begin().await?;
    windmill_common::datatable_roles::lock_instance_databases_governance(&mut tx, [dbname.trim()])
        .await?;
    let mut logs = CustomInstanceDbLogs::default();
    let result = setup_custom_instance_pg_database_inner(authed, &db, &dbname, &mut logs).await;
    let success = result.is_ok();
    let error = result.err().map(|e| e.to_string());
    let status = CustomInstanceDb {
        logs,
        success,
        error,
        tag: body.tag,
        used_by_workspaces: vec![],
        workspace_id: None,
    };
    let status_json = serde_json::to_value(&status).map_err(to_anyhow)?;
    // The fork reservation is carried over inside the write, from whatever the row holds then: a
    // rename migrating it while the setup above ran would otherwise be overwritten with the value
    // this request started from, stranding the copy under the archived workspace.
    let saved = sqlx::query_scalar::<_, serde_json::Value>(
        r#"UPDATE global_settings SET value = jsonb_set(value, '{databases}',
               COALESCE(value->'databases', '{}'::jsonb)
                   || jsonb_build_object($1::text, $2::jsonb || jsonb_build_object(
                          'workspace_id', value->'databases'->$1::text->'workspace_id')))
           WHERE name = 'custom_instance_pg_databases'
           RETURNING value->'databases'->$1::text"#,
    )
    .bind(&dbname)
    .bind(&status_json)
    .fetch_one(&mut *tx)
    .await?;
    tx.commit().await?;
    let status: CustomInstanceDb = serde_json::from_value(saved).map_err(to_anyhow)?;

    Ok(Json(status))
}

async fn setup_custom_instance_pg_database_inner(
    authed: ApiAuthed,
    db: &DB,
    dbname: &str,
    logs: &mut CustomInstanceDbLogs,
) -> Result<()> {
    require_super_admin(db, &authed).await?;
    logs.super_admin = "OK".to_string();
    let wmill_pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
    logs.database_credentials = "OK".to_string();

    // Validate name to ensure it only contains alphanumeric characters
    // Prevents SQL injection on the instance database
    lazy_static::lazy_static! {
        // Must start with a letter, then alphanumeric/underscore/hyphen
        static ref VALID_NAME: regex::Regex = regex::Regex::new(r"^[a-zA-Z][a-zA-Z0-9_-]*$").unwrap();
    }
    let dbname = dbname.trim();
    if dbname.is_empty() {
        return Err(error::Error::BadRequest(
            "Database name cannot be empty".to_string(),
        ));
    }
    // PostgreSQL identifier limit is 63 bytes
    if dbname.len() > 63 {
        return Err(error::Error::BadRequest(
            "Database name cannot exceed 63 characters".to_string(),
        ));
    }
    if !VALID_NAME.is_match(dbname) {
        return Err(error::Error::BadRequest(
            "Database name must start with a letter and contain only alphanumeric characters, underscores, or hyphens".to_string(),
        ));
    }
    // Additional check: block PostgreSQL reserved/special names
    let lower = dbname.to_lowercase();
    if lower == "template0" || lower == "template1" || lower == "postgres" {
        return Err(error::Error::BadRequest(
            "Cannot use reserved PostgreSQL database names".to_string(),
        ));
    }
    if wmill_pg_creds
        .dbname
        .trim()
        .eq_ignore_ascii_case(dbname.trim())
    {
        return Err(error::Error::BadRequest(
            "Database name cannot be the same as the main database".to_string(),
        ));
    }
    logs.valid_dbname = "OK".to_string();

    let db_exists = sqlx::query_scalar!(
        "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)",
        dbname
    )
    .fetch_one(db)
    .await?
    .unwrap_or(false);

    let pg_creds = PgDatabase { dbname: dbname.to_string(), ..wmill_pg_creds };

    logs.created_database = "SKIP".to_string();
    if !db_exists {
        // SAFETY: `dbname` has been validated by the VALID_NAME regex and length checks above (lines 1088–1120).
        sqlx::query(&format!("CREATE DATABASE \"{dbname}\""))
            .execute(db)
            .await?;
        logs.created_database = "OK".to_string();
    }

    // We have to connect to the newly created database as admin to grant permissions
    let (client, connection) = pg_creds.connect(Some(db)).await?;
    let join_handle = tokio::spawn(async move { connection.await });

    logs.db_connect = "OK".to_string();

    // SAFETY: `dbname` has been validated by the VALID_NAME regex and length checks above.
    client
        .batch_execute(&format!(
            "GRANT CONNECT ON DATABASE \"{dbname}\" TO custom_instance_user;
             GRANT USAGE ON SCHEMA public TO custom_instance_user;
             GRANT CREATE ON SCHEMA public TO custom_instance_user;
             GRANT CREATE ON DATABASE \"{dbname}\" TO custom_instance_user;
             ALTER DEFAULT PRIVILEGES IN SCHEMA public
                 GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO custom_instance_user;
             ALTER ROLE custom_instance_user CREATEROLE;"
        ))
        .await
        .map_err(|e| {
            error::Error::ExecutionErr(format!(
                "Failed to grant permissions to custom_instance_user: {}",
                pg_error_message(&e),
            ))
        })?;

    logs.grant_permissions = "OK".to_string();

    drop(client); // /!\ Drop before joining to avoid deadlock
    windmill_common::shutdown_pg_connection(join_handle).await?;

    // Roles are cluster-wide, so the dedicated role used by postgres trigger connections is
    // provisioned on the main pool rather than on the new database. Reported as its own step
    // rather than failing the setup: without the role the database still serves datatables, only
    // postgres triggers on them break.
    match windmill_common::utils::ensure_custom_instance_replication_user(db).await {
        Ok(()) => logs.replication_user = "OK".to_string(),
        Err(e) => {
            tracing::error!("Failed to provision custom_instance_replication_user: {e:#}");
            logs.replication_user = "FAIL".to_string();
            logs.replication_user_error = Some(e.to_string());
        }
    }

    // Everything above logged in as the DATABASE_URL user, but whatever uses the database logs
    // in as custom_instance_user. A proxy that routes on the login name can accept one and
    // refuse the other, which would otherwise surface only once a data table first connects.
    let user_creds = PgDatabase {
        user: Some(windmill_common::datatable_roles::CUSTOM_INSTANCE_USER.to_string()),
        password: Some(windmill_common::utils::get_custom_pg_instance_password(db).await?),
        ..pg_creds
    };
    let (client, connection) = user_creds
        .connect(Some(db))
        .await
        .map_err(|e| custom_instance_user_connect_error(&user_creds.host, dbname, e))?;
    let join_handle = tokio::spawn(async move { connection.await });
    logs.user_connect = "OK".to_string();
    drop(client); // /!\ Drop before joining to avoid deadlock
    windmill_common::shutdown_pg_connection(join_handle).await?;

    Ok(())
}

fn custom_instance_user_connect_error(host: &str, dbname: &str, e: error::Error) -> error::Error {
    let cause = match &e {
        error::Error::Anyhow { error, .. } => format!("{error:#}"),
        e => e.to_string(),
    };
    // Supavisor, Supabase's pooler, reads the tenant to route to from the login
    // (`<user>.<project_ref>`), and only knows the logins configured for that tenant.
    let lower = cause.to_lowercase();
    let routing_refused = [
        "enoidentifier",
        "tenant identifier",
        "tenant or user",
        "tenant/user",
    ]
    .iter()
    .any(|signature| lower.contains(signature));
    if routing_refused {
        error::Error::BadConfig(format!(
            "DATABASE_URL reaches Postgres through a connection pooler ({host}) that picks the \
             server to route to from the login name, and it refused custom_instance_user, the \
             role Windmill uses for instance databases ({cause}). Instance databases cannot be \
             used through this pooler: use your own Postgres database instead, or point \
             DATABASE_URL at the Postgres server directly rather than at the pooler."
        ))
    } else {
        error::Error::ExecutionErr(format!(
            "Could not connect to {dbname} as custom_instance_user: {cause}"
        ))
    }
}

async fn drop_custom_instance_pg_database(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path(dbname): Path<String>,
) -> Result<String> {
    require_super_admin(&db, &authed).await?;

    windmill_common::drop_custom_instance_database(&db, &dbname).await?;

    Ok(format!("Database '{}' dropped successfully", dbname))
}

// ============================================================================
// Secret Backend Settings (HashiCorp Vault Integration) - Enterprise Edition
// ============================================================================

/// Test connection to a secret backend (HashiCorp Vault)
///
/// This endpoint validates that the Vault settings are correct and that
/// Windmill can successfully authenticate and communicate with Vault.
///
/// This is an Enterprise Edition feature.
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn test_secret_backend(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<VaultSettings>,
) -> Result<String> {
    require_super_admin(&db, &authed).await?;

    windmill_common::secret_backend::test_vault_connection(&settings, Some(&db)).await?;

    Ok("Successfully connected to HashiCorp Vault".to_string())
}

/// Migrate existing secrets from database to HashiCorp Vault
///
/// This endpoint reads all encrypted secrets from the database, decrypts them,
/// and stores them in HashiCorp Vault. The database values are NOT deleted
/// automatically to allow for rollback if needed.
///
/// This is an Enterprise Edition feature.
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn migrate_secrets_to_vault(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<VaultSettings>,
) -> JsonResult<SecretMigrationReport> {
    require_super_admin(&db, &authed).await?;

    let report = windmill_common::secret_backend::migrate_secrets_to_vault(&db, &settings).await?;

    Ok(Json(report))
}

/// Migrate secrets from HashiCorp Vault back to database
///
/// This endpoint reads all secrets from HashiCorp Vault, encrypts them using
/// the workspace encryption keys, and stores them in the database. The Vault
/// values are NOT deleted automatically to allow for rollback if needed.
///
/// This is an Enterprise Edition feature.
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn migrate_secrets_to_database(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<VaultSettings>,
) -> JsonResult<SecretMigrationReport> {
    require_super_admin(&db, &authed).await?;

    let report =
        windmill_common::secret_backend::migrate_secrets_to_database(&db, &settings).await?;

    Ok(Json(report))
}

/// Test connection to Azure Key Vault
///
/// This is an Enterprise Edition feature.
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn test_azure_kv_backend(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<AzureKeyVaultSettings>,
) -> Result<String> {
    require_super_admin(&db, &authed).await?;

    windmill_common::secret_backend::test_azure_kv_connection(&settings).await?;

    Ok("Successfully connected to Azure Key Vault".to_string())
}

/// Migrate existing secrets from database to Azure Key Vault
///
/// This is an Enterprise Edition feature.
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn migrate_secrets_to_azure_kv(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<AzureKeyVaultSettings>,
) -> JsonResult<SecretMigrationReport> {
    require_super_admin(&db, &authed).await?;

    let report =
        windmill_common::secret_backend::migrate_secrets_to_azure_kv(&db, &settings).await?;

    Ok(Json(report))
}

/// Migrate secrets from Azure Key Vault back to database
///
/// This is an Enterprise Edition feature.
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn migrate_secrets_from_azure_kv(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<AzureKeyVaultSettings>,
) -> JsonResult<SecretMigrationReport> {
    require_super_admin(&db, &authed).await?;

    let report =
        windmill_common::secret_backend::migrate_secrets_from_azure_kv(&db, &settings).await?;

    Ok(Json(report))
}

/// Test connection to AWS Secrets Manager
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn test_aws_sm_backend(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<AwsSecretsManagerSettings>,
) -> Result<String> {
    require_super_admin(&db, &authed).await?;
    windmill_common::secret_backend::test_aws_sm_connection(&settings).await?;
    Ok("Successfully connected to AWS Secrets Manager".to_string())
}

/// Migrate existing secrets from database to AWS Secrets Manager
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn migrate_secrets_to_aws_sm(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<AwsSecretsManagerSettings>,
) -> JsonResult<SecretMigrationReport> {
    require_super_admin(&db, &authed).await?;
    let report = windmill_common::secret_backend::migrate_secrets_to_aws_sm(&db, &settings).await?;
    Ok(Json(report))
}

/// Migrate secrets from AWS Secrets Manager back to database
#[cfg(all(feature = "private", feature = "enterprise"))]
pub async fn migrate_secrets_from_aws_sm(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Json(settings): Json<AwsSecretsManagerSettings>,
) -> JsonResult<SecretMigrationReport> {
    require_super_admin(&db, &authed).await?;
    let report =
        windmill_common::secret_backend::migrate_secrets_from_aws_sm(&db, &settings).await?;
    Ok(Json(report))
}

// ============================================================================
// JWKS Endpoint for Vault JWT Authentication
// ============================================================================

/// JSON Web Key Set response structure
#[derive(Serialize)]
pub struct JwksResponse {
    pub keys: Vec<serde_json::Value>,
}

/// Fallback JWKS endpoint used when OIDC support is not compiled in.
///
/// When built with `private` + `enterprise` + `openidconnect`, the route in
/// `windmill-api` dispatches to `oidc_oss::jwks` (re-exported from
/// `oidc_ee::jwks`) instead, which serves the actual public keys.
pub async fn get_jwks() -> JsonResult<JwksResponse> {
    Ok(Json(JwksResponse { keys: vec![] }))
}

#[derive(serde::Deserialize, serde::Serialize)]
struct CachedResourceType {
    #[allow(dead_code)]
    id: i64,
    name: String,
    schema: Option<serde_json::Value>,
    #[allow(dead_code)]
    app: String,
    description: Option<String>,
    /// Doubly optional, and read through a wrapping deserializer: this struct also
    /// decodes the on-disk cache, where an absent key means "written before the
    /// column, leave the stored extension alone" and an explicit null means the hub
    /// dropped it. Plain serde folds both into `None`.
    #[serde(
        default,
        deserialize_with = "windmill_common::more_serde::double_option"
    )]
    format_extension: Option<Option<String>>,
    /// Doubly optional like `format_extension`: no key leaves the stored name alone, an explicit
    /// null (the hub naming nothing) clears it.
    #[serde(
        default,
        deserialize_with = "windmill_common::more_serde::double_option"
    )]
    display_name: Option<Option<String>>,
}

#[derive(serde::Deserialize)]
struct HubResourceTypeRaw {
    id: i64,
    name: String,
    schema: Option<String>,
    app: String,
    description: Option<String>,
    #[serde(default)]
    format_extension: Option<String>,
    #[serde(
        default,
        deserialize_with = "windmill_common::more_serde::double_option"
    )]
    display_name: Option<Option<String>>,
}

async fn fetch_resource_types_from_hub() -> error::Result<Vec<CachedResourceType>> {
    let response = HTTP_CLIENT
        .get(format!(
            "{}/resource_types/list",
            windmill_common::DEFAULT_HUB_BASE_URL
        ))
        .header("Accept", "application/json")
        .send()
        .await
        .map_err(|e| error::Error::InternalErr(format!("Failed to fetch from hub: {}", e)))?;

    if !response.status().is_success() {
        return Err(error::Error::InternalErr(format!(
            "Hub returned status {}",
            response.status()
        )));
    }

    let raw_types: Vec<HubResourceTypeRaw> = response
        .json()
        .await
        .map_err(|e| error::Error::InternalErr(format!("Failed to parse hub response: {}", e)))?;

    Ok(raw_types
        .into_iter()
        .filter_map(|rt| {
            let schema = match rt.schema {
                Some(s) => match serde_json::from_str(&s) {
                    Ok(v) => Some(v),
                    Err(_) => return None,
                },
                None => None,
            };
            Some(CachedResourceType {
                id: rt.id,
                name: rt.name,
                schema,
                app: rt.app,
                description: rt.description,
                format_extension: Some(rt.format_extension),
                display_name: rt.display_name,
            })
        })
        .collect())
}

#[derive(serde::Deserialize)]
struct SyncResourceTypesQuery {
    name: Option<String>,
}

async fn sync_cached_resource_types(
    Extension(db): Extension<DB>,
    authed: ApiAuthed,
    Query(SyncResourceTypesQuery { name }): Query<SyncResourceTypesQuery>,
) -> error::Result<String> {
    require_super_admin(&db, &authed).await?;

    use windmill_common::worker::HUB_RT_CACHE_DIR;
    let cache_path = format!("{}/resource_types.json", *HUB_RT_CACHE_DIR);

    // Manual sync is hub-first so it lands newly-published hub types on demand. The
    // on-disk cache is only a fallback for when the hub is unreachable (airgapped
    // installs / network error); refreshing it is left to the daily cache-rt cron and
    // the startup sync in main.rs, which own the offline path.
    let (resource_types, from_hub) = match fetch_resource_types_from_hub().await {
        Ok(types) => {
            tracing::info!("Fetched {} resource types live from the hub", types.len());
            (types, true)
        }
        Err(hub_err) => {
            tracing::warn!(
                "Live hub fetch failed ({hub_err}), falling back to on-disk cache at {cache_path}"
            );
            match tokio::fs::read_to_string(&cache_path).await {
                Ok(content) => {
                    let parsed = serde_json::from_str::<Vec<CachedResourceType>>(&content)
                        .map_err(|e| {
                            error::Error::InternalErr(format!(
                                "Failed to parse cached resource types: {}",
                                e
                            ))
                        })?;
                    (parsed, false)
                }
                Err(_) => return Err(hub_err),
            }
        }
    };

    let mut synced_count = 0;

    for rt in &resource_types {
        // A name too long for the column counts as absent, leaving the stored one alone: one bad
        // entry must not fail the upsert and end the rest of the sync.
        let display_name = match &rt.display_name {
            Some(Some(name)) if name.chars().count() > 100 => None,
            other => other.clone(),
        };
        let exists: Option<bool> = sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM resource_type WHERE workspace_id = 'admins' AND name = $1 AND schema IS NOT DISTINCT FROM $2 AND description IS NOT DISTINCT FROM $3 AND ($5 IS NOT TRUE OR format_extension IS NOT DISTINCT FROM $4) AND ($7 IS NOT TRUE OR display_name IS NOT DISTINCT FROM $6))",
            &rt.name,
            rt.schema.as_ref(),
            rt.description.as_deref(),
            rt.format_extension.clone().flatten(),
            rt.format_extension.is_some(),
            display_name.clone().flatten(),
            display_name.is_some(),
        )
        .fetch_one(&db)
        .await?;

        if exists.unwrap_or(false) {
            continue;
        }

        sqlx::query!(
            // Whether the payload carried the key at all is what decides: present
            // (even as null) is authoritative and may clear, absent means a cache
            // written before the column and must leave the stored value alone.
            "INSERT INTO resource_type (workspace_id, name, schema, description, format_extension, display_name, edited_at)
             VALUES ('admins', $1, $2, $3, $4, $6, now())
             ON CONFLICT (workspace_id, name) DO UPDATE
             SET schema = EXCLUDED.schema, description = EXCLUDED.description,
                 -- A fileset is a set of files, so it cannot also be one file.
                 -- Create and update reject the pair; this writer bypasses both, so
                 -- it declines the extension rather than persisting the forbidden
                 -- combination onto a same-named local fileset.
                 format_extension = CASE
                     WHEN resource_type.is_fileset THEN NULL
                     WHEN $5 THEN EXCLUDED.format_extension
                     ELSE resource_type.format_extension END,
                 display_name = CASE WHEN $7 THEN EXCLUDED.display_name ELSE resource_type.display_name END,
                 edited_at = now()",
            &rt.name,
            rt.schema.as_ref(),
            rt.description.as_deref(),
            rt.format_extension.clone().flatten(),
            rt.format_extension.is_some(),
            display_name.clone().flatten(),
            display_name.is_some(),
        )
        .execute(&db)
        .await?;

        synced_count += 1;
    }

    // If a specific type was requested and is still absent after syncing, surface an
    // explicit not-found instead of a silent "Synced 0". Word it by source so the
    // cache-fallback path does not claim it checked the hub.
    if let Some(name) = name.as_deref() {
        if !resource_types.iter().any(|rt| rt.name == name) {
            let source = if from_hub {
                "on the hub"
            } else {
                "in the cached resource types (hub unreachable)"
            };
            return Err(error::Error::NotFound(format!(
                "resource type '{}' not found {}",
                name, source
            )));
        }
    }

    Ok(format!(
        "Synced {} resource types ({} unchanged)",
        synced_count,
        resource_types.len() - synced_count
    ))
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use windmill_common::instance_config::{GlobalSettings, InstanceConfig, WorkerGroupConfig};

    #[test]
    fn supavisor_refusing_custom_instance_user_is_named() {
        use windmill_common::error::{to_anyhow, Error};
        let connect_error = |message: &str| {
            let e = Error::from(to_anyhow(std::io::Error::other(message.to_string())));
            super::custom_instance_user_connect_error(
                "aws-0-eu-west-1.pooler.supabase.com",
                "dt",
                e,
            )
        };
        for supavisor in [
            "db error: FATAL: (ENOIDENTIFIER) no tenant identifier provided",
            "db error: FATAL: Tenant or user not found",
        ] {
            assert!(
                matches!(connect_error(supavisor), Error::BadConfig(m) if m.contains("connection pooler")),
                "{supavisor}"
            );
        }
        assert!(matches!(
            connect_error(
                "db error: FATAL: password authentication failed for user \"custom_instance_user\""
            ),
            Error::ExecutionErr(_)
        ));
    }

    #[test]
    fn instance_config_yaml_round_trip() {
        let config = InstanceConfig {
            global_settings: GlobalSettings {
                base_url: Some("https://windmill.example.com".to_string()),
                retention_period_secs: Some(86400),
                expose_metrics: Some(true),
                ..Default::default()
            },
            worker_configs: BTreeMap::from([(
                "default".to_string(),
                WorkerGroupConfig {
                    worker_tags: Some(vec!["deno".to_string(), "python3".to_string()]),
                    init_bash: Some("apt-get update".to_string()),
                    ..Default::default()
                },
            )]),
        };

        let yaml = config.to_sorted_yaml().unwrap();

        // Verify key fields appear in the YAML output
        assert!(yaml.contains("base_url: https://windmill.example.com"));
        assert!(yaml.contains("retention_period_secs: 86400"));
        assert!(yaml.contains("expose_metrics: true"));
        assert!(yaml.contains("default:"));
        assert!(yaml.contains("- deno"));
        assert!(yaml.contains("- python3"));
        assert!(yaml.contains("init_bash: apt-get update"));

        // Round-trip back to struct
        let deserialized: InstanceConfig = serde_yml::from_str(&yaml).unwrap();
        assert_eq!(
            deserialized.global_settings.base_url.as_deref(),
            Some("https://windmill.example.com")
        );
        assert_eq!(
            deserialized.global_settings.retention_period_secs,
            Some(86400)
        );
        assert_eq!(deserialized.global_settings.expose_metrics, Some(true));
        let wc = &deserialized.worker_configs["default"];
        assert_eq!(
            wc.worker_tags.as_deref(),
            Some(["deno".to_string(), "python3".to_string()].as_slice())
        );
        assert_eq!(wc.init_bash.as_deref(), Some("apt-get update"));
    }

    #[test]
    fn sorted_yaml_global_settings_alphabetical() {
        let config = InstanceConfig {
            global_settings: GlobalSettings {
                retention_period_secs: Some(3600),
                base_url: Some("https://test.com".to_string()),
                expose_metrics: Some(true),
                email_domain: Some("example.com".to_string()),
                ..Default::default()
            },
            worker_configs: BTreeMap::new(),
        };

        let yaml = config.to_sorted_yaml().unwrap();

        // Keys must appear in alphabetical order
        let base_url_pos = yaml.find("base_url:").unwrap();
        let email_pos = yaml.find("email_domain:").unwrap();
        let expose_pos = yaml.find("expose_metrics:").unwrap();
        let retention_pos = yaml.find("retention_period_secs:").unwrap();

        assert!(
            base_url_pos < email_pos && email_pos < expose_pos && expose_pos < retention_pos,
            "global_settings keys should be alphabetically sorted, got yaml:\n{yaml}"
        );
    }

    #[test]
    fn sorted_yaml_worker_configs_default_and_native_first() {
        let config = InstanceConfig {
            global_settings: GlobalSettings::default(),
            worker_configs: BTreeMap::from([
                (
                    "gpu".to_string(),
                    WorkerGroupConfig {
                        init_bash: Some("echo gpu".to_string()),
                        ..Default::default()
                    },
                ),
                (
                    "native".to_string(),
                    WorkerGroupConfig {
                        init_bash: Some("echo native".to_string()),
                        ..Default::default()
                    },
                ),
                (
                    "default".to_string(),
                    WorkerGroupConfig {
                        init_bash: Some("echo default".to_string()),
                        ..Default::default()
                    },
                ),
                (
                    "alpha".to_string(),
                    WorkerGroupConfig {
                        init_bash: Some("echo alpha".to_string()),
                        ..Default::default()
                    },
                ),
            ]),
        };

        let yaml = config.to_sorted_yaml().unwrap();

        let default_pos = yaml.find("default:").unwrap();
        let native_pos = yaml.find("native:").unwrap();
        let alpha_pos = yaml.find("alpha:").unwrap();
        let gpu_pos = yaml.find("gpu:").unwrap();

        assert!(
            default_pos < native_pos
                && native_pos < alpha_pos
                && alpha_pos < gpu_pos,
            "worker_configs should have default, native first, then rest alphabetically, got yaml:\n{yaml}"
        );
    }

    #[test]
    fn sorted_yaml_roundtrips() {
        let config = InstanceConfig {
            global_settings: GlobalSettings {
                base_url: Some("https://rt.test".to_string()),
                retention_period_secs: Some(7200),
                expose_metrics: Some(false),
                ..Default::default()
            },
            worker_configs: BTreeMap::from([
                (
                    "default".to_string(),
                    WorkerGroupConfig {
                        worker_tags: Some(vec!["deno".to_string()]),
                        ..Default::default()
                    },
                ),
                (
                    "native".to_string(),
                    WorkerGroupConfig {
                        init_bash: Some("echo hi".to_string()),
                        ..Default::default()
                    },
                ),
            ]),
        };

        let yaml = config.to_sorted_yaml().unwrap();
        let deserialized: InstanceConfig = serde_yml::from_str(&yaml).unwrap();

        assert_eq!(
            deserialized.global_settings.base_url.as_deref(),
            Some("https://rt.test")
        );
        assert_eq!(
            deserialized.global_settings.retention_period_secs,
            Some(7200)
        );
        assert_eq!(deserialized.global_settings.expose_metrics, Some(false));
        assert_eq!(deserialized.worker_configs.len(), 2);
        assert_eq!(
            deserialized.worker_configs["default"]
                .worker_tags
                .as_deref(),
            Some(["deno".to_string()].as_slice())
        );
        assert_eq!(
            deserialized.worker_configs["native"].init_bash.as_deref(),
            Some("echo hi")
        );
    }
}

#[cfg(all(test, feature = "parquet"))]
mod object_storage_test_hardening {
    use super::{extract_host, validate_object_storage_test};
    use windmill_object_store::ObjectSettings;

    // IP literals (not hostnames) keep validate_public_endpoint deterministic — `lookup_host`
    // parses them without any network round-trip.
    fn gcs_settings(gcs_base_url: &str) -> ObjectSettings {
        serde_json::from_value(serde_json::json!({
            "type": "Gcs",
            "bucket": "b",
            "serviceAccountKey": { "gcs_base_url": gcs_base_url, "client_email": "x@y.z" }
        }))
        .unwrap()
    }

    #[tokio::test]
    async fn rejects_gcs_internal_base_url() {
        // gcs_base_url in the service-account key must not smuggle an internal host past the check,
        // including via a mixed-case scheme (URL schemes are case-insensitive).
        for url in [
            "http://169.254.169.254",
            "HTTP://169.254.169.254",
            "Https://10.0.0.5",
        ] {
            assert!(
                validate_object_storage_test(&gcs_settings(url))
                    .await
                    .is_err(),
                "{url} should be rejected"
            );
        }
    }

    #[tokio::test]
    async fn allows_gcs_public_base_url() {
        assert!(
            validate_object_storage_test(&gcs_settings("https://8.8.8.8"))
                .await
                .is_ok()
        );
    }

    #[tokio::test]
    async fn rejects_gcs_blank_service_account_key() {
        // A blank key makes build_gcs_client fall back to the instance's ambient credentials, so an
        // untrusted caller must not be allowed to test with it. The `serviceAccountKey` field is
        // serialized via serde's `as_string` (`to_string` of the JSON value), so the settings UI's
        // "no key" empty object arrives as `"{}"` and a null as `"null"` — both must be rejected.
        for key in [serde_json::json!({}), serde_json::json!(null)] {
            let settings: ObjectSettings = serde_json::from_value(serde_json::json!({
                "type": "Gcs",
                "bucket": "b",
                "serviceAccountKey": key
            }))
            .unwrap();
            assert!(
                validate_object_storage_test(&settings).await.is_err(),
                "blank key {key:?} should be rejected"
            );
        }
    }

    #[test]
    fn extracts_host_from_endpoint() {
        let cases = [
            ("s3.amazonaws.com", Some("s3.amazonaws.com")),
            ("https://minio.internal:9000", Some("minio.internal")),
            ("http://10.0.0.5:9000/bucket", Some("10.0.0.5")),
            ("user:pass@host.example:443", Some("host.example")),
            ("[::1]:9000", Some("::1")),
            ("https://[fe80::1]/x", Some("fe80::1")),
            ("", None),
            // Injection via region/bucket interpolation into the default endpoint string: the
            // userinfo `@` and the path `/` must not hide the real authority from the host check.
            (
                "https://s3.@169.254.169.254/.amazonaws.com",
                Some("169.254.169.254"),
            ),
            (
                "https://@169.254.169.254/mybucket.s3.amazonaws.com",
                Some("169.254.169.254"),
            ),
            ("s3.#@169.254.169.254/x.amazonaws.com", Some("s3.")),
            // Scheme is case-insensitive.
            ("HTTP://169.254.169.254", Some("169.254.169.254")),
        ];
        for (input, expected) in cases {
            assert_eq!(extract_host(input).as_deref(), expected, "input: {input}");
        }
    }
}