vtc-service 0.12.0

//! `config/{show,patch,reload,restart}/0.1` and `vtc/config/{export,import}/0.1`.
//!
//! Implements **M0.8.2** of the VTC MVP Phase 0 plan. Every one is a signed
//! document served by the spine (`trust_tasks::admin_tasks` and the portable
//! pair in `trust_tasks`); none has a REST route.
//!
//! - **show**: returns the four-layer-merged [`EffectiveConfig`].
//! - **patch**: writes overrides to the db-layer (`config` keyspace),
//!   returning `{ applied, pending_restart, rejected }` so the
//!   caller can tell which keys took effect immediately, which
//!   require a daemon restart (M0.8.3), and which were rejected
//!   (and why).
//!
//! Every mutating handler emits an audit event keyed to the calling
//! admin's real DID (the document's signer). Sensitive
//! values are run through `vti_common::audit::ConfigChange::redact_if`
//! before the `ConfigChanged` event is persisted. Audit is
//! fail-closed: a mutation that produces a change but cannot be
//! recorded (no `AuditWriter` configured) returns 503 rather than
//! applying silently.

use std::collections::HashMap;

use axum::http::StatusCode;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tracing::{info, warn};
use vti_common::error::AppError;

use crate::community::{CommunityProfile, CommunityProfileUpdate, load_profile, store_profile};
use crate::config_store::{
    ConfigStore, EffectiveConfig, compute_effective_config, lookup, validate_value,
};
use crate::server::AppState;
#[allow(unused_imports)]
use crate::supervisor::SupervisorKind;
use vti_common::audit::{
    AuditEvent, AuditWriter, CommunityProfileUpdatedData, ConfigChange, ConfigChangedData,
    ConfigReloadedData, ConfigSource, RestartRequestedData,
};

/// PATCH request body: a `key → value` map under `overrides`. Keys not in
/// [`crate::config_store::REGISTRY`] are reported back under
/// `rejected` rather than silently dropped.
///
/// The map is wrapped (rather than `#[serde(flatten)]`ed to the top level) to
/// match canonical `spec/config/patch/0.1`, whose envelope is
/// `additionalProperties: false` around a single `overrides` object.
#[derive(Debug, Deserialize, utoipa::ToSchema)]
#[serde(deny_unknown_fields)]
pub struct PatchRequest {
    pub overrides: HashMap<String, Value>,
}

/// PATCH response body. Lists which keys took effect immediately,
/// which await restart, and which were rejected.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
#[derive(utoipa::ToSchema)]
pub struct PatchResponse {
    pub applied: Vec<String>,
    pub pending_restart: Vec<String>,
    pub rejected: Vec<RejectedKey>,
}

/// One rejected key + the reason. Surfaced to the caller so the
/// admin UX can present a meaningful error inline.
#[derive(Debug, Serialize, utoipa::ToSchema)]
#[schema(as = ConfigRejectedKey)]
pub struct RejectedKey {
    pub key: String,
    pub reason: String,
}

/// GET handler.
pub(crate) async fn get_config(
    state: &AppState,
    keys: Option<&[String]>,
) -> Result<EffectiveConfig, AppError> {
    let cfg = state.config.read().await;
    let store = ConfigStore::new(state.config_ks.clone());
    let mut eff = compute_effective_config(&cfg, &store).await?;
    // `config/show`'s `keys` narrows the answer to the named keys; absent,
    // every key the registry knows.
    if let Some(keys) = keys {
        eff.fields.retain(|f| keys.contains(&f.key));
    }
    Ok(eff)
}

/// PATCH handler.
pub(crate) async fn patch_config(
    state: &AppState,
    actor: &str,
    req: PatchRequest,
) -> Result<PatchResponse, AppError> {
    let store = ConfigStore::new(state.config_ks.clone());
    // Snapshot the current db-layer overrides up front so each applied
    // key's audit record carries its real `old_value` + source.
    let current = store.snapshot().await?;
    let mut applied = Vec::new();
    let mut pending_restart = Vec::new();
    let mut rejected = Vec::new();
    let mut audit_changes: Vec<ConfigChange> = Vec::new();
    let mut requires_restart = false;

    for (key, value) in req.overrides {
        let Some(def) = lookup(&key) else {
            let reason = crate::config_store::host_only_refusal(&key)
                .unwrap_or_else(|| "unknown config key (not in registry)".into());
            rejected.push(RejectedKey { key, reason });
            continue;
        };

        if let Err(e) = validate_value(def, &value) {
            rejected.push(RejectedKey {
                key,
                reason: format!("validation failed: {e}"),
            });
            continue;
        }
        if let Err(e) = check_against_community(state, &key, &value).await {
            rejected.push(RejectedKey {
                key,
                reason: format!("validation failed: {e}"),
            });
            continue;
        }

        let old_value = current.get(&key).cloned();

        if let Err(e) = store.put(&key, &value).await {
            rejected.push(RejectedKey {
                key,
                reason: format!("persistence failed: {e}"),
            });
            continue;
        }
        crate::admin_actions::record_effect(state).await;

        info!(
            key = %key,
            requires_restart = def.requires_restart,
            sensitive = def.sensitive,
            "admin config PATCH applied"
        );
        if def.trust_semantics {
            // VTI-CMP-092: identifiable as a change to trust semantics, not
            // only as one more settings row.
            warn!(
                key = %key,
                actor = %actor,
                trust_semantics = true,
                "a configuration key that alters trust semantics was changed (VTI-CMP-090)"
            );
        }

        let mut change = ConfigChange {
            key: key.clone(),
            old_value: old_value.clone(),
            new_value: value,
            // The db-overlay is the only PATCH-writable layer, so a
            // prior db value was `Db`; absence means the resolved
            // value came from a lower (env/toml/default) layer — we
            // record `Default` to match the import path's convention.
            source_before: if old_value.is_some() {
                ConfigSource::Db
            } else {
                ConfigSource::Default
            },
        };
        change.redact_if(|k| matches!(lookup(k), Some(d) if d.sensitive));
        audit_changes.push(change);

        if def.requires_restart {
            requires_restart = true;
            pending_restart.push(key);
        } else {
            applied.push(key);
        }
    }

    // Fail-closed: a config mutation that can't be audited is refused
    // (matches reload/restart/import). No applied changes → nothing to
    // audit, so a rejects-only or empty PATCH never needs the writer.
    if !audit_changes.is_empty() {
        let audit_writer = require_audit_writer(state)?;
        audit_writer
            .write(
                actor,
                None,
                AuditEvent::ConfigChanged(ConfigChangedData {
                    changes: audit_changes,
                    requires_restart,
                }),
            )
            .await?;
    }

    Ok(PatchResponse {
        applied,
        pending_restart,
        rejected,
    })
}

// ---------------------------------------------------------------------------
// Reload
// ---------------------------------------------------------------------------

/// `restart.drain_timeout` default (seconds). Hardcoded for Phase 0
/// — surfaces in the `RestartRequested` audit event and bounds the
/// graceful-shutdown wait in `run_rest_thread`. A future
/// `restart.drain_timeout` config key plugs in here.
const DEFAULT_DRAIN_TIMEOUT_SECS: u64 = 30;

/// `POST /v1/admin/config/reload` response. Lists the keys whose
/// **db-layer** values were applied in-memory by this call. Keys
/// flagged `requires_restart` never appear here — they re-apply on
/// the next restart.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
#[derive(utoipa::ToSchema)]
pub struct ReloadResponse {
    pub keys_reloaded: Vec<String>,
}

/// `POST /v1/admin/config/reload` handler. Re-reads the
/// `EffectiveConfig` and diffs against the live in-memory config;
/// for each hot-reloadable key whose effective value differs, the
/// in-memory `AppConfig` is updated. Emits `ConfigReloaded` listing
/// the keys that actually changed.
///
/// **Phase 0 limitation**: only the Phase-0 registry's
/// hot-reloadable keys (`log.level` today) are propagated. Future
/// runtime-state subscribers (tracing subscriber filter handle,
/// session-cleanup interval, etc.) will plug into the same diff
/// loop.
pub(crate) async fn reload_config(
    state: &AppState,
    actor: &str,
) -> Result<ReloadResponse, AppError> {
    let audit_writer = require_audit_writer(state)?;

    let store = ConfigStore::new(state.config_ks.clone());

    // Snapshot the live in-memory config so we can diff against what
    // the four-layer overlay currently says. Read the latest
    // effective view first, then mutate the in-memory copy under a
    // write lock so concurrent reads see a single atomic flip per
    // key.
    let new_effective = {
        let cfg = state.config.read().await;
        compute_effective_config(&cfg, &store).await?
    };

    // Compare per-key effective values against `EffectiveConfig`'s
    // serialised snapshot of the same `AppConfig` shape. For Phase 0
    // the registry has three keys (`server.host`, `server.port`,
    // `log.level`). Server keys are `requires_restart` so they
    // never re-apply here.
    let mut keys_reloaded = Vec::new();
    {
        let mut cfg = state.config.write().await;
        for def in crate::config_store::REGISTRY {
            if def.requires_restart {
                continue;
            }
            let new_value = new_effective
                .fields
                .iter()
                .find(|f| f.key == def.key)
                .map(|f| f.value.clone())
                .unwrap_or(Value::Null);
            let live_value = lookup_live(&cfg, def.key);
            if new_value != live_value && apply_to_live(&mut cfg, def.key, &new_value) {
                keys_reloaded.push(def.key.to_string());
            }
        }
    }

    audit_writer
        .write(
            actor,
            None,
            AuditEvent::ConfigReloaded(ConfigReloadedData {
                keys_reloaded: keys_reloaded.clone(),
            }),
        )
        .await?;

    info!(?keys_reloaded, "config reloaded");

    Ok(ReloadResponse { keys_reloaded })
}

// ---------------------------------------------------------------------------
// Restart
// ---------------------------------------------------------------------------

/// `POST /v1/admin/config/restart` response when the supervisor
/// check passes.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
#[derive(utoipa::ToSchema)]
pub struct RestartResponse {
    /// Which supervisor the daemon detected (so the operator's
    /// admin UX can echo it back).
    pub supervisor: SupervisorKind,
    /// `drain_timeout` (seconds) the daemon will use for graceful
    /// shutdown. Mirrors `RestartRequestedData.drain_timeout_seconds`.
    pub drain_timeout_seconds: u64,
}

/// `POST /v1/admin/config/restart` handler.
///
/// Refuses (`412 Precondition Failed`,
/// `SupervisorRequired`) unless a supervisor is detected — restart
/// without an external supervisor is just "kill the process" and a
/// caller asking for `restart` likely means "have the daemon come
/// back up afterwards". Detection lives in
/// [`crate::supervisor::detect_supervisor`].
///
/// On success the handler emits `RestartRequested` to the audit
/// log *before* signalling shutdown — so the row survives even if
/// the drain wedges.
pub(crate) async fn restart_config(
    state: &AppState,
    actor: &str,
) -> Result<RestartResponse, AppError> {
    let audit_writer = require_audit_writer(state)?;

    let supervisor = state.supervisor.ok_or_else(|| AppError::ServiceError {
        status: StatusCode::PRECONDITION_FAILED,
        message: "SupervisorRequired: refusing to restart without a process supervisor \
            (set VTC_SUPERVISED=1 or run under systemd / kubernetes)"
            .into(),
    })?;

    audit_writer
        .write(
            actor,
            None,
            AuditEvent::RestartRequested(RestartRequestedData {
                drain_timeout_seconds: DEFAULT_DRAIN_TIMEOUT_SECS,
            }),
        )
        .await?;

    info!(?supervisor, "restart requested");

    // Flip the shared graceful-shutdown channel. The REST thread
    // observes this via `with_graceful_shutdown` and stops accepting
    // new connections; the storage thread flushes; supervisor
    // restarts the process. We send AFTER audit emission so a wedged
    // drain still leaves the row behind.
    let _ = state.shutdown_tx.send(true);

    Ok(RestartResponse {
        supervisor,
        drain_timeout_seconds: DEFAULT_DRAIN_TIMEOUT_SECS,
    })
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

fn require_audit_writer(state: &AppState) -> Result<&AuditWriter, AppError> {
    state
        .audit_writer
        .as_ref()
        .ok_or_else(|| AppError::ServiceError {
            status: StatusCode::SERVICE_UNAVAILABLE,
            message: "audit writer not configured".into(),
        })
}

/// The checks a value's shape cannot settle: whether the community, as it is
/// now, can live with it. Run on every runtime write path — `config/patch` and
/// the import — after [`validate_value`].
///
/// Today that is one key: an unrestricted-admin consent threshold the
/// community cannot meet is refused here, when it is written, rather than
/// discovered when it blocks a grant (VTI-APV-009).
async fn check_against_community(
    state: &AppState,
    key: &str,
    value: &Value,
) -> Result<(), AppError> {
    if key == crate::config_store::UNRESTRICTED_ADMIN_CONSENT_THRESHOLD
        && let Some(n) = value.as_u64()
    {
        crate::acl::admin_consent::check_threshold_meetable(state, n).await?;
    }
    // A community never witnesses (VTI-KEY-007): its own DID on its own list
    // would let it vouch for edges it is a party to.
    if key == crate::credentials::trusted_witnesses::WITNESSES_TRUSTED {
        let own = state.config.read().await.vtc_did.clone();
        crate::credentials::trusted_witnesses::parse_list(value, own.as_deref())?;
    }
    Ok(())
}

/// The threshold a write of `overrides` would **lower**
/// `acl.unrestricted_admin_consent_threshold` from, and to — `None` when it
/// raises it, leaves it alone, or names a value that would be refused anyway.
///
/// **VTI-APV-020**: lowering a consent threshold needs consent at the threshold
/// as it stood. Without it one unrestricted admin could turn N-of-M back into
/// 1-of-M alone, then promote at will (`vtc-action-list.md` §8.1, hole 2). Read
/// before the write, so `config/patch` and `vtc/config/import` can settle the
/// consent ([`crate::acl::admin_consent::Act::LowerThreshold`]) first. Raising
/// it stays immediate: it only ever asks for more agreement.
pub(crate) async fn lowered_threshold<'a>(
    state: &AppState,
    overrides: impl IntoIterator<Item = (&'a String, &'a Value)>,
) -> Result<Option<(u64, u64)>, AppError> {
    let key = crate::config_store::UNRESTRICTED_ADMIN_CONSENT_THRESHOLD;
    let Some((_, value)) = overrides.into_iter().find(|(k, _)| k.as_str() == key) else {
        return Ok(None);
    };
    // A value the write would reject is not a lowering; it is refused there.
    let Some(def) = lookup(key) else {
        return Ok(None);
    };
    if validate_value(def, value).is_err() {
        return Ok(None);
    }
    let Some(next) = value.as_u64() else {
        return Ok(None);
    };
    let current = crate::acl::admin_consent::threshold(state).await?;
    Ok((next < current).then_some((current, next)))
}

/// Read the live in-memory value for `key` out of an `AppConfig`.
/// Phase-0 keys only; unknown keys return `Value::Null`.
fn lookup_live(cfg: &crate::config::AppConfig, key: &str) -> Value {
    match key {
        "server.host" => Value::String(cfg.server.host.clone()),
        "server.port" => Value::Number(cfg.server.port.into()),
        "log.level" => Value::String(cfg.log.level.clone()),
        "auth.admin_idle_timeout" => Value::Number(cfg.auth.admin_idle_timeout.into()),
        crate::config_store::UNRESTRICTED_ADMIN_CONSENT_THRESHOLD => {
            Value::Number(cfg.acl.unrestricted_admin_consent_threshold.into())
        }
        crate::config_store::CONSENT_REQUEST_PUSH => Value::Bool(cfg.acl.consent_request_push),
        crate::credentials::trusted_witnesses::WITNESSES_TRUSTED => {
            serde_json::to_value(&cfg.witnesses.trusted).unwrap_or(Value::Null)
        }
        other => crate::config_store::action_key_live(cfg, other)
            .map(|n| Value::Number(n.into()))
            .unwrap_or(Value::Null),
    }
}

/// Apply `value` to the live in-memory `AppConfig` for `key`.
/// Returns `true` if the field changed (it should; the caller has
/// already diffed). Phase-0 keys only; unknown keys are a no-op.
///
/// **Phase 0 limitation**: this updates the field but does NOT
/// notify downstream subscribers (e.g., `tracing-subscriber`'s
/// reload Handle for `log.level`). Plumbing those subscribers is
/// a Phase-1 follow-up; for now the new value sticks for any
/// future reads of `state.config`, and `requires_restart`-flagged
/// keys (`server.*`) keep behaving correctly because they're never
/// touched here.
fn apply_to_live(cfg: &mut crate::config::AppConfig, key: &str, value: &Value) -> bool {
    // server.host / server.port are requires_restart and never reach
    // this function. Future hot-reloadable keys plug in alongside
    // `log.level` with their own arms.
    if key == "log.level"
        && let Some(s) = value.as_str()
    {
        cfg.log.level = s.to_string();
        return true;
    }
    // Unlike `log.level`, this one needs no subscriber plumbing to take
    // effect: `VtcAuthBackend::from_state` snapshots `state.config` on
    // every auth call, so the next refresh already measures against the
    // new value.
    if key == "auth.admin_idle_timeout"
        && let Some(n) = value.as_u64()
    {
        cfg.auth.admin_idle_timeout = n;
        return true;
    }
    // The gate reads the effective value itself (`admin_consent::threshold`),
    // so this only keeps the in-memory copy in step with what it enforces.
    if key == crate::config_store::UNRESTRICTED_ADMIN_CONSENT_THRESHOLD
        && let Some(n) = value.as_u64().filter(|n| *n >= 1)
    {
        cfg.acl.unrestricted_admin_consent_threshold = n;
        return true;
    }
    if key == crate::config_store::CONSENT_REQUEST_PUSH
        && let Some(b) = value.as_bool()
    {
        cfg.acl.consent_request_push = b;
        return true;
    }
    // Read live by `trusted_witnesses::live`; this keeps the in-memory copy in
    // step with what it enforces.
    if key == crate::credentials::trusted_witnesses::WITNESSES_TRUSTED
        && let Ok(list) =
            crate::credentials::trusted_witnesses::parse_list(value, cfg.vtc_did.as_deref())
    {
        cfg.witnesses.trusted = list;
        return true;
    }
    // The action-list settings are read live by `crate::admin_actions` too;
    // this keeps the in-memory copy in step.
    if let Some(n) = value.as_u64()
        && crate::config_store::set_action_key(cfg, key, n)
    {
        return true;
    }
    false
}

// ---------------------------------------------------------------------------
// Export
// ---------------------------------------------------------------------------

/// Schema version for the export/import envelope. Bumped on any
/// breaking shape change so the import handler can refuse old
/// exports cleanly. Phase 0 ships v1.
pub const EXPORT_SCHEMA_VERSION: u32 = 1;

/// The portable configuration document — canonical
/// `vtc/_shared/0.1/config-portability#ConfigExportDocument`.
///
/// `community_profile` is `None` when the community hasn't been
/// initialised yet (pre-bootstrap); the member is omitted rather
/// than emitted as `null`, because an import cannot distinguish a
/// synthesised empty profile from a real blank one.
/// `config_overrides` carries *only* the db-layer keys — env-layer
/// and toml-layer values stay per-host and aren't portable.
///
/// `schema_version` is not redundant with the `0.1` in the Trust
/// Task URI. The URI versions the *envelope*; the moment an operator
/// writes this document to a file the envelope is gone and it is
/// bare JSON, so `schemaVersion` is the only thing a later reader has
/// to check it against.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ConfigExportDocument {
    pub schema_version: u32,
    pub exported_at: DateTime<Utc>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub community_profile: Option<CommunityProfile>,
    pub config_overrides: HashMap<String, Value>,
    /// The published shape's extension point. Accepted and not
    /// interpreted: `deny_unknown_fields` refused it until #1641 batch 3,
    /// so a schema-valid document carrying `ext` was refused as malformed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ext: Option<Value>,
}

/// `vtc/config/export/0.1#response`. The document is returned under a
/// named member rather than as the bare body: the registry response
/// convention requires `additionalProperties: false` plus an `ext`
/// extension point, and neither attaches to a bare `$ref`.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ExportResponse {
    pub document: ConfigExportDocument,
}

/// The export, as the signed `vtc/config/export/0.1` document
/// (`trust_tasks::handle_config_export`) answers it.
///
/// There is no bearer route for this task. It declares `proof` REQUIRED, so
/// the signed document is its only binding (#1641 phase 2, batch 3); the
/// `POST /v1/admin/config/export` route it replaced had no client.
pub(crate) async fn export_inner(state: &AppState) -> Result<ExportResponse, AppError> {
    let community_profile = load_profile(&state.community_ks).await?;
    let store = ConfigStore::new(state.config_ks.clone());
    let config_overrides = store.snapshot().await?;

    Ok(ExportResponse {
        document: ConfigExportDocument {
            schema_version: EXPORT_SCHEMA_VERSION,
            exported_at: Utc::now(),
            community_profile,
            config_overrides,
            ext: None,
        },
    })
}

// ---------------------------------------------------------------------------
// Import (diff-and-confirm)
// ---------------------------------------------------------------------------

/// The `vtc/config/import/0.1` payload, as the import reads it.
///
/// `confirm` rides in the **payload**, not a query string: a Trust
/// Task is the same interface over REST, DIDComm and TSP, and only
/// one of those three has a query string to put it in.
///
/// It defaults to `false` because the safe direction of a default is
/// the one whose mistake is recoverable — a caller who meant to apply
/// and previewed loses a round-trip, where the reverse has already
/// overwritten a live community's configuration.
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct ImportRequest {
    pub document: ConfigExportDocument,
    #[serde(default)]
    pub confirm: bool,
    /// The task's extension point — accepted and not interpreted, for the
    /// reason given on [`ConfigExportDocument::ext`]. Declared so that
    /// `deny_unknown_fields` admits it, and never read.
    #[serde(default)]
    #[allow(dead_code)]
    pub ext: Option<Value>,
}

/// A single field an import would change, or did — canonical
/// `vtc/_shared/0.1/config-portability#ConfigFieldChange`.
///
/// `oldValue` is omitted when the key isn't currently set (no profile
/// yet, or no db-layer override). `newValue` is omitted when the
/// document leaves the field out, which means "leave the live value
/// alone" — distinct from an explicit `null`, which means "clear it".
/// Both are serialised as *absent* rather than `null` so that
/// distinction survives on the wire.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct FieldDiff {
    pub key: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub old_value: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub new_value: Option<Value>,
}

/// The import's response — canonical `vtc/config/import/0.1#response`.
///
/// One shape for both paths. On a preview the change arrays are what
/// *would* be written; on an apply they are what *was*. That is why
/// there are no separate `*Applied` lists: a caller reads `status` to
/// know which it is holding.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ImportResponse {
    /// `preview` when `confirm` was not set; `imported` after the
    /// document was applied.
    pub status: ImportStatus,
    /// Community-profile members that differ from what is in force.
    /// Empty when the document omits `communityProfile`, or when
    /// nothing differs.
    pub profile_changes: Vec<FieldDiff>,
    /// Configuration-override keys that differ from what is in force.
    /// Rejected keys are not listed here — they appear under
    /// `rejected`.
    pub override_changes: Vec<FieldDiff>,
    /// Applied keys whose new value takes effect only after a
    /// restart. Reported on a preview too, so an operator learns that
    /// the import implies downtime *before* confirming it.
    pub pending_restart: Vec<String>,
    /// Keys rejected by validation (unknown registry key, type
    /// mismatch, value-out-of-range, oversized extensions blob).
    /// Populated identically on preview and apply, so a preview
    /// surfaces every rejection before anything is written.
    pub rejected: Vec<RejectedKey>,
}

/// Whether an import response describes a dry run or a completed apply.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum ImportStatus {
    Preview,
    Imported,
}

/// `vtc/config/import:unsupportedSchemaVersion`.
pub const IMPORT_ERR_UNSUPPORTED_SCHEMA_VERSION: &str =
    trust_tasks_rs::specs::vtc::config::import::v0_1::error_codes::UNSUPPORTED_SCHEMA_VERSION.code;
/// `vtc/config/import:communityDidMismatch`.
pub const IMPORT_ERR_COMMUNITY_DID_MISMATCH: &str =
    trust_tasks_rs::specs::vtc::config::import::v0_1::error_codes::COMMUNITY_DID_MISMATCH.code;

/// The import — preview or apply — as the signed `vtc/config/import/0.1`
/// document (`trust_tasks::handle_config_import`) runs it. `actor_did` is the
/// document's verified signer, and it is what the audit rows name.
///
/// There is no bearer route for this task, for the reason given on
/// [`export_inner`].
pub(crate) async fn import_inner(
    state: &AppState,
    actor_did: &str,
    body: ImportRequest,
) -> Result<ImportResponse, crate::error::TaskError> {
    use crate::error::TaskError;
    let ImportRequest {
        document, confirm, ..
    } = body;
    let req = document;

    // Version first: a document whose shape we cannot vouch for must not be
    // diffed, because the diff would be against members we may be misreading.
    if req.schema_version != EXPORT_SCHEMA_VERSION {
        return Err(TaskError::declared(
            IMPORT_ERR_UNSUPPORTED_SCHEMA_VERSION,
            AppError::Validation(format!(
                "unsupported export schemaVersion: got {}, expected {EXPORT_SCHEMA_VERSION}",
                req.schema_version
            )),
        ));
    }

    // community_did mismatch is a 409 — refuse to clobber a
    // different community's profile with this import. A
    // freshly-installed VTC with no profile accepts any
    // community_did from the import.
    let current_profile = load_profile(&state.community_ks).await?;
    if let (Some(current), Some(incoming)) = (&current_profile, &req.community_profile)
        && current.community_did != incoming.community_did
    {
        return Err(TaskError::declared(
            IMPORT_ERR_COMMUNITY_DID_MISMATCH,
            AppError::Conflict(format!(
                "communityDid mismatch: current is {}, import carries {}",
                current.community_did, incoming.community_did
            )),
        ));
    }

    let store = ConfigStore::new(state.config_ks.clone());
    let current_overrides = store.snapshot().await?;

    // --- diff ---------------------------------------------------------
    let mut profile_diff: Vec<FieldDiff> = Vec::new();
    let mut overrides_diff: Vec<FieldDiff> = Vec::new();
    let mut rejected: Vec<RejectedKey> = Vec::new();

    if let Some(incoming) = &req.community_profile {
        for (key, old, new) in
            crate::community::profile::profile_field_pairs(current_profile.as_ref(), incoming)
        {
            if old != new {
                profile_diff.push(FieldDiff {
                    key: key.into(),
                    old_value: old,
                    new_value: new,
                });
            }
        }
    }

    for (key, new_value) in &req.config_overrides {
        let Some(def) = lookup(key) else {
            let reason = crate::config_store::host_only_refusal(key)
                .unwrap_or_else(|| "unknown config key (not in registry)".into());
            rejected.push(RejectedKey {
                key: key.clone(),
                reason,
            });
            continue;
        };
        if let Err(e) = validate_value(def, new_value) {
            rejected.push(RejectedKey {
                key: key.clone(),
                reason: format!("validation failed: {e}"),
            });
            continue;
        }
        if let Err(e) = check_against_community(state, key, new_value).await {
            rejected.push(RejectedKey {
                key: key.clone(),
                reason: format!("validation failed: {e}"),
            });
            continue;
        }
        let old = current_overrides.get(key).cloned();
        if old.as_ref() != Some(new_value) {
            overrides_diff.push(FieldDiff {
                key: key.clone(),
                old_value: old,
                new_value: Some(new_value.clone()),
            });
        }
    }

    // --- preview? -----------------------------------------------------
    // `pendingRestart` is reported here too: an operator should learn that
    // confirming implies downtime while they are still deciding, not after.
    if !confirm {
        let pending_restart = overrides_diff
            .iter()
            .filter(|d| matches!(lookup(&d.key), Some(def) if def.requires_restart))
            .map(|d| d.key.clone())
            .collect();
        return Ok(ImportResponse {
            status: ImportStatus::Preview,
            profile_changes: profile_diff,
            override_changes: overrides_diff,
            pending_restart,
            rejected,
        });
    }

    // --- apply --------------------------------------------------------
    // Profile first so a partial failure on overrides doesn't leave
    // half a profile applied. Overrides are persisted one key at a
    // time, so a fjall-side error mid-loop reports back via
    // `rejected` rather than aborting.
    let audit_writer = require_audit_writer(state)?;

    let community_profile_applied = if let Some(incoming) = req.community_profile.clone() {
        apply_profile_import(state, incoming, current_profile.as_ref()).await?
    } else {
        Vec::new()
    };

    let mut config_overrides_applied: Vec<String> = Vec::new();
    let mut pending_restart: Vec<String> = Vec::new();
    let mut audit_changes: Vec<ConfigChange> = Vec::new();
    let mut requires_restart = false;
    for FieldDiff {
        key,
        old_value,
        new_value,
    } in &overrides_diff
    {
        let Some(def) = lookup(key) else { continue };
        let new_value = match new_value {
            Some(v) => v.clone(),
            None => continue,
        };
        if let Err(e) = store.put(key, &new_value).await {
            rejected.push(RejectedKey {
                key: key.clone(),
                reason: format!("persistence failed: {e}"),
            });
            continue;
        }
        config_overrides_applied.push(key.clone());
        crate::admin_actions::record_effect(state).await;
        let mut change = ConfigChange {
            key: key.clone(),
            old_value: old_value.clone(),
            new_value,
            source_before: if old_value.is_some() {
                ConfigSource::Db
            } else {
                ConfigSource::Default
            },
        };
        change.redact_if(|k| matches!(lookup(k), Some(d) if d.sensitive));
        audit_changes.push(change);
        if def.trust_semantics {
            warn!(
                key = %key,
                trust_semantics = true,
                "a configuration key that alters trust semantics was imported (VTI-CMP-090)"
            );
        }
        if def.requires_restart {
            requires_restart = true;
            pending_restart.push(key.clone());
        }
    }

    if !audit_changes.is_empty() {
        audit_writer
            .write(
                actor_did,
                None,
                AuditEvent::ConfigChanged(ConfigChangedData {
                    changes: audit_changes,
                    requires_restart,
                }),
            )
            .await?;
    }
    if !community_profile_applied.is_empty() {
        audit_writer
            .write(
                actor_did,
                None,
                AuditEvent::CommunityProfileUpdated(CommunityProfileUpdatedData {
                    fields_changed: community_profile_applied.clone(),
                    // Reuse the diff already computed above for the
                    // import preview — each FieldDiff is a before/after.
                    changes: profile_diff
                        .iter()
                        .map(|d| vti_common::audit::FieldChange {
                            field: d.key.clone(),
                            old: d.old_value.clone(),
                            new: d.new_value.clone(),
                        })
                        .collect(),
                }),
            )
            .await?;
    }

    info!(
        profile_changed = community_profile_applied.len(),
        overrides_applied = config_overrides_applied.len(),
        rejected = rejected.len(),
        "config imported"
    );

    // On `imported` the change arrays carry what was actually written, not
    // what was proposed. A key that made it into `overrides_diff` but then
    // failed to persist has already been pushed to `rejected` above; leaving
    // it in `overrideChanges` would report it as applied.
    let profile_changes: Vec<FieldDiff> = profile_diff
        .iter()
        .filter(|d| community_profile_applied.contains(&d.key))
        .cloned()
        .collect();
    let override_changes: Vec<FieldDiff> = overrides_diff
        .into_iter()
        .filter(|d| config_overrides_applied.contains(&d.key))
        .collect();

    Ok(ImportResponse {
        status: ImportStatus::Imported,
        profile_changes,
        override_changes,
        pending_restart,
        rejected,
    })
}

/// Apply the incoming profile to `community_ks`. Returns the list of
/// field names that changed (driving the
/// `CommunityProfileUpdated.fieldsChanged` audit payload).
///
/// Every import goes through `CommunityProfileUpdate::apply`, so every
/// imported profile meets the caps an edit must: the text-length bounds, the
/// 16 KiB `extensions` bound, the `http(s)`-only `logoUrl` and governance URL,
/// the personhood overclaim refusal. Those caps exist because the profile is
/// served on the unauthenticated public-profile endpoint and `logoUrl` lands
/// in an `<img src>` there.
///
/// When no profile is stored yet, the patch is applied to a default profile
/// for the incoming community DID — keeping the imported `createdAt` — and
/// the result is stored even if nothing differs from the defaults, because
/// the community had no profile at all. This path used to store the import
/// verbatim, which skipped every one of those caps: an import was the one way
/// to publish a `javascript:` logo URL.
async fn apply_profile_import(
    state: &AppState,
    incoming: CommunityProfile,
    current: Option<&CommunityProfile>,
) -> Result<Vec<String>, AppError> {
    let (mut updated, fresh) = match current {
        Some(current) => (current.clone(), false),
        None => {
            let mut base = CommunityProfile::new(incoming.community_did.clone(), "");
            base.created_at = incoming.created_at;
            (base, true)
        }
    };
    let patch = CommunityProfileUpdate {
        name: Some(incoming.name),
        description: Some(incoming.description),
        logo_url: Some(incoming.logo_url),
        public_url: Some(incoming.public_url),
        contact_email: Some(incoming.contact_email),
        language: Some(incoming.language),
        relationship_identifier_default: Some(incoming.relationship_identifier_default),
        personhood: Some(incoming.personhood),
        extensions: Some(incoming.extensions),
    };
    let changed = patch.apply(&mut updated)?;
    if fresh || !changed.is_empty() {
        store_profile(&state.community_ks, &updated).await?;
    }
    Ok(changed)
}

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

#[cfg(test)]
mod tests {
    // Behavioural coverage lives in `tests/admin_config.rs` — those
    // exercise the full router stack (Trust-Task header, AdminAuth
    // extractor, JSON body, three-layer effective view) via
    // `Router::oneshot`. Unit tests for the overlay + validation
    // semantics live in `crate::config_store::tests`.
}