use std::collections::BTreeMap;
use std::fmt;

use serde::{Deserialize, Serialize};

use crate::global_settings::LICENSE_KEY_SETTING;

// ---------------------------------------------------------------------------
// Kubernetes Secret reference support
// ---------------------------------------------------------------------------

/// Reference to a key inside a Kubernetes Secret.
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct SecretKeyRef {
    pub name: String,
    pub key: String,
}

/// Wrapper that produces `{ "secretKeyRef": { "name": "…", "key": "…" } }` in YAML/JSON.
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct SecretKeyRefWrapper {
    #[serde(rename = "secretKeyRef")]
    pub secret_key_ref: SecretKeyRef,
}

/// Reference to an environment variable.
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct EnvRefWrapper {
    #[serde(rename = "envRef")]
    pub env_ref: String,
}

/// A string field that can be a literal value, a reference to a Kubernetes
/// Secret key, or a reference to an environment variable.
///
/// Uses `#[serde(untagged)]` so that in YAML/JSON:
///
/// - A plain string deserializes as `Literal("…")`
/// - An object `{ secretKeyRef: { name, key } }` deserializes as `SecretRef(…)`
/// - An object `{ envRef: "VAR_NAME" }` deserializes as `EnvRef(…)`
///
/// `Literal` serializes back to a plain JSON string, preserving backwards
/// compatibility with existing consumers.
#[derive(Deserialize, Serialize, Clone)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
#[serde(untagged)]
pub enum StringOrSecretRef {
    Literal(String),
    SecretRef(SecretKeyRefWrapper),
    EnvRef(EnvRefWrapper),
}

impl fmt::Debug for StringOrSecretRef {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Literal(_) => f.write_str("Literal(****)"),
            Self::SecretRef(w) => f.debug_tuple("SecretRef").field(w).finish(),
            Self::EnvRef(w) => f.debug_tuple("EnvRef").field(w).finish(),
        }
    }
}

impl StringOrSecretRef {
    /// Returns the literal string value, or `None` if this is an unresolved ref.
    pub fn as_literal(&self) -> Option<&str> {
        match self {
            Self::Literal(s) => Some(s),
            Self::SecretRef(_) | Self::EnvRef(_) => None,
        }
    }

    /// Returns the literal value, panicking if this is an unresolved ref.
    /// Use only after resolution.
    pub fn literal_value(&self) -> &str {
        match self {
            Self::Literal(s) => s,
            Self::SecretRef(_) => panic!("literal_value() called on unresolved secret ref"),
            Self::EnvRef(_) => panic!("literal_value() called on unresolved env ref"),
        }
    }

    /// Returns `true` if this is a secret reference (not yet resolved).
    pub fn is_secret_ref(&self) -> bool {
        matches!(self, Self::SecretRef(_))
    }

    /// Returns the inner `SecretKeyRef` if this is a secret reference.
    pub fn as_secret_ref(&self) -> Option<&SecretKeyRef> {
        match self {
            Self::SecretRef(w) => Some(&w.secret_key_ref),
            _ => None,
        }
    }

    /// Returns `true` if this is an environment variable reference.
    pub fn is_env_ref(&self) -> bool {
        matches!(self, Self::EnvRef(_))
    }

    /// Returns the env var name if this is an env ref.
    pub fn as_env_ref(&self) -> Option<&str> {
        match self {
            Self::EnvRef(w) => Some(&w.env_ref),
            _ => None,
        }
    }
}

impl Default for StringOrSecretRef {
    fn default() -> Self {
        Self::Literal(String::new())
    }
}

impl From<String> for StringOrSecretRef {
    fn from(s: String) -> Self {
        Self::Literal(s)
    }
}

impl From<&str> for StringOrSecretRef {
    fn from(s: &str) -> Self {
        Self::Literal(s.to_string())
    }
}

impl PartialEq<str> for StringOrSecretRef {
    fn eq(&self, other: &str) -> bool {
        self.as_literal() == Some(other)
    }
}

impl PartialEq<StringOrSecretRef> for StringOrSecretRef {
    fn eq(&self, other: &StringOrSecretRef) -> bool {
        match (self, other) {
            (Self::Literal(a), Self::Literal(b)) => a == b,
            (Self::SecretRef(a), Self::SecretRef(b)) => a == b,
            (Self::EnvRef(a), Self::EnvRef(b)) => a == b,
            _ => false,
        }
    }
}

impl fmt::Display for StringOrSecretRef {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Literal(s) => f.write_str(s),
            Self::SecretRef(w) => write!(
                f,
                "secretKeyRef({}/{})",
                w.secret_key_ref.name, w.secret_key_ref.key
            ),
            Self::EnvRef(w) => write!(f, "envRef({})", w.env_ref),
        }
    }
}

// ---------------------------------------------------------------------------
// Top-level wrapper
// ---------------------------------------------------------------------------

/// Unified instance configuration combining global settings and worker configs.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct InstanceConfig {
    #[serde(default)]
    pub global_settings: GlobalSettings,
    #[serde(default)]
    pub worker_configs: BTreeMap<String, WorkerGroupConfig>,
}

// ---------------------------------------------------------------------------
// Global settings
// ---------------------------------------------------------------------------

/// Generate a schema for opaque JSON objects (used for EE-private settings).
/// Produces `{"type": "object", "nullable": true}` so the CRD passes K8s
/// structural schema validation while still accepting any JSON object.
#[cfg(feature = "instance_config_schema")]
fn opaque_json_schema(_: &mut schemars::gen::SchemaGenerator) -> schemars::schema::Schema {
    schemars::schema::SchemaObject {
        instance_type: Some(schemars::schema::InstanceType::Object.into()),
        metadata: Some(Box::default()),
        extensions: {
            let mut m = schemars::Map::new();
            m.insert("nullable".to_string(), serde_json::Value::Bool(true));
            m.insert(
                "x-kubernetes-preserve-unknown-fields".to_string(),
                serde_json::Value::Bool(true),
            );
            m
        },
        ..Default::default()
    }
    .into()
}

/// Typed global settings with schema validation.
/// Known settings have explicit fields; unknown settings pass through via `extra`.
///
/// `#[serde(remote = "Self")]` turns the derived (de)serializers into inherent
/// associated functions so we can wrap them with the manual `Deserialize` impl
/// below. The wrapper preserves explicit `null` values (which the typed
/// `Option<T>` fields would otherwise silently drop on round-trip through
/// `to_settings_map`) by stashing them in `extra`, where they survive
/// re-serialization and reach `diff_global_settings` as proper deletes.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[serde(remote = "Self")]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct GlobalSettings {
    // Numeric settings
    #[serde(skip_serializing_if = "Option::is_none")]
    pub base_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub license_key: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub retention_period_secs: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub job_default_timeout: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub nsjail_tmpfs_size_mb: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub nsjail_tmp_backing: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub bun_install_min_release_age: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub uv_exclude_newer: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub request_size_limit_mb: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timeout_wait_result: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cancel_stranded_jobs_after_days: Option<i64>,

    // Boolean settings
    #[serde(skip_serializing_if = "Option::is_none")]
    pub expose_metrics: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub expose_debug_metrics: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub keep_job_dir: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub disable_stats: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub require_preexisting_user_for_oauth: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dev_instance: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub critical_alert_mute_ui: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub critical_alert_mute_stranded_jobs: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub monitor_logs_on_s3: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub app_workspaced_route: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub http_route_workspaced_route: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub no_default_maven: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub default_tags_per_workspace: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub preview_tags_override: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub disable_hub: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub auto_build_binary_on_deploy: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ai_sessions_instance_storage_fallback: Option<bool>,

    // String settings
    #[serde(skip_serializing_if = "Option::is_none")]
    pub auto_build_binary_tag: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dependency_job_tag: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ws_base_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub github_app_webhook_base_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub email_domain: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hub_base_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hub_accessible_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub hub_api_secret: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub jwt_secret: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scim_token: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub saml_metadata: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub instance_events_webhook: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub openai_azure_base_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub min_keep_alive_version: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub instance_python_version: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub pip_index_url: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub pip_extra_index_url: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ruff_config: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub npm_config_registry: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub bunfig_install_scopes: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub npmrc: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub nuget_config: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub maven_repos: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ruby_repos: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub powershell_repo_url: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub powershell_repo_pat: Option<StringOrSecretRef>,

    // Array settings
    #[serde(skip_serializing_if = "Option::is_none")]
    pub custom_tags: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub default_tags_workspaces: Option<Vec<String>>,

    // Structured settings
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_settings: Option<SmtpSettings>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub indexer_settings: Option<IndexerSettings>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub oauths: Option<BTreeMap<String, OAuthClientEntry>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub otel: Option<OtelSettings>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub otel_tracing_proxy: Option<OtelTracingProxySettings>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(
        feature = "instance_config_schema",
        schemars(schema_with = "opaque_json_schema")
    )]
    pub object_store_cache_config: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub critical_error_channels: Option<Vec<CriticalErrorChannel>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub critical_alerts_on_db_oversize: Option<DbOversizeAlert>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub ducklake_settings: Option<DucklakeSettings>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub custom_instance_pg_databases: Option<CustomInstancePgDatabases>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub external_instance_pg: Option<ExternalInstancePg>,

    // Opaque settings (EE-private structs or no clear schema)
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(
        feature = "instance_config_schema",
        schemars(schema_with = "opaque_json_schema")
    )]
    pub secret_backend: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(
        feature = "instance_config_schema",
        schemars(schema_with = "opaque_json_schema")
    )]
    pub slack: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(
        feature = "instance_config_schema",
        schemars(schema_with = "opaque_json_schema")
    )]
    pub teams: Option<serde_json::Value>,

    // Workspace-specific registry overrides
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(
        feature = "instance_config_schema",
        schemars(schema_with = "opaque_json_schema")
    )]
    pub workspace_registries: Option<
        std::collections::HashMap<String, std::collections::HashMap<String, serde_json::Value>>,
    >,

    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(
        feature = "instance_config_schema",
        schemars(schema_with = "opaque_json_schema")
    )]
    pub ai_config: Option<serde_json::Value>,

    /// Catch-all for settings not yet covered by typed fields.
    #[serde(flatten)]
    pub extra: BTreeMap<String, serde_json::Value>,
}

impl Serialize for GlobalSettings {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        Self::serialize(self, serializer)
    }
}

impl<'de> Deserialize<'de> for GlobalSettings {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        // Capture top-level keys explicitly set to `null` so they survive the
        // `to_settings_map` round-trip — typed `Option<T>` fields all use
        // `skip_serializing_if = "Option::is_none"`, which would otherwise
        // silently drop the null. We stash the null entries in `extra`, which
        // serializes them back out as `null` for `diff_global_settings` to
        // route to deletes.
        let mut value = serde_json::Value::deserialize(deserializer)?;
        let null_keys: Vec<String> = value
            .as_object()
            .map(|m| {
                m.iter()
                    .filter_map(|(k, v)| v.is_null().then(|| k.clone()))
                    .collect()
            })
            .unwrap_or_default();
        if let Some(obj) = value.as_object_mut() {
            for k in &null_keys {
                obj.remove(k);
            }
        }
        let mut s = Self::deserialize(value).map_err(serde::de::Error::custom)?;
        for k in null_keys {
            s.extra.insert(k, serde_json::Value::Null);
        }
        Ok(s)
    }
}

impl GlobalSettings {
    /// Convert to a flat `BTreeMap` suitable for DB sync.
    pub fn to_settings_map(&self) -> BTreeMap<String, serde_json::Value> {
        // Serialize the whole struct to a JSON object, which flattens all fields
        // including `extra` into a single map. skip_serializing_if ensures
        // None fields are omitted.
        let value = serde_json::to_value(self).expect("GlobalSettings serialization cannot fail");
        let mut map: BTreeMap<String, serde_json::Value> = match value {
            serde_json::Value::Object(map) => map.into_iter().collect(),
            _ => unreachable!(),
        };
        // Strip runtime-only `databases` sub-field from custom_instance_pg_databases.
        // It contains setup status/logs managed by the setup endpoint, not configuration.
        if let Some(pg) = map.get_mut("custom_instance_pg_databases") {
            if let Some(obj) = pg.as_object_mut() {
                obj.remove("databases");
            }
        }
        map
    }
}

// ---------------------------------------------------------------------------
// SMTP
// ---------------------------------------------------------------------------

/// SMTP server configuration.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct SmtpSettings {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_host: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_username: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_password: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_port: Option<u16>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_from: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_tls_implicit: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_disable_tls: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub smtp_clicktracking_off: Option<bool>,
}

// ---------------------------------------------------------------------------
// Indexer
// ---------------------------------------------------------------------------

/// Full-text search indexer configuration.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct IndexerSettings {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub writer_memory_budget: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub commit_job_max_batch_size: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub commit_log_max_batch_size: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub refresh_index_period: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub refresh_log_index_period: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_indexed_job_log_size: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_index_time_window_secs: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub should_clear_job_index: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub should_clear_log_index: Option<bool>,
}

// ---------------------------------------------------------------------------
// OAuth
// ---------------------------------------------------------------------------

/// Wrapper that tries to deserialize as a typed [`OAuthClient`] and falls
/// back to a raw JSON value when the stored config contains unexpected types
/// (e.g. `"true"` instead of `true` for a boolean field).  This prevents the
/// entire `/api/settings/instance_config` endpoint from failing because of
/// one malformed provider entry.
#[derive(Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
#[serde(untagged)]
pub enum OAuthClientEntry {
    Typed(OAuthClient),
    /// Fallback: the raw JSON value plus the error that prevented typed parsing.
    Raw {
        value: serde_json::Value,
        /// Human-readable deserialization error (not serialized).
        #[serde(skip)]
        error: String,
    },
}

impl<'de> serde::Deserialize<'de> for OAuthClientEntry {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let raw = serde_json::Value::deserialize(deserializer)?;
        match serde_json::from_value::<OAuthClient>(raw.clone()) {
            Ok(client) => Ok(OAuthClientEntry::Typed(client)),
            Err(e) => Ok(OAuthClientEntry::Raw { value: raw, error: e.to_string() }),
        }
    }
}

impl OAuthClientEntry {
    /// Returns a mutable reference to the inner [`OAuthClient`] if this entry
    /// was successfully parsed, or `None` for raw/fallback entries.
    pub fn as_typed_mut(&mut self) -> Option<&mut OAuthClient> {
        match self {
            OAuthClientEntry::Typed(c) => Some(c),
            OAuthClientEntry::Raw { .. } => None,
        }
    }
}

/// OAuth client configuration for a single provider.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct OAuthClient {
    pub id: String,
    pub secret: StringOrSecretRef,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub display_name: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub allowed_domains: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub connect_config: Option<OAuthConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub login_config: Option<OAuthConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tenant: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub share_with_workspaces: Option<bool>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub grant_types: Vec<String>,

    /// Catch-all for provider-specific fields (e.g. Keycloak `org`, Okta `domain`/`custom`).
    #[serde(flatten)]
    pub extra: BTreeMap<String, serde_json::Value>,
}

/// OAuth provider endpoint configuration.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct OAuthConfig {
    pub auth_url: String,
    pub token_url: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub userinfo_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scopes: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub extra_params: Option<BTreeMap<String, String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub extra_params_callback: Option<BTreeMap<String, String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub req_body_auth: Option<bool>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub grant_types: Vec<String>,
    /// Optional URL overrides for the provider's sandbox environment.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sandbox: Option<OAuthSandboxOverride>,
}

/// URL overrides for an OAuth provider's sandbox environment.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct OAuthSandboxOverride {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub auth_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub token_url: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub userinfo_url: Option<String>,
}

// ---------------------------------------------------------------------------
// OpenTelemetry
// ---------------------------------------------------------------------------

/// OpenTelemetry exporter configuration.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct OtelSettings {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub metrics_enabled: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub logs_enabled: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tracing_enabled: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub otel_exporter_otlp_endpoint: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub otel_exporter_otlp_headers: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub otel_exporter_otlp_protocol: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub otel_exporter_otlp_compression: Option<String>,
}

/// Per-language HTTP request tracing proxy configuration.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct OtelTracingProxySettings {
    #[serde(default)]
    pub enabled: bool,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub enabled_languages: Vec<ScriptLang>,
    /// Comma-separated list of host patterns injected as NO_PROXY into jobs so their HTTP
    /// clients bypass the local MITM tracing proxy. Independent of the worker's own
    /// NO_PROXY env (which governs the proxy's upstream relay). Use this for clients that
    /// pin their own CA (kubectl, helm, terraform providers, aws cli for EKS, etc.).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub no_proxy_hosts: Option<String>,
    /// Comma-separated host/IP patterns for which the MITM proxy skips upstream TLS
    /// verification. Unlike `no_proxy_hosts` the hosts stay traced — only the proxy's own
    /// upstream certificate check is disabled.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub insecure_upstream_hosts: Option<String>,
    /// Extra CA certificates (PEM bundle) added to the MITM proxy's upstream trust store,
    /// on top of the system roots, so internal endpoints signed by a private CA verify.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub upstream_ca_certs: Option<String>,
}

/// Script language identifier (for instance config use).
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "lowercase")]
pub enum ScriptLang {
    Python3,
    Deno,
    Go,
    Bash,
    Powershell,
    Postgresql,
    Bun,
    Bunnative,
    Mysql,
    Bigquery,
    Snowflake,
    Graphql,
    Nativets,
    Mssql,
    #[serde(rename = "oracledb")]
    OracleDB,
    #[serde(rename = "duckdb")]
    DuckDb,
    Php,
    Rust,
    Ansible,
    #[serde(rename = "csharp")]
    CSharp,
    Nu,
    Java,
    Ruby,
    Rlang,
}

// ---------------------------------------------------------------------------
// Critical error channels
// ---------------------------------------------------------------------------

/// A channel for delivering critical error alerts.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
#[serde(untagged)]
pub enum CriticalErrorChannel {
    Email { email: String },
    Slack { slack_channel: String },
    Teams { teams_channel: TeamsChannel },
}

/// Microsoft Teams channel reference.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct TeamsChannel {
    pub team_id: String,
    pub team_name: String,
    pub channel_id: String,
    pub channel_name: String,
}

// ---------------------------------------------------------------------------
// DB oversize alert
// ---------------------------------------------------------------------------

/// Configuration for critical alerts when the database exceeds a size threshold.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct DbOversizeAlert {
    #[serde(default)]
    pub enabled: bool,
    #[serde(default)]
    pub value: f32,
}

// ---------------------------------------------------------------------------
// DuckLake
// ---------------------------------------------------------------------------

/// DuckLake catalog database settings.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct DucklakeSettings {
    pub ducklakes: BTreeMap<String, Ducklake>,
}

/// A single DuckLake instance configuration.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct Ducklake {
    pub catalog: DucklakeCatalog,
    pub storage: DucklakeStorage,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub extra_args: Option<String>,
}

/// DuckLake catalog backend reference.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct DucklakeCatalog {
    pub resource_type: DucklakeCatalogResourceType,
    pub resource_path: String,
}

/// DuckLake storage location.
#[derive(Deserialize, Serialize, Clone, Debug)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct DucklakeStorage {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub storage: Option<String>,
    pub path: String,
}

/// The type of database backing a DuckLake catalog.
#[derive(Deserialize, Serialize, Clone, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
#[serde(rename_all = "lowercase")]
pub enum DucklakeCatalogResourceType {
    Postgresql,
    Mysql,
    Instance,
}

// ---------------------------------------------------------------------------
// Custom instance PG databases
// ---------------------------------------------------------------------------

/// Custom PostgreSQL databases managed by the instance. `user_pwd` is operator-configurable
/// (resolved from a Kubernetes secretKeyRef by the EE operator); `databases` is runtime
/// setup status. The replication-role password lives in a separate hidden setting
/// (`custom_instance_replication_pwd`), never in this operator-facing config row.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct CustomInstancePgDatabases {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub user_pwd: Option<StringOrSecretRef>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub databases: BTreeMap<String, CustomInstanceDb>,
}

/// Status of a single custom instance database.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct CustomInstanceDb {
    #[serde(default)]
    pub logs: CustomInstanceDbLogs,
    #[serde(default)]
    pub success: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tag: Option<String>,
    /// The workspace a member created this fork copy for. Absent when a superadmin created it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub workspace_id: Option<String>,
}

/// Setup log entries for a custom instance database.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
#[serde(default)]
pub struct CustomInstanceDbLogs {
    pub super_admin: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub database_credentials: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub valid_dbname: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub created_database: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub db_connect: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub grant_permissions: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub replication_user: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub replication_user_error: Option<String>,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub user_connect: String,
}

// ---------------------------------------------------------------------------
// External instance PG cluster
// ---------------------------------------------------------------------------

/// The external Postgres cluster Windmill manages for `external_instance` data tables and Ducklake
/// catalogs. `user` logs in as the cluster's administrator: it needs `CREATEDB` and `CREATEROLE`.
/// `dbname` is only where that login connects to run cluster-wide statements.
///
/// Every field defaults rather than being required: this deserializes as part of the whole
/// instance config, and one malformed row must not make every other setting unreadable. The
/// write path and every use reject an incomplete value instead.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct ExternalInstancePg {
    #[serde(default)]
    pub host: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub port: Option<u16>,
    #[serde(default)]
    pub user: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub password: Option<StringOrSecretRef>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dbname: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub sslmode: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub root_certificate_pem: Option<String>,
}

// ---------------------------------------------------------------------------
// Autoscaling (worker config)
// ---------------------------------------------------------------------------

/// Worker group autoscaling configuration.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct AutoscalingConfig {
    #[serde(default)]
    pub enabled: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub min_workers: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_workers: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cooldown_seconds: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inc_scale_num_jobs_waiting: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub full_scale_cooldown_seconds: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub full_scale_jobs_waiting: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dec_scale_occupancy_rate: Option<u8>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inc_scale_occupancy_rate: Option<u8>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub inc_num_workers: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub custom_tags: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub integration: Option<AutoscalingIntegration>,
}

/// Autoscaling integration backend.
///
/// The `type` field selects the backend: `"script"`, `"dryrun"`, or `"kubernetes"`.
/// For `"script"`, `path` is required and `tag` is optional.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct AutoscalingIntegration {
    #[serde(rename = "type")]
    pub integration_type: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tag: Option<String>,
}

// ---------------------------------------------------------------------------
// Worker group config
// ---------------------------------------------------------------------------

/// Worker group configuration.
#[derive(Deserialize, Serialize, Clone, Debug, Default)]
#[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
pub struct WorkerGroupConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worker_tags: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub priority_tags: Option<BTreeMap<String, u8>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dedicated_worker: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub dedicated_workers: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub init_bash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub periodic_script_bash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub periodic_script_interval_seconds: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub additional_python_paths: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub pip_local_dependencies: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub env_vars_static: Option<BTreeMap<String, String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub env_vars_allowlist: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub min_alive_workers_alert_threshold: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub autoscaling: Option<AutoscalingConfig>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub native_mode: Option<bool>,
    /// Workers of the group pull no new job while this is set; queued jobs wait.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub paused: Option<bool>,
    /// Object store this group's dependency cache uses instead of the instance one. Same shape
    /// as the instance `object_store_cache_config`.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(
        feature = "instance_config_schema",
        schemars(schema_with = "opaque_json_schema")
    )]
    pub object_store_cache_config: Option<serde_json::Value>,

    /// Catch-all for fields not yet covered by typed fields.
    #[serde(flatten)]
    pub extra: BTreeMap<String, serde_json::Value>,
}

// ---------------------------------------------------------------------------
// Diff + apply logic
// ---------------------------------------------------------------------------

/// Controls whether absent keys are deleted.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ApplyMode {
    /// UI: only upsert changed keys, never delete absent ones.
    Merge,
    /// Operator: upsert changed + delete absent (except protected).
    Replace,
}

/// The diff result for global settings.
#[derive(Debug, Default)]
pub struct SettingsDiff {
    pub upserts: BTreeMap<String, serde_json::Value>,
    pub deletes: Vec<String>,
    /// Previous values for keys being updated or deleted (for logging).
    pub previous_values: BTreeMap<String, serde_json::Value>,
    /// Count of desired keys that matched the current value exactly.
    pub unchanged_count: usize,
}

/// The diff result for worker configs.
#[derive(Debug, Default)]
pub struct ConfigsDiff {
    pub upserts: BTreeMap<String, serde_json::Value>,
    pub deletes: Vec<String>,
}

/// Settings that must never be deleted by the operator or bulk API.
pub const PROTECTED_SETTINGS: &[&str] = &[
    "ducklake_user_pg_pwd",
    "ducklake_settings",
    "custom_instance_pg_databases",
    "custom_instance_replication_pwd",
    "external_instance_pg_state",
    "uid",
    "rsa_keys",
    "jwt_secret",
    "min_keep_alive_version",
];

/// Secrets the server signs with or generated for itself. `jwt_secret` signs API and job
/// tokens and `rsa_keys` signs job OIDC tokens, so reading either is enough to mint tokens
/// for any user. Superadmin reads return them only while `server_secrets_exported()`.
/// Withholding them from reads does not make them unwritable: the two signing keys are not
/// in `HIDDEN_SETTINGS`, so config can still set them (an operator ConfigMap may set
/// `jwt_secret`).
pub const SERVER_SECRET_SETTINGS: &[&str] = &[
    "jwt_secret",
    "rsa_keys",
    "custom_instance_replication_pwd",
    "external_instance_pg_state",
];

/// Whether superadmin API reads (settings list, single-setting read, config export) return
/// `SERVER_SECRET_SETTINGS`. On by default so `wmill instance get-config` and `pull` carry a
/// full migration; `EXPORT_SERVER_SECRETS=false` withholds them from every API response, so a
/// leaked superadmin token cannot take the signing keys. Fails closed: once set to a
/// non-empty value, only `true`/`1`/`yes`/`on` keep exporting, so `0` or a typo withholds.
/// Read per call: the flag is cheap and tests flip it in-process.
pub fn server_secrets_exported() -> bool {
    match std::env::var("EXPORT_SERVER_SECRETS") {
        Ok(v) if !v.trim().is_empty() => matches!(
            v.trim().to_ascii_lowercase().as_str(),
            "true" | "1" | "yes" | "on"
        ),
        _ => true,
    }
}

pub fn is_withheld_server_secret(name: &str) -> bool {
    SERVER_SECRET_SETTINGS.contains(&name) && !server_secrets_exported()
}

/// Internal settings that are never exposed via the API or included in config exports.
pub const HIDDEN_SETTINGS: &[&str] = &[
    "uid",
    "min_keep_alive_version",
    "automate_username_creation",
    "_restart_coordination",
    // Legacy ghost: worker configs live in the `config` table with a
    // `worker__` prefix, not as a single blob in `global_settings`. Older
    // Windmill versions stored them here and the row would be resurrected on
    // every bulk InstanceSettings save via `GlobalSettings::extra`. Hiding it
    // on read + rejecting it in `diff_global_settings` breaks that loop.
    "worker_configs",
    // Auto-generated password for the REPLICATION role used by postgres triggers.
    // Server-only (written by setup/refresh via direct SQL), never operator-authored —
    // hidden so the config machinery can't read, rewrite, or drop it.
    "custom_instance_replication_pwd",
    // Same for the passwords and database registry Windmill keeps for the external cluster.
    "external_instance_pg_state",
];

/// Top-level settings whose entire value is sensitive and must be fully redacted in logs.
const SENSITIVE_SETTINGS: &[&str] = &[
    "jwt_secret",
    "rsa_keys",
    "scim_token",
    "hub_api_secret",
    "license_key",
    "ducklake_user_pg_pwd",
    "custom_instance_replication_pwd",
    "external_instance_pg_state",
    "pip_index_url",
    "pip_extra_index_url",
    "npm_config_registry",
    "bunfig_install_scopes",
    "npmrc",
    "maven_repos",
    "ruby_repos",
    "powershell_repo_pat",
    "workspace_registries",
    "sandbox_registry_auth",
];

/// Object-valued settings that contain sensitive sub-fields.
/// Maps a top-level key to the sub-field names that must be redacted.
const NESTED_SENSITIVE_FIELDS: &[(&str, &[&str])] = &[
    ("smtp_settings", &["smtp_password"]),
    (
        "secret_backend",
        &["token", "client_secret", "secret_access_key"],
    ),
    (
        "object_store_cache_config",
        &["secret_key", "serviceAccountKey", "accessKey"],
    ),
    ("custom_instance_pg_databases", &["user_pwd"]),
    ("external_instance_pg", &["password"]),
    ("github_enterprise_app", &["private_key"]),
];

fn redact_json_value(value: &serde_json::Value) -> serde_json::Value {
    match value {
        serde_json::Value::String(s) => serde_json::json!(redact_string(s)),
        _ => serde_json::json!("[redacted]"),
    }
}

fn mask_nested_sensitive(key: &str, value: &serde_json::Value) -> serde_json::Value {
    for &(parent_key, sub_fields) in NESTED_SENSITIVE_FIELDS {
        if key == parent_key {
            if let serde_json::Value::Object(map) = value {
                let mut masked = map.clone();
                for &field in sub_fields {
                    if let Some(v) = masked.get(field) {
                        masked.insert(field.to_string(), redact_json_value(v));
                    }
                }
                return serde_json::Value::Object(masked);
            }
        }
    }
    // Settings that are maps-of-objects where each child has a sensitive sub-field.
    const NESTED_MAP_SENSITIVE: &[(&str, &str)] = &[("oauths", "secret")];
    for &(parent_key, child_field) in NESTED_MAP_SENSITIVE {
        if key == parent_key {
            if let serde_json::Value::Object(entries) = value {
                let mut masked = entries.clone();
                for (_entry_key, entry_val) in masked.iter_mut() {
                    if let serde_json::Value::Object(ref mut obj) = entry_val {
                        if let Some(v) = obj.get(child_field) {
                            obj.insert(child_field.to_string(), redact_json_value(v));
                        }
                    }
                }
                return serde_json::Value::Object(masked);
            }
        }
    }
    value.clone()
}

fn redact_string(s: &str) -> String {
    let char_count = s.chars().count();
    if char_count <= 6 {
        "****".to_string()
    } else {
        let show = (char_count / 4).min(4);
        let prefix: String = s.chars().take(show).collect();
        let suffix: String = s.chars().skip(char_count - show).collect();
        format!("{prefix}****{suffix}")
    }
}

pub fn format_setting_value(key: &str, value: &serde_json::Value) -> String {
    if SENSITIVE_SETTINGS.contains(&key) {
        return match value {
            serde_json::Value::String(s) => format!("\"{}\"", redact_string(s)),
            _ => "[redacted]".to_string(),
        };
    }
    let value = mask_nested_sensitive(key, value);
    crate::utils::truncate_with_ellipsis(&value.to_string(), 197)
}

/// Extract the expiry timestamp from a license key JSON value.
///
/// License keys have the format `<client_id>.<expiry>.<signature>`.
/// Returns `None` if the value is not a string or doesn't match the format.
fn license_key_expiry(value: &serde_json::Value) -> Option<u64> {
    let s = value.as_str()?;
    let parts: Vec<&str> = s.split('.').collect();
    if parts.len() != 3 {
        return None;
    }
    parts[1].parse::<u64>().ok()
}

/// Returns true if two license key values share the same client ID
/// (i.e. they belong to the same customer, possibly with different expiry/signature).
fn license_keys_same_client(a: &serde_json::Value, b: &serde_json::Value) -> bool {
    let (Some(a_str), Some(b_str)) = (a.as_str(), b.as_str()) else {
        return false;
    };
    let a_parts: Vec<&str> = a_str.split('.').collect();
    let b_parts: Vec<&str> = b_str.split('.').collect();
    if a_parts.len() != 3 || b_parts.len() != 3 {
        return false;
    }
    a_parts[0] == b_parts[0]
}

fn is_empty_or_null(value: &serde_json::Value) -> bool {
    value.is_null() || value.as_str().map_or(false, |s| s.trim().is_empty())
}

/// Maximum retention period in seconds for CE builds (30 days).
pub const CE_MAX_RETENTION_PERIOD_SECS: i64 = 30 * 24 * 3600;

/// Clamp `retention_period_secs` to `CE_MAX_RETENTION_PERIOD_SECS` on CE builds.
/// Returns the (possibly clamped) value. On EE builds this is a no-op.
pub fn clamp_retention_period(value: serde_json::Value) -> serde_json::Value {
    #[cfg(feature = "enterprise")]
    {
        value
    }
    #[cfg(not(feature = "enterprise"))]
    {
        if let Some(secs) = value.as_i64() {
            if secs > CE_MAX_RETENTION_PERIOD_SECS {
                tracing::warn!(
                    "Clamping retention_period_secs from {} to {} (CE max: 30 days)",
                    secs,
                    CE_MAX_RETENTION_PERIOD_SECS
                );
                return serde_json::json!(CE_MAX_RETENTION_PERIOD_SECS);
            }
        }
        value
    }
}

/// Compute the diff between current and desired global settings.
pub fn diff_global_settings(
    current: &BTreeMap<String, serde_json::Value>,
    desired: &BTreeMap<String, serde_json::Value>,
    mode: ApplyMode,
) -> SettingsDiff {
    let mut upserts = BTreeMap::new();
    let mut deletes = Vec::new();
    let mut previous_values = BTreeMap::new();
    let mut unchanged_count: usize = 0;
    for (key, desired_value) in desired {
        // Hidden settings are server-managed and never driven by config: they are
        // filtered out on read (`from_db`) and must be ignored on write too, so a client
        // PUT that flattened one into `GlobalSettings::extra` can't resurrect or clobber
        // the row (e.g. `worker_configs`, or the custom-instance credentials/status).
        if HIDDEN_SETTINGS.contains(&key.as_str()) {
            tracing::warn!(
                "Ignoring hidden setting '{key}' in global_settings diff (server-managed, not configurable)"
            );
            continue;
        }
        if PROTECTED_SETTINGS.contains(&key.as_str())
            && is_empty_or_null(desired_value)
            && current.contains_key(key)
        {
            tracing::warn!(
                "Skipping {key} update: protected setting cannot be overwritten with empty/null value"
            );
            continue;
        }
        // An empty string or explicit null means "unset": route it to deletes
        // so clearing a setting via the bulk endpoint has the same effect as
        // clearing it via the per-key endpoint (`set_global_setting_internal`
        // already treats both `""` and `null` as deletes).
        // Without this, the InstanceSettings YAML editor cannot delete keys
        // (Merge mode silently preserves missing/null values), and stale `""`
        // rows linger and trip the EE registry gate, producing spurious
        // "requires Enterprise" warnings on every CE job.
        if is_empty_or_null(desired_value) {
            if let Some(existing) = current.get(key) {
                previous_values.insert(key.clone(), existing.clone());
                deletes.push(key.clone());
            } else {
                unchanged_count += 1;
            }
            continue;
        }
        let mut value = if key == "retention_period_secs" {
            clamp_retention_period(desired_value.clone())
        } else {
            desired_value.clone()
        };
        // Preserve the runtime-only `databases` sub-field inside
        // `custom_instance_pg_databases` so that config sync never wipes
        // setup status/logs that are managed by the setup endpoint.
        if key == "custom_instance_pg_databases" {
            if let Some(existing) = current.get(key) {
                if let Some(databases) = existing.get("databases") {
                    if let Some(obj) = value.as_object_mut() {
                        obj.entry("databases").or_insert_with(|| databases.clone());
                    }
                }
            }
        }
        match current.get(key) {
            Some(existing) if *existing == value => {
                unchanged_count += 1;
            }
            Some(existing) if key == LICENSE_KEY_SETTING => {
                if license_keys_same_client(existing, &value) {
                    let current_expiry = license_key_expiry(existing).unwrap_or(0);
                    let desired_expiry = license_key_expiry(&value).unwrap_or(0);
                    if desired_expiry > current_expiry {
                        previous_values.insert(key.clone(), existing.clone());
                        upserts.insert(key.clone(), value);
                    } else {
                        tracing::info!(
                            "Skipping license_key update: desired expiry ({}) is not posterior to current expiry ({})",
                            desired_expiry,
                            current_expiry
                        );
                        unchanged_count += 1;
                    }
                } else {
                    previous_values.insert(key.clone(), existing.clone());
                    upserts.insert(key.clone(), value);
                }
            }
            Some(existing) => {
                previous_values.insert(key.clone(), existing.clone());
                upserts.insert(key.clone(), value);
            }
            None => {
                upserts.insert(key.clone(), value);
            }
        }
    }
    if matches!(mode, ApplyMode::Replace) {
        for key in current.keys() {
            if !desired.contains_key(key)
                && !PROTECTED_SETTINGS.contains(&key.as_str())
                && !HIDDEN_SETTINGS.contains(&key.as_str())
            {
                previous_values.insert(key.clone(), current[key].clone());
                deletes.push(key.clone());
            }
        }
    }
    SettingsDiff { upserts, deletes, previous_values, unchanged_count }
}

/// Compute the diff between current and desired worker configs.
pub fn diff_worker_configs(
    current: &BTreeMap<String, serde_json::Value>,
    desired: &BTreeMap<String, serde_json::Value>,
    mode: ApplyMode,
) -> ConfigsDiff {
    let mut upserts = BTreeMap::new();
    for (key, value) in desired {
        match current.get(key) {
            Some(existing) if existing == value => {} // no change
            _ => {
                upserts.insert(key.clone(), value.clone());
            }
        }
    }
    let mut deletes = Vec::new();
    if matches!(mode, ApplyMode::Replace) {
        for key in current.keys() {
            if !desired.contains_key(key) {
                deletes.push(key.clone());
            }
        }
    }
    ConfigsDiff { upserts, deletes }
}

/// Declaratively replace the global settings, rejecting a
/// `github_app_webhook_base_url` or `http_route_default_allowed_origins` the
/// API would reject.
///
/// Every declarative writer (the `sync-config` CLI, the Kubernetes operator's
/// ConfigMap sync) MUST go through this rather than calling
/// [`apply_settings_diff`] directly, or that validation is silently skipped.
/// Note this is *not* the full equivalent of the HTTP layer's pre-write hook —
/// that also enforces rules for `automate_username_creation`,
/// `critical_alert_mute_ui` and `app_workspaced_route`, which the declarative
/// paths have never applied. A new rule added there does not appear here for free.
///
/// Registered webhooks are deliberately not re-pointed here. The receiver is set at
/// instance setup, before any GitHub App is connected, so there is normally nothing
/// registered to move; an instance that changes it later finds the affected
/// workspaces listed in instance settings and re-saves them.
///
/// AUTHORIZATION: replaces instance-wide settings and takes no authed context, so
/// callers MUST have established superadmin or equivalent system authority (the CLI
/// and the operator both run with direct instance credentials).
pub async fn sync_global_settings_declarative(
    db: &sqlx::Pool<sqlx::Postgres>,
    current: &BTreeMap<String, serde_json::Value>,
    desired: &BTreeMap<String, serde_json::Value>,
) -> anyhow::Result<()> {
    let webhook_key = crate::global_settings::GITHUB_APP_WEBHOOK_BASE_URL_SETTING;
    // Non-string shapes are rejected rather than ignored: `as_str()` alone would let a
    // bool/number/object through as if the key were absent, and the diff below would
    // then persist it — where the HTTP path answers "must be a URL string".
    match desired.get(webhook_key) {
        None | Some(serde_json::Value::Null) => {}
        Some(serde_json::Value::String(s)) if s.trim().is_empty() => {}
        Some(serde_json::Value::String(s)) => crate::global_settings::validate_webhook_base_url(s)
            .map_err(|e| anyhow::anyhow!("{webhook_key}: {e}"))?,
        // Names the JSON kind rather than printing it: this is the last message on
        // this path that could report submitted content, and an object or array could
        // carry a secret into `sync-config` output and operator logs.
        Some(other) => {
            let kind = match other {
                serde_json::Value::Bool(_) => "a boolean",
                serde_json::Value::Number(_) => "a number",
                serde_json::Value::Array(_) => "an array",
                serde_json::Value::Object(_) => "an object",
                _ => "a non-string value",
            };
            return Err(anyhow::anyhow!(
                "{webhook_key} must be a URL string, got {kind}"
            ));
        }
    }

    let banner_key = crate::global_settings::INSTANCE_BANNER_SETTING;
    match desired.get(banner_key) {
        None | Some(serde_json::Value::Null) => {}
        Some(serde_json::Value::String(s)) if s.trim().is_empty() => {}
        Some(banner) => crate::global_settings::validate_instance_banner(banner)
            // The validator's messages name the offending field and its expected type,
            // never the submitted value, so they are safe to surface here.
            .map_err(|e| anyhow::anyhow!("{banner_key}: {e}"))?,
    }

    let accent_key = crate::global_settings::ACCENT_COLOR_SETTING;
    match desired.get(accent_key) {
        None | Some(serde_json::Value::Null) => {}
        Some(serde_json::Value::String(s)) if s.trim().is_empty() => {}
        Some(color) => crate::global_settings::validate_accent_color(color)
            .map_err(|e| anyhow::anyhow!("{accent_key}: {e}"))?,
    }

    // An origin list that cannot be parsed is dropped at boot, leaving the
    // empty default — which is no restriction at all. Rejecting it here is what
    // keeps a typo in a ConfigMap from silently widening CORS instance-wide.
    let origins_key = crate::global_settings::HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING;
    crate::global_settings::parse_allowed_origins_setting(desired.get(origins_key))
        .map_err(|e| anyhow::anyhow!("{origins_key}: {e}"))?;

    let max_expiration_key = crate::global_settings::MAX_TOKEN_EXPIRATION_DAYS_SETTING;
    crate::global_settings::parse_max_token_expiration_days(desired.get(max_expiration_key))
        .map_err(|e| anyhow::anyhow!("{max_expiration_key}: {e}"))?;

    let mut diff = diff_global_settings(current, desired, ApplyMode::Replace);
    crate::external_instance_pg::write_external_instance_pg_from_diff(db, &mut diff).await?;
    apply_settings_diff(db, &diff).await?;

    Ok(())
}

/// Apply a settings diff to the global_settings table.
pub async fn apply_settings_diff(
    db: &sqlx::Pool<sqlx::Postgres>,
    diff: &SettingsDiff,
) -> anyhow::Result<()> {
    for (key, value) in &diff.upserts {
        sqlx::query(
            "INSERT INTO global_settings (name, value) VALUES ($1, $2) \
             ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value, updated_at = now()",
        )
        .bind(key)
        .bind(value)
        .execute(db)
        .await?;
        if let Some(old_value) = diff.previous_values.get(key) {
            tracing::info!(
                "Updated global setting: {key} ({} -> {})",
                format_setting_value(key, old_value),
                format_setting_value(key, value)
            );
        } else {
            tracing::info!(
                "Created global setting: {key} (value: {})",
                format_setting_value(key, value)
            );
        }
    }
    for key in &diff.deletes {
        sqlx::query("DELETE FROM global_settings WHERE name = $1")
            .bind(key)
            .execute(db)
            .await?;
        if let Some(old_value) = diff.previous_values.get(key) {
            tracing::info!(
                "Deleted global setting: {key} (was: {})",
                format_setting_value(key, old_value)
            );
        } else {
            tracing::info!("Deleted global setting: {key}");
        }
    }
    if diff.unchanged_count > 0 {
        tracing::info!("{} global setting(s) unchanged", diff.unchanged_count);
    }
    Ok(())
}

/// Apply a configs diff to the config table (worker__ prefix).
pub async fn apply_configs_diff(
    db: &sqlx::Pool<sqlx::Postgres>,
    diff: &ConfigsDiff,
) -> anyhow::Result<()> {
    for (group_name, config_value) in &diff.upserts {
        let db_key = format!("worker__{group_name}");
        sqlx::query(
            "INSERT INTO config (name, config) VALUES ($1, $2) \
             ON CONFLICT (name) DO UPDATE SET config = EXCLUDED.config",
        )
        .bind(&db_key)
        .bind(config_value)
        .execute(db)
        .await?;
        tracing::info!("Synced worker config: {db_key}");
    }
    for group_name in &diff.deletes {
        let db_key = format!("worker__{group_name}");
        sqlx::query("DELETE FROM config WHERE name = $1")
            .bind(&db_key)
            .execute(db)
            .await?;
        tracing::info!("Deleted worker config: {db_key}");
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Env ref resolution
// ---------------------------------------------------------------------------

fn resolve_env_field(field: &mut StringOrSecretRef) -> Result<(), String> {
    if let Some(var_name) = field.as_env_ref() {
        let var_name = var_name.to_string();
        let value = std::env::var(&var_name).map_err(|_| var_name)?;
        *field = StringOrSecretRef::Literal(value);
    }
    Ok(())
}

fn resolve_env_option(field: &mut Option<StringOrSecretRef>) -> Result<(), String> {
    if let Some(val) = field {
        resolve_env_field(val)?;
    }
    Ok(())
}

/// Resolve all `EnvRef` fields in `GlobalSettings` by reading the
/// corresponding environment variables. Each resolved `EnvRef` is replaced
/// with a `Literal`. Returns `Err(var_name)` if a referenced env var is not set.
pub fn resolve_env_refs(settings: &mut GlobalSettings) -> Result<(), String> {
    resolve_env_option(&mut settings.license_key)?;
    resolve_env_option(&mut settings.hub_api_secret)?;
    resolve_env_option(&mut settings.scim_token)?;

    if let Some(smtp) = &mut settings.smtp_settings {
        resolve_env_option(&mut smtp.smtp_password)?;
    }

    if let Some(oauths) = &mut settings.oauths {
        for (name, entry) in oauths.iter_mut() {
            match entry {
                OAuthClientEntry::Typed(oauth) => {
                    resolve_env_field(&mut oauth.secret)?;
                }
                OAuthClientEntry::Raw { error, .. } => {
                    tracing::error!(
                        "OAuth entry '{name}' could not be deserialized as OAuthClient \
                         and was kept as raw JSON — it likely has an unexpected shape: {error}"
                    );
                }
            }
        }
    }

    if let Some(pg) = &mut settings.custom_instance_pg_databases {
        resolve_env_option(&mut pg.user_pwd)?;
    }

    if let Some(pg) = &mut settings.external_instance_pg {
        resolve_env_option(&mut pg.password)?;
    }

    Ok(())
}

// ---------------------------------------------------------------------------
// Read from DB
// ---------------------------------------------------------------------------

impl InstanceConfig {
    /// Read the full instance configuration from the database.
    pub async fn from_db(db: &sqlx::Pool<sqlx::Postgres>) -> anyhow::Result<Self> {
        Self::read_from_db(db, true).await
    }

    /// The instance configuration as an API response may carry it: `from_db` without
    /// `SERVER_SECRET_SETTINGS` unless `server_secrets_exported()`. It holds admin-entered
    /// credentials (`license_key`, `scim_token`, SMTP and OAuth secrets), so callers must
    /// have checked superadmin.
    pub async fn from_db_for_api(db: &sqlx::Pool<sqlx::Postgres>) -> anyhow::Result<Self> {
        Self::read_from_db(db, server_secrets_exported()).await
    }

    async fn read_from_db(
        db: &sqlx::Pool<sqlx::Postgres>,
        include_server_secrets: bool,
    ) -> anyhow::Result<Self> {
        // Read global_settings table → flat map → deserialize into GlobalSettings
        let rows: Vec<(String, serde_json::Value)> =
            sqlx::query_as("SELECT name, value FROM global_settings")
                .fetch_all(db)
                .await?;
        let map: serde_json::Map<String, serde_json::Value> = rows
            .into_iter()
            .filter(|(name, _)| !HIDDEN_SETTINGS.contains(&name.as_str()))
            .filter(|(name, _)| {
                include_server_secrets || !SERVER_SECRET_SETTINGS.contains(&name.as_str())
            })
            .collect();
        let global_settings: GlobalSettings =
            serde_json::from_value(serde_json::Value::Object(map))?;

        // Read config table (worker__ prefix) → BTreeMap<String, WorkerGroupConfig>
        let config_rows: Vec<(String, serde_json::Value)> =
            sqlx::query_as("SELECT name, config FROM config WHERE name LIKE 'worker__%'")
                .fetch_all(db)
                .await?;
        let worker_configs = config_rows
            .into_iter()
            .map(|(name, config)| {
                let group = name.strip_prefix("worker__").unwrap_or(&name).to_string();
                let wc: WorkerGroupConfig = serde_json::from_value(config).unwrap_or_default();
                (group, wc)
            })
            .collect();

        Ok(Self { global_settings, worker_configs })
    }

    /// Sync this config to the database using `ApplyMode::Replace`.
    ///
    /// Reads the current state, computes a diff, then applies upserts and
    /// deletes for both global settings and worker configs.
    pub async fn sync_to_db(&self, db: &sqlx::Pool<sqlx::Postgres>) -> anyhow::Result<()> {
        let desired_settings = self.global_settings.to_settings_map();

        // Current global settings
        let rows: Vec<(String, serde_json::Value)> =
            sqlx::query_as("SELECT name, value FROM global_settings")
                .fetch_all(db)
                .await?;
        let current_settings: BTreeMap<String, serde_json::Value> = rows.into_iter().collect();

        sync_global_settings_declarative(db, &current_settings, &desired_settings).await?;

        // Worker configs
        let desired_configs: BTreeMap<String, serde_json::Value> = self
            .worker_configs
            .iter()
            .map(|(k, v)| {
                (
                    k.clone(),
                    serde_json::to_value(v).expect("WorkerGroupConfig serialization cannot fail"),
                )
            })
            .collect();

        let config_rows: Vec<(String, serde_json::Value)> =
            sqlx::query_as("SELECT name, config FROM config WHERE name LIKE 'worker__%'")
                .fetch_all(db)
                .await?;
        let current_configs: BTreeMap<String, serde_json::Value> = config_rows
            .into_iter()
            .map(|(name, config)| {
                let group = name.strip_prefix("worker__").unwrap_or(&name).to_string();
                (group, config)
            })
            .collect();

        let configs_diff =
            diff_worker_configs(&current_configs, &desired_configs, ApplyMode::Replace);
        apply_configs_diff(db, &configs_diff).await?;

        Ok(())
    }

    /// Serialize to YAML with sorted global_settings keys and worker_configs
    /// ordered with "default" and "native" first.
    pub fn to_sorted_yaml(&self) -> Result<String, String> {
        let settings_map = self.global_settings.to_settings_map();

        let mut yaml = String::from("global_settings:\n");
        for (key, value) in &settings_map {
            let value_yaml = serde_yml::to_string(value)
                .map_err(|e| format!("YAML serialization failed for {key}: {e}"))?;
            write_yaml_field(&mut yaml, key, value, &value_yaml, 1);
        }

        if !self.worker_configs.is_empty() {
            yaml.push_str("worker_configs:\n");
            let priority_keys = ["default", "native"];
            for &pk in &priority_keys {
                if let Some(wc) = self.worker_configs.get(pk) {
                    let wc_value = serde_json::to_value(wc)
                        .map_err(|e| format!("JSON serialization failed for {pk}: {e}"))?;
                    let wc_yaml = serde_yml::to_string(&wc_value)
                        .map_err(|e| format!("YAML serialization failed for {pk}: {e}"))?;
                    write_yaml_field(&mut yaml, pk, &wc_value, &wc_yaml, 1);
                }
            }
            for (key, wc) in &self.worker_configs {
                if priority_keys.contains(&key.as_str()) {
                    continue;
                }
                let wc_value = serde_json::to_value(wc)
                    .map_err(|e| format!("JSON serialization failed for {key}: {e}"))?;
                let wc_yaml = serde_yml::to_string(&wc_value)
                    .map_err(|e| format!("YAML serialization failed for {key}: {e}"))?;
                write_yaml_field(&mut yaml, key, &wc_value, &wc_yaml, 1);
            }
        }

        Ok(yaml)
    }
}

fn write_yaml_field(
    yaml: &mut String,
    key: &str,
    value: &serde_json::Value,
    value_yaml: &str,
    indent: usize,
) {
    use std::fmt::Write;

    let prefix = "  ".repeat(indent);
    let trimmed = value_yaml.trim();
    let is_nested = match value {
        serde_json::Value::Object(m) => !m.is_empty(),
        serde_json::Value::Array(a) => !a.is_empty(),
        _ => false,
    };
    if is_nested {
        let _ = writeln!(yaml, "{prefix}{key}:");
        let inner_prefix = "  ".repeat(indent + 1);
        for line in trimmed.lines() {
            let _ = writeln!(yaml, "{inner_prefix}{line}");
        }
    } else {
        let _ = writeln!(yaml, "{prefix}{key}: {trimmed}");
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn global_settings_to_map() {
        let settings = GlobalSettings {
            base_url: Some("https://example.com".to_string()),
            retention_period_secs: Some(86400),
            expose_metrics: Some(true),
            smtp_settings: Some(SmtpSettings {
                smtp_host: Some("mail.example.com".to_string()),
                smtp_port: Some(587),
                ..Default::default()
            }),
            ..Default::default()
        };
        let map = settings.to_settings_map();
        assert_eq!(map["base_url"], serde_json::json!("https://example.com"));
        assert_eq!(map["retention_period_secs"], serde_json::json!(86400));
        assert_eq!(map["expose_metrics"], serde_json::json!(true));
        assert_eq!(
            map["smtp_settings"],
            serde_json::json!({"smtp_host": "mail.example.com", "smtp_port": 587})
        );
        assert!(
            !map.contains_key("license_key"),
            "None fields should be omitted"
        );
    }

    #[test]
    fn global_settings_extra_fields_passthrough() {
        let json = r#"{
            "base_url": "https://example.com",
            "some_future_setting": {"nested": true}
        }"#;
        let settings: GlobalSettings =
            serde_json::from_str(json).expect("Should deserialize with unknown fields");
        assert_eq!(settings.base_url.as_deref(), Some("https://example.com"));
        assert_eq!(
            settings.extra["some_future_setting"],
            serde_json::json!({"nested": true})
        );

        let map = settings.to_settings_map();
        assert!(map.contains_key("some_future_setting"));
    }

    #[test]
    fn worker_config_extra_fields_passthrough() {
        let json = r#"{
            "init_bash": "echo hello",
            "some_future_field": 42
        }"#;
        let config: WorkerGroupConfig =
            serde_json::from_str(json).expect("Should deserialize with unknown fields");
        assert_eq!(config.init_bash.as_deref(), Some("echo hello"));
        assert_eq!(config.extra["some_future_field"], serde_json::json!(42));
    }

    #[test]
    fn instance_config_deserializes_full_example() {
        let json = r##"{
            "global_settings": {
                "base_url": "https://windmill.example.com",
                "license_key": "my-key",
                "retention_period_secs": 2592000,
                "smtp_settings": {"smtp_host": "smtp.example.com", "smtp_port": 587},
                "otel": {
                    "metrics_enabled": true,
                    "otel_exporter_otlp_endpoint": "http://otel:4317"
                },
                "critical_error_channels": [
                    {"email": "admin@example.com"},
                    {"slack_channel": "#alerts"}
                ],
                "custom_tags": ["gpu", "high-mem"]
            },
            "worker_configs": {
                "default": {"init_bash": "echo hello"},
                "gpu": {
                    "dedicated_worker": "ws:f/gpu",
                    "autoscaling": {
                        "enabled": true,
                        "min_workers": 1,
                        "max_workers": 10,
                        "integration": {"type": "kubernetes"}
                    }
                }
            }
        }"##;
        let config: InstanceConfig =
            serde_json::from_str(json).expect("Should deserialize full config");
        assert_eq!(
            config.global_settings.base_url.as_deref(),
            Some("https://windmill.example.com")
        );
        assert_eq!(config.global_settings.retention_period_secs, Some(2592000));
        assert!(config.global_settings.smtp_settings.is_some());
        let smtp = config.global_settings.smtp_settings.as_ref().unwrap();
        assert_eq!(smtp.smtp_host.as_deref(), Some("smtp.example.com"));
        assert_eq!(smtp.smtp_port, Some(587));

        let otel = config.global_settings.otel.as_ref().unwrap();
        assert_eq!(otel.metrics_enabled, Some(true));
        assert_eq!(
            otel.otel_exporter_otlp_endpoint.as_deref(),
            Some("http://otel:4317")
        );

        let channels = config
            .global_settings
            .critical_error_channels
            .as_ref()
            .unwrap();
        assert_eq!(channels.len(), 2);

        let tags = config.global_settings.custom_tags.as_ref().unwrap();
        assert_eq!(tags, &["gpu", "high-mem"]);

        assert_eq!(config.worker_configs.len(), 2);
        assert_eq!(
            config.worker_configs["default"].init_bash.as_deref(),
            Some("echo hello")
        );
        assert_eq!(
            config.worker_configs["gpu"].dedicated_worker.as_deref(),
            Some("ws:f/gpu")
        );
        let autoscaling = config.worker_configs["gpu"].autoscaling.as_ref().unwrap();
        assert!(autoscaling.enabled);
        assert_eq!(autoscaling.min_workers, Some(1));
        assert_eq!(autoscaling.max_workers, Some(10));
        let integration = autoscaling.integration.as_ref().unwrap();
        assert_eq!(integration.integration_type, "kubernetes");
    }

    #[test]
    fn instance_config_roundtrips() {
        let config = InstanceConfig {
            global_settings: GlobalSettings {
                base_url: Some("https://example.com".to_string()),
                expose_metrics: Some(true),
                ..Default::default()
            },
            worker_configs: {
                let mut m = BTreeMap::new();
                m.insert(
                    "default".to_string(),
                    WorkerGroupConfig {
                        init_bash: Some("echo hello".to_string()),
                        ..Default::default()
                    },
                );
                m
            },
        };

        let json = serde_json::to_string(&config).expect("Should serialize");
        let deserialized: InstanceConfig = serde_json::from_str(&json).expect("Should deserialize");
        assert_eq!(
            config.global_settings.base_url,
            deserialized.global_settings.base_url
        );
        assert_eq!(
            config.global_settings.expose_metrics,
            deserialized.global_settings.expose_metrics
        );
        assert_eq!(
            config.worker_configs["default"].init_bash,
            deserialized.worker_configs["default"].init_bash
        );
    }

    #[test]
    fn autoscaling_script_integration_roundtrips() {
        let json = r#"{
            "enabled": true,
            "min_workers": 2,
            "max_workers": 8,
            "integration": {"type": "script", "path": "f/scale", "tag": "admin"}
        }"#;
        let config: AutoscalingConfig =
            serde_json::from_str(json).expect("Should deserialize autoscaling config");
        assert!(config.enabled);
        let integration = config.integration.as_ref().unwrap();
        assert_eq!(integration.integration_type, "script");
        assert_eq!(integration.path.as_deref(), Some("f/scale"));
        assert_eq!(integration.tag.as_deref(), Some("admin"));
    }

    #[test]
    fn ducklake_settings_roundtrips() {
        let json = r#"{
            "ducklakes": {
                "main": {
                    "catalog": {"resource_type": "postgresql", "resource_path": "u/admin/pg"},
                    "storage": {"path": "/data/ducklake"}
                }
            }
        }"#;
        let settings: DucklakeSettings =
            serde_json::from_str(json).expect("Should deserialize ducklake settings");
        assert!(settings.ducklakes.contains_key("main"));
        assert_eq!(
            settings.ducklakes["main"].catalog.resource_type,
            DucklakeCatalogResourceType::Postgresql
        );
    }

    #[test]
    fn critical_error_channels_roundtrips() {
        let json = r##"[
            {"email": "admin@example.com"},
            {"slack_channel": "#alerts"},
            {"teams_channel": {"team_id": "t1", "team_name": "T1", "channel_id": "c1", "channel_name": "C1"}}
        ]"##;
        let channels: Vec<CriticalErrorChannel> =
            serde_json::from_str(json).expect("Should deserialize channels");
        assert_eq!(channels.len(), 3);
        assert!(matches!(channels[0], CriticalErrorChannel::Email { .. }));
        assert!(matches!(channels[1], CriticalErrorChannel::Slack { .. }));
        assert!(matches!(channels[2], CriticalErrorChannel::Teams { .. }));
    }

    // -----------------------------------------------------------------------
    // Diff tests
    // -----------------------------------------------------------------------

    #[test]
    fn diff_global_settings_detects_changes() {
        let mut current = BTreeMap::new();
        current.insert("a".to_string(), serde_json::json!("old"));
        current.insert("b".to_string(), serde_json::json!(1));

        let mut desired = BTreeMap::new();
        desired.insert("a".to_string(), serde_json::json!("new"));
        desired.insert("b".to_string(), serde_json::json!(1)); // unchanged
        desired.insert("c".to_string(), serde_json::json!(true)); // new

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(diff.upserts.len(), 2); // a (changed) + c (new)
        assert!(diff.upserts.contains_key("a"));
        assert!(diff.upserts.contains_key("c"));
        assert!(!diff.upserts.contains_key("b"));
        assert!(diff.deletes.is_empty(), "Merge mode should not delete");
    }

    #[test]
    fn diff_global_settings_replace_mode_deletes() {
        let mut current = BTreeMap::new();
        current.insert("keep".to_string(), serde_json::json!("v"));
        current.insert("remove".to_string(), serde_json::json!("v"));
        current.insert(
            "ducklake_settings".to_string(),
            serde_json::json!("protected"),
        );

        let mut desired = BTreeMap::new();
        desired.insert("keep".to_string(), serde_json::json!("v"));

        let diff = diff_global_settings(&current, &desired, ApplyMode::Replace);
        assert!(diff.upserts.is_empty(), "No changes to upsert");
        assert_eq!(diff.deletes, vec!["remove".to_string()]);
        assert!(
            !diff.deletes.contains(&"ducklake_settings".to_string()),
            "Protected settings should not be deleted"
        );
    }

    #[test]
    fn diff_worker_configs_merge() {
        let mut current = BTreeMap::new();
        current.insert("default".to_string(), serde_json::json!({"a": 1}));
        current.insert("old_group".to_string(), serde_json::json!({"b": 2}));

        let mut desired = BTreeMap::new();
        desired.insert("default".to_string(), serde_json::json!({"a": 1})); // unchanged
        desired.insert("new_group".to_string(), serde_json::json!({"c": 3}));

        let diff = diff_worker_configs(&current, &desired, ApplyMode::Merge);
        assert_eq!(diff.upserts.len(), 1);
        assert!(diff.upserts.contains_key("new_group"));
        assert!(diff.deletes.is_empty());
    }

    #[test]
    fn diff_worker_configs_replace() {
        let mut current = BTreeMap::new();
        current.insert("default".to_string(), serde_json::json!({"a": 1}));
        current.insert("old_group".to_string(), serde_json::json!({"b": 2}));

        let mut desired = BTreeMap::new();
        desired.insert("default".to_string(), serde_json::json!({"a": 1}));

        let diff = diff_worker_configs(&current, &desired, ApplyMode::Replace);
        assert!(diff.upserts.is_empty());
        assert_eq!(diff.deletes, vec!["old_group".to_string()]);
    }

    // -----------------------------------------------------------------------
    // Diff edge cases
    // -----------------------------------------------------------------------

    #[test]
    fn diff_global_settings_both_empty() {
        let current = BTreeMap::new();
        let desired = BTreeMap::new();

        let diff_merge = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert!(diff_merge.upserts.is_empty());
        assert!(diff_merge.deletes.is_empty());

        let diff_replace = diff_global_settings(&current, &desired, ApplyMode::Replace);
        assert!(diff_replace.upserts.is_empty());
        assert!(diff_replace.deletes.is_empty());
    }

    #[test]
    fn diff_global_settings_empty_current() {
        let current = BTreeMap::new();
        let mut desired = BTreeMap::new();
        desired.insert("a".to_string(), serde_json::json!(1));
        desired.insert("b".to_string(), serde_json::json!("two"));

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(diff.upserts.len(), 2);
        assert!(diff.deletes.is_empty());
    }

    #[test]
    fn diff_global_settings_identical() {
        let mut current = BTreeMap::new();
        current.insert("a".to_string(), serde_json::json!(1));
        current.insert("b".to_string(), serde_json::json!({"nested": true}));

        let desired = current.clone();

        let diff_merge = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert!(diff_merge.upserts.is_empty());
        assert!(diff_merge.deletes.is_empty());

        let diff_replace = diff_global_settings(&current, &desired, ApplyMode::Replace);
        assert!(diff_replace.upserts.is_empty());
        assert!(diff_replace.deletes.is_empty());
    }

    #[test]
    fn diff_global_settings_replace_protects_all() {
        let mut current = BTreeMap::new();
        for key in PROTECTED_SETTINGS {
            current.insert(key.to_string(), serde_json::json!("val"));
        }
        current.insert("unprotected".to_string(), serde_json::json!("val"));

        let desired = BTreeMap::new();
        let diff = diff_global_settings(&current, &desired, ApplyMode::Replace);
        assert_eq!(diff.deletes, vec!["unprotected".to_string()]);
        for key in PROTECTED_SETTINGS {
            assert!(
                !diff.deletes.contains(&key.to_string()),
                "Protected key {key} should not be in deletes"
            );
        }
    }

    #[test]
    fn diff_global_settings_merge_deletes_on_null_or_empty_string() {
        // Frontend sends explicit null when the user removes a key (e.g.
        // deletes `object_store_cache_config` in the YAML editor). Merge mode
        // must honor this as a deletion, just like `set_global_setting_internal`.
        let mut current = BTreeMap::new();
        current.insert(
            "object_store_cache_config".to_string(),
            serde_json::json!({"type": "S3"}),
        );
        current.insert("base_url".to_string(), serde_json::json!("http://x"));
        current.insert("hub_base_url".to_string(), serde_json::json!("http://y"));

        let mut desired = BTreeMap::new();
        desired.insert(
            "object_store_cache_config".to_string(),
            serde_json::Value::Null,
        );
        desired.insert("base_url".to_string(), serde_json::json!("http://x"));
        desired.insert("hub_base_url".to_string(), serde_json::json!(""));

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert!(diff.upserts.is_empty(), "no upserts expected");
        let mut deletes = diff.deletes.clone();
        deletes.sort();
        assert_eq!(
            deletes,
            vec![
                "hub_base_url".to_string(),
                "object_store_cache_config".to_string()
            ]
        );
    }

    #[test]
    fn diff_global_settings_merge_never_deletes() {
        let mut current = BTreeMap::new();
        current.insert("a".to_string(), serde_json::json!(1));
        current.insert("b".to_string(), serde_json::json!(2));
        current.insert("c".to_string(), serde_json::json!(3));

        let mut desired = BTreeMap::new();
        desired.insert("a".to_string(), serde_json::json!(1));
        // b and c are absent from desired

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert!(
            diff.deletes.is_empty(),
            "Merge mode should never produce deletes regardless of absent keys"
        );
    }

    #[test]
    fn diff_worker_configs_both_empty() {
        let current = BTreeMap::new();
        let desired = BTreeMap::new();

        let diff = diff_worker_configs(&current, &desired, ApplyMode::Replace);
        assert!(diff.upserts.is_empty());
        assert!(diff.deletes.is_empty());
    }

    #[test]
    fn diff_worker_configs_empty_current() {
        let current = BTreeMap::new();
        let mut desired = BTreeMap::new();
        desired.insert("default".to_string(), serde_json::json!({"init": "bash"}));

        let diff = diff_worker_configs(&current, &desired, ApplyMode::Replace);
        assert_eq!(diff.upserts.len(), 1);
        assert!(diff.deletes.is_empty());
    }

    #[test]
    fn diff_worker_configs_identical() {
        let mut data = BTreeMap::new();
        data.insert("g1".to_string(), serde_json::json!({"a": 1}));

        let diff = diff_worker_configs(&data, &data.clone(), ApplyMode::Replace);
        assert!(diff.upserts.is_empty());
        assert!(diff.deletes.is_empty());
    }

    // -----------------------------------------------------------------------
    // to_settings_map edge cases
    // -----------------------------------------------------------------------

    /// Regression test for bulk-endpoint deletion of typed settings.
    ///
    /// A naive `#[derive(Deserialize)]` would route
    /// `{"object_store_cache_config": null}` to the typed `Option<...>` field
    /// as `None`, and `skip_serializing_if = "Option::is_none"` would then
    /// strip it from `to_settings_map`, so `diff_global_settings` (Merge mode)
    /// would never see the deletion. The manual `Deserialize` impl on
    /// `GlobalSettings` captures explicit top-level nulls into `extra` so they
    /// survive the round-trip and reach `diff_global_settings` as proper
    /// deletes.
    #[test]
    fn explicit_null_on_typed_field_survives_round_trip() {
        let json = serde_json::json!({
            "object_store_cache_config": null,
            "secret_backend": null,
            "smtp_settings": null,
            "base_url": "https://x",
        });
        let settings: GlobalSettings =
            serde_json::from_value(json).expect("null should deserialize");
        assert!(settings.object_store_cache_config.is_none());
        assert!(settings.secret_backend.is_none());
        assert!(settings.smtp_settings.is_none());
        assert_eq!(settings.base_url.as_deref(), Some("https://x"));
        let map = settings.to_settings_map();
        assert_eq!(map["object_store_cache_config"], serde_json::Value::Null);
        assert_eq!(map["secret_backend"], serde_json::Value::Null);
        assert_eq!(map["smtp_settings"], serde_json::Value::Null);
        assert_eq!(map["base_url"], serde_json::json!("https://x"));
    }

    /// Absent keys remain absent — critical so a PUT that only sets a single
    /// field doesn't accidentally delete every other setting in Merge mode.
    #[test]
    fn absent_typed_fields_do_not_appear_in_map() {
        let settings: GlobalSettings =
            serde_json::from_value(serde_json::json!({"base_url": "https://x"})).unwrap();
        let map = settings.to_settings_map();
        assert!(!map.contains_key("object_store_cache_config"));
        assert!(!map.contains_key("smtp_settings"));
        assert_eq!(map.get("base_url"), Some(&serde_json::json!("https://x")));
    }

    /// End-to-end: deserialize → `to_settings_map` → diff in Merge mode with
    /// an explicit null produces a delete (the scenario that silently failed
    /// before the manual `Deserialize` impl).
    #[test]
    fn deserialize_then_diff_deletes_typed_null() {
        let mut current = BTreeMap::new();
        current.insert(
            "object_store_cache_config".to_string(),
            serde_json::json!({"type": "S3", "bucket": "b"}),
        );
        let desired: GlobalSettings =
            serde_json::from_value(serde_json::json!({"object_store_cache_config": null})).unwrap();
        let desired_map = desired.to_settings_map();
        let diff = diff_global_settings(&current, &desired_map, ApplyMode::Merge);
        assert!(diff.upserts.is_empty());
        assert_eq!(diff.deletes, vec!["object_store_cache_config".to_string()]);
    }

    /// Nested nulls inside a typed sub-struct are not promoted to top-level
    /// deletes — only the top-level key matters, mirroring the per-key API.
    #[test]
    fn nested_null_inside_typed_field_is_not_treated_as_top_level_null() {
        let settings: GlobalSettings =
            serde_json::from_value(serde_json::json!({"smtp_settings": {"smtp_host": null}}))
                .unwrap();
        let map = settings.to_settings_map();
        assert!(
            map["smtp_settings"].is_object(),
            "smtp_settings should be an object, not null"
        );
    }

    /// Unknown/extra keys with explicit null still flow through `extra` — this
    /// was already correct before the manual impl; guard against regression.
    #[test]
    fn explicit_null_on_extra_field_survives_round_trip() {
        let settings: GlobalSettings =
            serde_json::from_value(serde_json::json!({"unknown_legacy_setting": null})).unwrap();
        let map = settings.to_settings_map();
        assert_eq!(map["unknown_legacy_setting"], serde_json::Value::Null);
    }

    /// Top-level non-object input must reject with a deserialize error rather
    /// than silently producing defaults. Matches the previous derive behavior.
    #[test]
    fn non_object_top_level_input_errors() {
        assert!(serde_json::from_value::<GlobalSettings>(serde_json::json!(null)).is_err());
        assert!(serde_json::from_value::<GlobalSettings>(serde_json::json!("s")).is_err());
        assert!(serde_json::from_value::<GlobalSettings>(serde_json::json!(42)).is_err());
        assert!(serde_json::from_value::<GlobalSettings>(serde_json::json!([])).is_err());
    }

    #[test]
    fn to_settings_map_empty_defaults() {
        let settings = GlobalSettings::default();
        let map = settings.to_settings_map();
        assert!(
            map.is_empty(),
            "Default GlobalSettings should produce an empty map"
        );
    }

    #[test]
    fn to_settings_map_preserves_extra_and_typed_fields() {
        let mut settings = GlobalSettings {
            base_url: Some("https://example.com".to_string()),
            expose_metrics: Some(true),
            ..Default::default()
        };
        settings
            .extra
            .insert("custom_key".to_string(), serde_json::json!("custom_val"));

        let map = settings.to_settings_map();
        assert_eq!(map["base_url"], serde_json::json!("https://example.com"));
        assert_eq!(map["expose_metrics"], serde_json::json!(true));
        assert_eq!(map["custom_key"], serde_json::json!("custom_val"));
    }

    #[test]
    fn to_settings_map_roundtrip_through_deserialization() {
        let original = GlobalSettings {
            base_url: Some("https://windmill.dev".to_string()),
            retention_period_secs: Some(86400),
            expose_metrics: Some(false),
            smtp_settings: Some(SmtpSettings {
                smtp_host: Some("mail.example.com".to_string()),
                smtp_port: Some(465),
                smtp_tls_implicit: Some(true),
                ..Default::default()
            }),
            custom_tags: Some(vec!["gpu".to_string(), "mem".to_string()]),
            ..Default::default()
        };

        let map = original.to_settings_map();
        let json_map: serde_json::Map<String, serde_json::Value> = map.into_iter().collect();
        let reconstructed: GlobalSettings =
            serde_json::from_value(serde_json::Value::Object(json_map))
                .expect("Should deserialize from settings map");

        assert_eq!(original.base_url, reconstructed.base_url);
        assert_eq!(
            original.retention_period_secs,
            reconstructed.retention_period_secs
        );
        assert_eq!(original.expose_metrics, reconstructed.expose_metrics);
        assert_eq!(
            original.smtp_settings.as_ref().unwrap().smtp_host,
            reconstructed.smtp_settings.as_ref().unwrap().smtp_host
        );
        assert_eq!(
            original.smtp_settings.as_ref().unwrap().smtp_port,
            reconstructed.smtp_settings.as_ref().unwrap().smtp_port
        );
        assert_eq!(original.custom_tags, reconstructed.custom_tags);
    }

    #[test]
    fn smtp_settings_keep_clicktracking_off() {
        let smtp = serde_json::json!({
            "smtp_host": "smtp.example.com",
            "smtp_port": 587,
            "smtp_tls_implicit": false,
            "smtp_disable_tls": false,
            "smtp_clicktracking_off": true,
        });
        let settings: GlobalSettings =
            serde_json::from_value(serde_json::json!({ "smtp_settings": smtp })).unwrap();
        assert_eq!(settings.to_settings_map()["smtp_settings"], smtp);
    }

    // -----------------------------------------------------------------------
    // Serialization edge cases
    // -----------------------------------------------------------------------

    #[test]
    fn global_settings_null_json_value_in_extra() {
        let json = r#"{"unknown_null_key": null, "base_url": "https://x.com"}"#;
        let settings: GlobalSettings = serde_json::from_str(json).unwrap();
        assert_eq!(settings.base_url.as_deref(), Some("https://x.com"));
        assert_eq!(settings.extra["unknown_null_key"], serde_json::Value::Null);

        let map = settings.to_settings_map();
        assert_eq!(map["unknown_null_key"], serde_json::Value::Null);
    }

    #[test]
    fn worker_config_all_typed_fields() {
        let json = r#"{
            "worker_tags": ["tag1", "tag2"],
            "priority_tags": {"tag1": 5},
            "dedicated_worker": "ws:f/my_script",
            "dedicated_workers": ["ws:f/a", "ws:f/b"],
            "init_bash": "apt-get install -y curl",
            "periodic_script_bash": "echo ping",
            "periodic_script_interval_seconds": 300,
            "cache_clear": 7,
            "additional_python_paths": ["/opt/py"],
            "pip_local_dependencies": ["requests"],
            "env_vars_static": {"FOO": "bar"},
            "env_vars_allowlist": ["PATH"],
            "min_alive_workers_alert_threshold": 2,
            "autoscaling": {
                "enabled": true,
                "min_workers": 1,
                "max_workers": 5,
                "integration": {"type": "dryrun"}
            }
        }"#;
        let config: WorkerGroupConfig = serde_json::from_str(json).unwrap();
        assert_eq!(config.worker_tags.as_ref().unwrap().len(), 2);
        assert_eq!(config.priority_tags.as_ref().unwrap()["tag1"], 5);
        assert_eq!(config.dedicated_worker.as_deref(), Some("ws:f/my_script"));
        assert_eq!(config.dedicated_workers.as_ref().unwrap().len(), 2);
        assert_eq!(config.init_bash.as_deref(), Some("apt-get install -y curl"));
        assert_eq!(config.periodic_script_bash.as_deref(), Some("echo ping"));
        assert_eq!(config.periodic_script_interval_seconds, Some(300));
        assert_eq!(config.extra["cache_clear"], serde_json::json!(7));
        assert_eq!(config.additional_python_paths.as_ref().unwrap().len(), 1);
        assert_eq!(config.pip_local_dependencies.as_ref().unwrap().len(), 1);
        assert_eq!(config.env_vars_static.as_ref().unwrap()["FOO"], "bar");
        assert_eq!(config.env_vars_allowlist.as_ref().unwrap().len(), 1);
        assert_eq!(config.min_alive_workers_alert_threshold, Some(2));
        let auto = config.autoscaling.as_ref().unwrap();
        assert!(auto.enabled);
        assert_eq!(auto.min_workers, Some(1));
        assert_eq!(auto.max_workers, Some(5));
        assert_eq!(
            auto.integration.as_ref().unwrap().integration_type,
            "dryrun"
        );
    }

    #[test]
    fn otel_tracing_proxy_languages_roundtrip() {
        let json = r#"{"enabled": true, "enabled_languages": ["python3", "bun", "deno"]}"#;
        let settings: OtelTracingProxySettings = serde_json::from_str(json).unwrap();
        assert!(settings.enabled);
        assert_eq!(settings.enabled_languages.len(), 3);
        assert_eq!(settings.enabled_languages[0], ScriptLang::Python3);
        assert_eq!(settings.enabled_languages[1], ScriptLang::Bun);
        assert_eq!(settings.enabled_languages[2], ScriptLang::Deno);

        let serialized = serde_json::to_string(&settings).unwrap();
        let deserialized: OtelTracingProxySettings = serde_json::from_str(&serialized).unwrap();
        assert_eq!(deserialized.enabled_languages.len(), 3);
    }

    #[test]
    fn oauth_full_config_roundtrips() {
        let json = r#"{
            "id": "client_id_123",
            "secret": "secret_456",
            "allowed_domains": ["example.com", "test.com"],
            "connect_config": {
                "auth_url": "https://auth.example.com/authorize",
                "token_url": "https://auth.example.com/token",
                "userinfo_url": "https://auth.example.com/userinfo",
                "scopes": ["openid", "profile"],
                "extra_params": {"prompt": "consent"},
                "req_body_auth": true
            },
            "login_config": {
                "auth_url": "https://login.example.com/authorize",
                "token_url": "https://login.example.com/token"
            }
        }"#;
        let client: OAuthClient = serde_json::from_str(json).unwrap();
        assert_eq!(client.id, "client_id_123");
        assert_eq!(client.secret, *"secret_456");
        assert_eq!(client.allowed_domains.as_ref().unwrap().len(), 2);
        let cc = client.connect_config.as_ref().unwrap();
        assert_eq!(cc.auth_url, "https://auth.example.com/authorize");
        assert_eq!(cc.scopes.as_ref().unwrap().len(), 2);
        assert_eq!(cc.extra_params.as_ref().unwrap()["prompt"], "consent");
        assert_eq!(cc.req_body_auth, Some(true));
        let lc = client.login_config.as_ref().unwrap();
        assert_eq!(lc.token_url, "https://login.example.com/token");
        assert!(lc.userinfo_url.is_none());

        let roundtripped = serde_json::to_string(&client).unwrap();
        let parsed: OAuthClient = serde_json::from_str(&roundtripped).unwrap();
        assert_eq!(parsed.id, client.id);
    }

    #[test]
    fn oauth_entry_valid_config_deserializes_as_typed() {
        let json = r#"{
            "id": "slack_id",
            "secret": "slack_secret",
            "connect_config": {
                "auth_url": "https://slack.com/oauth/v2/authorize",
                "token_url": "https://slack.com/api/oauth.v2.access",
                "scopes": ["channels:read", "chat:write"],
                "req_body_auth": true
            }
        }"#;
        let entry: OAuthClientEntry = serde_json::from_str(json).unwrap();
        match &entry {
            OAuthClientEntry::Typed(c) => {
                assert_eq!(c.id, "slack_id");
                let cc = c.connect_config.as_ref().unwrap();
                assert_eq!(cc.req_body_auth, Some(true));
            }
            OAuthClientEntry::Raw { .. } => panic!("expected Typed variant"),
        }
        // round-trip preserves the value
        let serialized = serde_json::to_string(&entry).unwrap();
        assert!(serialized.contains("slack_id"));
    }

    #[test]
    fn oauth_entry_string_bool_falls_back_to_raw_with_error() {
        // This is the real-world scenario: req_body_auth stored as "true" (string)
        let json = r#"{
            "id": "custom_provider",
            "secret": "secret",
            "connect_config": {
                "auth_url": "https://example.com/auth",
                "token_url": "https://example.com/token",
                "req_body_auth": "true"
            }
        }"#;
        let entry: OAuthClientEntry = serde_json::from_str(json).unwrap();
        match &entry {
            OAuthClientEntry::Raw { error, .. } => {
                assert!(
                    error.contains("invalid type"),
                    "error should mention type mismatch, got: {error}"
                );
            }
            OAuthClientEntry::Typed(_) => panic!("expected Raw fallback for string bool"),
        }
        // Serialization still round-trips the raw JSON faithfully
        let serialized = serde_json::to_string(&entry).unwrap();
        assert!(serialized.contains("custom_provider"));
        assert!(serialized.contains(r#""req_body_auth":"true""#));
    }

    #[test]
    fn oauth_entry_map_with_mixed_valid_and_invalid() {
        // Simulates what InstanceConfig.oauths looks like when one provider is
        // well-formed and another has a string-typed boolean.
        let json = r#"{
            "google": {
                "id": "google_id",
                "secret": "google_secret",
                "connect_config": {
                    "auth_url": "https://accounts.google.com/o/oauth2/v2/auth",
                    "token_url": "https://oauth2.googleapis.com/token"
                }
            },
            "broken": {
                "id": "broken_id",
                "secret": "broken_secret",
                "connect_config": {
                    "auth_url": "https://example.com/auth",
                    "token_url": "https://example.com/token",
                    "req_body_auth": "false"
                }
            }
        }"#;
        let map: BTreeMap<String, OAuthClientEntry> = serde_json::from_str(json).unwrap();
        assert!(matches!(map["google"], OAuthClientEntry::Typed(_)));
        assert!(matches!(map["broken"], OAuthClientEntry::Raw { .. }));
    }

    #[test]
    fn server_generated_db_passwords_are_isolated_from_config() {
        // These hold passwords the server generates: written by setup/refresh via direct SQL,
        // never operator-authored. They must stay out of the declarative config surface
        // (hidden on read) and be undeletable, so config sync can't read, rewrite, or drop them.
        for key in [
            "custom_instance_replication_pwd",
            "external_instance_pg_state",
        ] {
            assert!(HIDDEN_SETTINGS.contains(&key), "{key}");
            assert!(PROTECTED_SETTINGS.contains(&key), "{key}");
            assert!(SENSITIVE_SETTINGS.contains(&key), "{key}");

            // A stray desired value (e.g. flattened into `extra`) is ignored, not upserted.
            let mut desired = BTreeMap::new();
            desired.insert(key.to_string(), serde_json::json!("attacker-set"));
            let diff = diff_global_settings(&BTreeMap::new(), &desired, ApplyMode::Merge);
            assert!(diff.upserts.is_empty(), "{key} must not be upserted");

            // A current value is never deleted by a Replace that omits it.
            let mut current = BTreeMap::new();
            current.insert(key.to_string(), serde_json::json!("live"));
            let diff = diff_global_settings(&current, &BTreeMap::new(), ApplyMode::Replace);
            assert!(
                !diff.deletes.contains(&key.to_string()),
                "{key} must not be deleted"
            );
        }
    }

    #[test]
    fn custom_instance_pg_databases_roundtrips_and_redacts_user_pwd() {
        // user_pwd stays operator-configurable (EE secretKeyRef); databases is runtime status.
        let json = r#"{
            "user_pwd": "secret123",
            "databases": { "mydb": { "success": true, "tag": "production" } }
        }"#;
        let pg: CustomInstancePgDatabases = serde_json::from_str(json).unwrap();
        assert_eq!(
            pg.user_pwd.as_ref().and_then(|v| v.as_literal()),
            Some("secret123")
        );
        assert!(pg.databases["mydb"].success);

        let out = format_setting_value(
            "custom_instance_pg_databases",
            &serde_json::json!({ "user_pwd": "user-plaintext-password" }),
        );
        assert!(
            !out.contains("user-plaintext-password"),
            "user_pwd leaked: {out}"
        );
    }

    #[test]
    fn db_oversize_alert_defaults() {
        let alert: DbOversizeAlert = serde_json::from_str("{}").unwrap();
        assert!(!alert.enabled);
        assert_eq!(alert.value, 0.0);

        let alert_set: DbOversizeAlert =
            serde_json::from_str(r#"{"enabled": true, "value": 50.5}"#).unwrap();
        assert!(alert_set.enabled);
        assert!((alert_set.value - 50.5).abs() < f32::EPSILON);
    }

    #[test]
    fn instance_config_empty_roundtrips() {
        let config = InstanceConfig::default();
        let json = serde_json::to_string(&config).unwrap();
        let parsed: InstanceConfig = serde_json::from_str(&json).unwrap();
        assert!(parsed.global_settings.base_url.is_none());
        assert!(parsed.worker_configs.is_empty());
    }

    #[test]
    fn diff_value_type_change_detected() {
        let mut current = BTreeMap::new();
        current.insert("key".to_string(), serde_json::json!("string_value"));

        let mut desired = BTreeMap::new();
        desired.insert("key".to_string(), serde_json::json!(42));

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(diff.upserts.len(), 1);
        assert_eq!(diff.upserts["key"], serde_json::json!(42));
    }

    #[test]
    fn diff_nested_json_change_detected() {
        let mut current = BTreeMap::new();
        current.insert(
            "smtp".to_string(),
            serde_json::json!({"host": "a.com", "port": 25}),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "smtp".to_string(),
            serde_json::json!({"host": "a.com", "port": 587}),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(diff.upserts.len(), 1);
        assert_eq!(diff.upserts["smtp"]["port"], 587);
    }

    // -----------------------------------------------------------------------
    // StringOrSecretRef tests
    // -----------------------------------------------------------------------

    #[test]
    fn string_or_secret_ref_deserialize_literal() {
        let v: StringOrSecretRef = serde_json::from_str(r#""hello""#).unwrap();
        assert_eq!(v.as_literal(), Some("hello"));
        assert!(!v.is_secret_ref());
    }

    #[test]
    fn string_or_secret_ref_deserialize_secret_ref() {
        let json = r#"{"secretKeyRef": {"name": "my-secret", "key": "password"}}"#;
        let v: StringOrSecretRef = serde_json::from_str(json).unwrap();
        assert!(v.is_secret_ref());
        assert!(v.as_literal().is_none());
        let r = v.as_secret_ref().unwrap();
        assert_eq!(r.name, "my-secret");
        assert_eq!(r.key, "password");
    }

    #[test]
    fn string_or_secret_ref_serialize_literal() {
        let v = StringOrSecretRef::Literal("plain".to_string());
        let json = serde_json::to_value(&v).unwrap();
        assert_eq!(json, serde_json::json!("plain"));
    }

    #[test]
    fn string_or_secret_ref_serialize_secret_ref() {
        let v = StringOrSecretRef::SecretRef(SecretKeyRefWrapper {
            secret_key_ref: SecretKeyRef { name: "s".to_string(), key: "k".to_string() },
        });
        let json = serde_json::to_value(&v).unwrap();
        assert_eq!(
            json,
            serde_json::json!({"secretKeyRef": {"name": "s", "key": "k"}})
        );
    }

    #[test]
    fn string_or_secret_ref_from_impls() {
        let v: StringOrSecretRef = "hello".into();
        assert_eq!(v, *"hello");

        let v: StringOrSecretRef = String::from("world").into();
        assert_eq!(v, *"world");
    }

    #[test]
    fn string_or_secret_ref_debug_masks_literal() {
        let v = StringOrSecretRef::Literal("super-secret-value".to_string());
        let debug = format!("{v:?}");
        assert_eq!(debug, "Literal(****)");
        assert!(!debug.contains("super-secret-value"));
    }

    #[test]
    fn format_setting_value_redacts_oauth_secrets() {
        let val = serde_json::json!({
            "google": {"id": "client-id", "secret": "my-super-secret-12345"}
        });
        let formatted = format_setting_value("oauths", &val);
        assert!(!formatted.contains("my-super-secret-12345"));
        assert!(formatted.contains("client-id"));
        assert!(formatted.contains("****"));
    }

    #[test]
    fn format_setting_value_redacts_nested_credentials() {
        let val = serde_json::json!({
            "type": "Azure",
            "accountName": "acct",
            "containerName": "c",
            "accessKey": "azure-storage-account-key-12345"
        });
        let formatted = format_setting_value("object_store_cache_config", &val);
        assert!(!formatted.contains("azure-storage-account-key-12345"));
        assert!(formatted.contains("acct"));

        let val = serde_json::json!({
            "app_id": 1,
            "private_key": "-----BEGIN RSA PRIVATE KEY-----\nMIIEsecretbody\n-----END RSA PRIVATE KEY-----"
        });
        let formatted = format_setting_value("github_enterprise_app", &val);
        assert!(!formatted.contains("MIIEsecretbody"));
    }

    #[test]
    #[should_panic(expected = "literal_value() called on unresolved secret ref")]
    fn string_or_secret_ref_literal_value_panics_on_ref() {
        let v = StringOrSecretRef::SecretRef(SecretKeyRefWrapper {
            secret_key_ref: SecretKeyRef { name: "s".to_string(), key: "k".to_string() },
        });
        let _ = v.literal_value();
    }

    #[test]
    fn global_settings_with_mixed_literal_and_ref() {
        let json = r#"{
            "license_key": {"secretKeyRef": {"name": "wm-secrets", "key": "license"}},
            "hub_api_secret": "plain-secret",
            "base_url": "https://example.com"
        }"#;
        let gs: GlobalSettings = serde_json::from_str(json).unwrap();
        assert!(gs.license_key.as_ref().unwrap().is_secret_ref());
        assert_eq!(
            gs.hub_api_secret.as_ref().and_then(|v| v.as_literal()),
            Some("plain-secret")
        );
        assert_eq!(gs.base_url.as_deref(), Some("https://example.com"));
    }

    #[test]
    fn string_or_secret_ref_backward_compat_plain_string() {
        let json = r#"{"license_key": "my-key", "scim_token": "tok"}"#;
        let gs: GlobalSettings = serde_json::from_str(json).unwrap();
        assert_eq!(
            gs.license_key.as_ref().and_then(|v| v.as_literal()),
            Some("my-key")
        );
        assert_eq!(
            gs.scim_token.as_ref().and_then(|v| v.as_literal()),
            Some("tok")
        );

        let roundtripped = serde_json::to_string(&gs).unwrap();
        let gs2: GlobalSettings = serde_json::from_str(&roundtripped).unwrap();
        assert_eq!(
            gs2.license_key.as_ref().and_then(|v| v.as_literal()),
            Some("my-key")
        );
    }

    // -----------------------------------------------------------------------
    // EnvRef tests
    // -----------------------------------------------------------------------

    #[test]
    fn env_ref_deserialize() {
        let json = r#"{"envRef": "MY_LICENSE_KEY"}"#;
        let v: StringOrSecretRef = serde_json::from_str(json).unwrap();
        assert!(v.is_env_ref());
        assert!(!v.is_secret_ref());
        assert!(v.as_literal().is_none());
        assert_eq!(v.as_env_ref(), Some("MY_LICENSE_KEY"));
    }

    #[test]
    fn env_ref_serialize() {
        let v = StringOrSecretRef::EnvRef(EnvRefWrapper { env_ref: "MY_VAR".to_string() });
        let json = serde_json::to_value(&v).unwrap();
        assert_eq!(json, serde_json::json!({"envRef": "MY_VAR"}));
    }

    #[test]
    fn env_ref_display() {
        let v = StringOrSecretRef::EnvRef(EnvRefWrapper { env_ref: "FOO".to_string() });
        assert_eq!(format!("{v}"), "envRef(FOO)");
    }

    #[test]
    #[should_panic(expected = "literal_value() called on unresolved env ref")]
    fn env_ref_literal_value_panics() {
        let v = StringOrSecretRef::EnvRef(EnvRefWrapper { env_ref: "X".to_string() });
        let _ = v.literal_value();
    }

    #[test]
    fn env_ref_equality() {
        let a = StringOrSecretRef::EnvRef(EnvRefWrapper { env_ref: "X".to_string() });
        let b = StringOrSecretRef::EnvRef(EnvRefWrapper { env_ref: "X".to_string() });
        let c = StringOrSecretRef::Literal("X".to_string());
        assert_eq!(a, b);
        assert_ne!(a, c);
    }

    #[test]
    fn global_settings_with_env_ref() {
        let json = r#"{
            "license_key": {"envRef": "WM_LICENSE_KEY"},
            "hub_api_secret": "plain",
            "base_url": "https://example.com"
        }"#;
        let gs: GlobalSettings = serde_json::from_str(json).unwrap();
        assert!(gs.license_key.as_ref().unwrap().is_env_ref());
        assert_eq!(
            gs.license_key.as_ref().unwrap().as_env_ref(),
            Some("WM_LICENSE_KEY")
        );
        assert_eq!(
            gs.hub_api_secret.as_ref().and_then(|v| v.as_literal()),
            Some("plain")
        );
    }

    #[test]
    fn resolve_env_refs_replaces_env_ref() {
        // Set a test env var
        unsafe { std::env::set_var("__WM_TEST_LICENSE", "resolved-value") };

        let mut gs = GlobalSettings {
            license_key: Some(StringOrSecretRef::EnvRef(EnvRefWrapper {
                env_ref: "__WM_TEST_LICENSE".to_string(),
            })),
            hub_api_secret: Some(StringOrSecretRef::Literal("plain".to_string())),
            ..Default::default()
        };

        resolve_env_refs(&mut gs).unwrap();
        assert_eq!(
            gs.license_key.as_ref().and_then(|v| v.as_literal()),
            Some("resolved-value")
        );
        // Literal fields are unchanged
        assert_eq!(
            gs.hub_api_secret.as_ref().and_then(|v| v.as_literal()),
            Some("plain")
        );

        unsafe { std::env::remove_var("__WM_TEST_LICENSE") };
    }

    #[test]
    fn resolve_env_refs_errors_on_missing_var() {
        let mut gs = GlobalSettings {
            license_key: Some(StringOrSecretRef::EnvRef(EnvRefWrapper {
                env_ref: "__WM_TEST_NONEXISTENT_12345".to_string(),
            })),
            ..Default::default()
        };

        let err = resolve_env_refs(&mut gs).unwrap_err();
        assert_eq!(err, "__WM_TEST_NONEXISTENT_12345");
    }

    #[test]
    fn resolve_env_refs_handles_smtp_and_oauth() {
        unsafe {
            std::env::set_var("__WM_TEST_SMTP_PWD", "smtp-secret");
            std::env::set_var("__WM_TEST_OAUTH_SECRET", "oauth-secret");
        }

        let mut gs = GlobalSettings {
            smtp_settings: Some(SmtpSettings {
                smtp_password: Some(StringOrSecretRef::EnvRef(EnvRefWrapper {
                    env_ref: "__WM_TEST_SMTP_PWD".to_string(),
                })),
                ..Default::default()
            }),
            oauths: Some({
                let mut m = BTreeMap::new();
                m.insert(
                    "google".to_string(),
                    OAuthClientEntry::Typed(OAuthClient {
                        id: "id".to_string(),
                        secret: StringOrSecretRef::EnvRef(EnvRefWrapper {
                            env_ref: "__WM_TEST_OAUTH_SECRET".to_string(),
                        }),
                        display_name: None,
                        allowed_domains: None,
                        connect_config: None,
                        login_config: None,
                        tenant: None,
                        share_with_workspaces: None,
                        grant_types: vec![],
                        extra: BTreeMap::new(),
                    }),
                );
                m
            }),
            ..Default::default()
        };

        resolve_env_refs(&mut gs).unwrap();

        assert_eq!(
            gs.smtp_settings
                .as_ref()
                .unwrap()
                .smtp_password
                .as_ref()
                .and_then(|v| v.as_literal()),
            Some("smtp-secret")
        );
        match &gs.oauths.as_ref().unwrap()["google"] {
            OAuthClientEntry::Typed(c) => {
                assert_eq!(c.secret.as_literal(), Some("oauth-secret"));
            }
            OAuthClientEntry::Raw { .. } => panic!("expected Typed variant"),
        }

        unsafe {
            std::env::remove_var("__WM_TEST_SMTP_PWD");
            std::env::remove_var("__WM_TEST_OAUTH_SECRET");
        }
    }

    #[test]
    fn mixed_literal_secret_ref_env_ref() {
        let json = r#"{
            "license_key": {"envRef": "MY_LIC"},
            "hub_api_secret": {"secretKeyRef": {"name": "s", "key": "k"}},
            "scim_token": "plain-token"
        }"#;
        let gs: GlobalSettings = serde_json::from_str(json).unwrap();
        assert!(gs.license_key.as_ref().unwrap().is_env_ref());
        assert!(gs.hub_api_secret.as_ref().unwrap().is_secret_ref());
        assert_eq!(
            gs.scim_token.as_ref().and_then(|v| v.as_literal()),
            Some("plain-token")
        );
    }

    // -----------------------------------------------------------------------
    // License key expiry diff tests
    // -----------------------------------------------------------------------

    #[test]
    fn diff_license_key_skips_older_expiry() {
        let mut current = BTreeMap::new();
        current.insert(
            "license_key".to_string(),
            serde_json::json!("client1.2000000000.sig123"),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "license_key".to_string(),
            serde_json::json!("client1.1000000000.sig123"),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert!(
            diff.upserts.is_empty(),
            "Should not update license_key when desired expiry is older"
        );
    }

    #[test]
    fn diff_license_key_skips_equal_expiry_different_sig() {
        let mut current = BTreeMap::new();
        current.insert(
            "license_key".to_string(),
            serde_json::json!("client1.2000000000.sig_a"),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "license_key".to_string(),
            serde_json::json!("client1.2000000000.sig_b"),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert!(
            diff.upserts.is_empty(),
            "Same client and equal expiry should not update even with different signature"
        );
    }

    #[test]
    fn diff_license_key_updates_newer_expiry() {
        let mut current = BTreeMap::new();
        current.insert(
            "license_key".to_string(),
            serde_json::json!("client1.1000000000.sig123"),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "license_key".to_string(),
            serde_json::json!("client1.2000000000.sig123"),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(diff.upserts.len(), 1);
        assert_eq!(
            diff.upserts["license_key"],
            serde_json::json!("client1.2000000000.sig123")
        );
    }

    #[test]
    fn diff_license_key_updates_newer_expiry_different_sig() {
        let mut current = BTreeMap::new();
        current.insert(
            "license_key".to_string(),
            serde_json::json!("client1.1000000000.sig_old"),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "license_key".to_string(),
            serde_json::json!("client1.2000000000.sig_new"),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(
            diff.upserts.len(),
            1,
            "Same client with newer expiry should update even with different signature"
        );
    }

    #[test]
    fn diff_license_key_skips_older_expiry_different_sig() {
        let mut current = BTreeMap::new();
        current.insert(
            "license_key".to_string(),
            serde_json::json!("client1.2000000000.sig_new"),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "license_key".to_string(),
            serde_json::json!("client1.1000000000.sig_old"),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert!(
            diff.upserts.is_empty(),
            "Same client with older expiry should not update even with different signature"
        );
    }

    #[test]
    fn diff_license_key_updates_different_client() {
        let mut current = BTreeMap::new();
        current.insert(
            "license_key".to_string(),
            serde_json::json!("client1.2000000000.sig123"),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "license_key".to_string(),
            serde_json::json!("client2.1000000000.sig456"),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(
            diff.upserts.len(),
            1,
            "Different client ID means different key, should always update"
        );
    }

    #[test]
    fn diff_license_key_non_string_always_updates() {
        let mut current = BTreeMap::new();
        current.insert(
            "license_key".to_string(),
            serde_json::json!({"envRef": "LIC_KEY"}),
        );

        let mut desired = BTreeMap::new();
        desired.insert(
            "license_key".to_string(),
            serde_json::json!({"envRef": "NEW_LIC_KEY"}),
        );

        let diff = diff_global_settings(&current, &desired, ApplyMode::Merge);
        assert_eq!(
            diff.upserts.len(),
            1,
            "Non-string license keys should always be updated when different"
        );
    }
}