did-hosting-server 0.10.1

use std::sync::{Arc, OnceLock};
use std::time::Duration;

use affinidi_did_resolver_cache_sdk::DIDCacheClient;
use affinidi_messaging_didcomm_service::{
    DIDCommService, DIDCommServiceConfig, ListenerConfig, ListenerEvent, Protocols, RestartPolicy,
    RetryConfig,
};
use affinidi_tdk::secrets_resolver::ThreadedSecretsResolver;
use axum::routing::get;
use did_hosting_common::server::domain::parse_trusted_cidrs;
use did_hosting_common::server::store::KS_DIDS;
use ipnetwork::IpNetwork;

use did_hosting_common::server::didcomm_profile::{
    advertised_protocols, build_tdk_profile_for_identity, reconcile_listener_protocols,
    wait_for_did_resolution,
};
use did_hosting_common::server::identity::{self, ServiceIdentity};
use did_hosting_common::server::init;
use tokio_util::sync::CancellationToken;

use crate::config::{AppConfig, AuthConfig};
use crate::control_register;
use crate::did_ops::cleanup_empty_dids;
use crate::error::AppError;
use crate::messaging;
use crate::routes;
use crate::secret_store::ServerSecrets;
use crate::stats;
use crate::store::{KeyspaceHandle, Store};
use tokio::sync::{oneshot, watch};
use tower_http::trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer};
use tracing::{Level, debug, error, info, warn};

#[derive(Clone)]
pub struct AppState {
    pub store: Store,
    pub dids_ks: KeyspaceHandle,
    pub config: Arc<AppConfig>,
    pub did_resolver: Option<DIDCacheClient>,
    /// The proof verifier for control-plane documents, built once over
    /// `did_resolver` so its DID-status verdicts and forced-refresh rate limit
    /// persist across messages. `None` when no resolver is configured, in which
    /// case no control-plane document can be accepted.
    pub trust_tasks_verifier:
        Option<Arc<did_hosting_common::server::trust_tasks::TransportBoundVerifier>>,
    pub secrets_resolver: Option<Arc<ThreadedSecretsResolver>>,
    /// The service's own DID identity: every generation of key material still
    /// honoured, and the kids each one answers to. `did_resolver` and
    /// `secrets_resolver` above are cheap clones taken from this; it is the
    /// source of truth for which kids they answer to, and the listener's
    /// profile is built from it so the two cannot drift apart.
    pub identity: Option<Arc<ServiceIdentity>>,
    /// The running messaging service, once started.
    ///
    /// Lifted into `AppState` (it used to be a local in `run()`, unreachable at
    /// runtime) so listeners can be hot-swapped without bouncing the process:
    /// `remove_listener` / `add_listener` take `&self`, which is what lets a
    /// rotation rebuild the profile in place. A `OnceLock` suffices precisely
    /// because the *service* is never replaced — only its listeners are.
    pub didcomm_service: Arc<OnceLock<DIDCommService>>,
    pub stats_collector: Option<Arc<stats::StatsCollector>>,
    /// In-memory cache for DID content (did.jsonl). TTL-based eviction on read.
    pub did_cache: Arc<crate::cache::ContentCache>,
    /// Parsed `server.trusted_proxy_cidrs` — peers inside this set
    /// have their `Forwarded` / `X-Forwarded-Host` headers honoured
    /// for request-host detection (multi-domain, T19/T21).
    pub trusted_proxy_cidrs: Arc<Vec<IpNetwork>>,
    /// How far this edge is behind its control plane — the reconcile loop
    /// writes it, `/api/health` and `server/metrics` read it.
    pub replication: Arc<crate::replication::ReplicationStatus>,
    /// Per-address rate limiter for `POST /api/trust-tasks`. Applied before
    /// the document's claimed issuer is checked against `control_did`, so a
    /// single address cannot force unbounded DID-resolution work by posting
    /// documents that name a fresh issuer on every request.
    pub trust_tasks_rate_limiter: Arc<did_hosting_common::server::rate_limit::IpRateLimiter>,
    /// Serialises `sync/update`, `sync/batch` and `sync/delete`: each reads
    /// what this edge holds for a slot, checks the new log extends it (or, for
    /// a delete, reads the record to remove), and writes — two such
    /// operations running at once could each pass their read-check against
    /// the same stale state. Mirrors the watcher's `sync_lock`.
    pub sync_lock: Arc<tokio::sync::Mutex<()>>,
}

impl AppState {
    /// An edge over `store` with nothing wired but its configuration and a
    /// fresh content cache: no identity, no resolver, no messaging service.
    /// The caller — `run`, the daemon, a test — fills in what it has.
    pub fn new(store: Store, config: AppConfig) -> Result<Self, AppError> {
        let (parsed_cidrs, bad_cidrs) = parse_trusted_cidrs(&config.server.trusted_proxy_cidrs);
        if !bad_cidrs.is_empty() {
            warn!(
                bad_cidrs = ?bad_cidrs,
                "server.trusted_proxy_cidrs contains unparseable entries; ignoring them"
            );
        }
        Ok(Self {
            dids_ks: store.keyspace(KS_DIDS)?,
            store,
            config: Arc::new(config),
            did_resolver: None,
            trust_tasks_verifier: None,
            secrets_resolver: None,
            identity: None,
            didcomm_service: Arc::new(OnceLock::new()),
            stats_collector: None,
            did_cache: Arc::new(crate::cache::ContentCache::new(Duration::from_secs(300))),
            trusted_proxy_cidrs: Arc::new(parsed_cidrs),
            replication: Arc::new(crate::replication::ReplicationStatus::new(
                did_hosting_common::server::auth::session::now_epoch(),
            )),
            trust_tasks_rate_limiter: Arc::new(
                did_hosting_common::server::rate_limit::IpRateLimiter::new(
                    did_hosting_common::server::rate_limit::TRUST_TASKS_RATE_LIMIT_NAME,
                    did_hosting_common::server::rate_limit::TRUST_TASKS_MAX_PER_WINDOW,
                    did_hosting_common::server::rate_limit::TRUST_TASKS_WINDOW_SECS,
                ),
            ),
            sync_lock: Arc::new(tokio::sync::Mutex::new(())),
        })
    }

    /// Carry `identity`: the resolvers and proof verifier are taken from it.
    pub fn with_identity(mut self, identity: Option<Arc<ServiceIdentity>>) -> Self {
        self.did_resolver = identity.as_ref().map(|i| i.did_resolver.clone());
        self.secrets_resolver = identity.as_ref().map(|i| i.secrets_resolver.clone());
        self.trust_tasks_verifier = crate::messaging::build_verifier(self.did_resolver.as_ref());
        self.identity = identity;
        self
    }
}

/// Refuse to start an edge that no control plane drives.
///
/// An edge is a cache of its control plane: every write it applies is a Trust
/// Task that control plane signed, and it has no management surface of its
/// own. Without `control_did` nothing could ever write to it, so a
/// single-host deployment runs `did-hosting-daemon` instead.
pub fn require_control_plane(config: &AppConfig) -> Result<(), AppError> {
    config.replication.validate()?;
    if config.control_did.as_deref().is_none_or(str::is_empty) {
        return Err(AppError::Config(
            "control_did is not set. did-hosting-server is an edge that a control plane drives \
             over Trust Tasks, and does not run without one; for a single-host deployment run \
             did-hosting-daemon"
                .into(),
        ));
    }
    if config.server_did.as_deref().is_none_or(str::is_empty) {
        return Err(AppError::Config(
            "server_did is not set: the control plane addresses and verifies this edge by it"
                .into(),
        ));
    }
    Ok(())
}

pub async fn run(config: AppConfig, store: Store, secrets: ServerSecrets) -> Result<(), AppError> {
    require_control_plane(&config)?;
    let dids_ks = store.keyspace(KS_DIDS)?;

    // Integrity check on DID keyspace
    match dids_ks.verify_integrity().await {
        Ok(0) => debug!("store integrity check passed"),
        Ok(n) => warn!(
            corrupted = n,
            "store integrity check found corrupted entries"
        ),
        Err(e) => warn!(error = %e, "store integrity check failed"),
    }

    // First-boot multi-domain init (daemon parity with
    // did-hosting-daemon/src/main.rs:544-622). Three idempotent steps:
    //
    //   1. seed_domains_first_boot — populates KS_DOMAINS from
    //      `[hosting] bootstrap_domains`, falling back to legacy
    //      `public_url` host on upgrade. Without this, the
    //      resolve-side safety check (assert_resolution_allowed) takes
    //      the "permissive skip" branch on every resolve and emits a
    //      warn-log per request — annoying noise on standalone v0.6
    //      → v0.7 upgrades.
    //   2. seed_assignments_first_boot — same tier chain, populates
    //      KS_ASSIGNMENTS so this server knows which domains it
    //      serves (matters once a control plane starts driving
    //      replica/domain/assign / unassign).
    //   3. MigrationRunner::run_pending — runs T13 M-01 which fills
    //      DidRecord.domain for legacy records by parsing the
    //      embedded did_id host. Required for write-side checks
    //      that key on domain once domains are configured.
    //
    // All three are idempotent (markers in `meta` keyspace gate the
    // migration; existing keyspace entries short-circuit the seeds),
    // so re-running on every boot is cheap.
    run_first_boot_init(&store, &config).await;
    // Auto-bootstrap DIDs if public_url is set and they don't exist yet
    let config = auto_bootstrap_dids(config, &store, &dids_ks, &secrets).await;

    // Load the service's own identity (requires server_did). Resolves the DID
    // document for the *real* verification-method key IDs and seeds the secrets
    // resolver under them, rather than assuming `#key-0` / `#key-1`.
    let identity = identity::load_identity(
        config.server_did.as_deref(),
        config.mediator_did.as_deref(),
        identity::ProtocolSet {
            didcomm: config.features.didcomm,
            tsp: config.features.tsp,
        },
        &secrets,
        &store,
    )
    .await;
    // Bind TCP before anything else: public resolution and the Trust Task
    // listener's HTTPS binding are served from it.
    let addr = format!("{}:{}", config.server.host, config.server.port);
    let std_listener = {
        let listener = std::net::TcpListener::bind(&addr).map_err(AppError::Io)?;
        listener.set_nonblocking(true).map_err(AppError::Io)?;
        info!("server listening addr={addr}");
        listener
    };

    // Gather storage thread inputs before moving config into Arc
    let storage_dids_ks = dids_ks.clone();
    let storage_auth_config = config.auth.clone();

    let upload_body_limit = config.limits.upload_body_limit;
    let stats_config = config.stats.clone();

    // Initialize in-memory stats collector (starts at zero — no disk persistence)
    let stats_collector = {
        let collector = stats::StatsCollector::new();
        let total_dids = storage_dids_ks
            .prefix_iter_raw("did:")
            .await
            .map(|v| v.len())
            .unwrap_or(0) as u64;
        collector.set_total_dids(total_dids);
        info!(total_dids, "stats collector initialized");
        Arc::new(collector)
    };

    let mut state = AppState::new(store.clone(), config)?.with_identity(identity);
    state.stats_collector = Some(stats_collector.clone());

    // Log startup configuration
    info!("--- enabled services ---");
    info!(
        "  DIDComm  : {}",
        if state.config.features.didcomm {
            "enabled"
        } else {
            "disabled"
        }
    );
    if let Some(ref did) = state.config.server_did {
        info!("  server DID   : {did}");
    }
    if let Some(ref did) = state.config.mediator_did {
        info!("  mediator DID : {did}");
    }

    // Separate shutdown channels for ordered shutdown (DIDComm → REST → Storage)
    let (rest_shutdown_tx, rest_shutdown_rx) = watch::channel(false);
    let (storage_shutdown_tx, storage_shutdown_rx) = watch::channel(false);

    // REST ready signal — DIDComm waits for this before starting
    let (rest_ready_tx, rest_ready_rx) = oneshot::channel::<()>();

    // 1. Spawn HTTP thread first — must be serving before DIDComm starts
    //    (the server's own DID needs to be resolvable for mediator auth)
    let mut rest_shutdown = rest_shutdown_rx.clone();
    let rest_state = state.clone();
    let rest_handle = std::thread::Builder::new()
        .name("webvh-rest".into())
        .spawn(move || {
            run_rest_thread(
                std_listener,
                rest_state,
                upload_body_limit,
                &mut rest_shutdown,
                rest_ready_tx,
            )
        })
        .map_err(|e| AppError::Internal(format!("failed to spawn HTTP thread: {e}")))?;

    // 2. Spawn storage thread (independent cleanup, flush, sync)
    let mut storage_shutdown = storage_shutdown_rx.clone();
    let storage_collector = stats_collector.clone();
    let storage_stats_config = stats_config;
    let storage_handle = std::thread::Builder::new()
        .name("webvh-storage".into())
        .spawn(move || {
            run_storage_thread(
                StorageThreadParams {
                    store,
                    dids_ks: storage_dids_ks,
                    auth_config: storage_auth_config,
                    collector: storage_collector,
                    stats_config: storage_stats_config,
                },
                &mut storage_shutdown,
            )
        })
        .map_err(|e| AppError::Internal(format!("failed to spawn storage thread: {e}")))?;

    // 3. Wait for REST to be serving before starting DIDComm
    //    (the server's own DID needs to be resolvable for mediator auth)
    let _ = rest_ready_rx.await;

    // Now that we are serving HTTP, resolve our *own* DID for the first time.
    // A self-hosting service cannot resolve its own DID at boot — it is the thing
    // that serves it — so `load_identity` came up on guessed `#key-0`/`#key-1`
    // kids and persisted nothing. This is the first moment the document is
    // fetchable, and it must run before the listener is built on the guess.
    if let Err(e) = crate::identity_rotation::reload_now(&state).await {
        warn!("failed to establish the service identity from its DID document: {e}");
    }

    // 4. Start the mediator messaging service (DIDComm and/or TSP) — one
    //    connection for both receiving and sending. TSP-only deployments
    //    (didcomm off, tsp on) must still start it.
    let didcomm_shutdown = CancellationToken::new();
    if state.config.features.didcomm || state.config.features.tsp {
        match start_didcomm_service(&state, didcomm_shutdown.clone()).await {
            Ok(Some(svc)) => {
                let _ = state.didcomm_service.set(svc);
            }
            Ok(None) => {}
            Err(e) => warn!("failed to start DIDComm service: {e}"),
        }
    }
    let didcomm_service = state.didcomm_service.get();

    // 5. Register with control plane via DIDComm (uses the shared connection)
    if let Some(svc) = didcomm_service
        && state.config.control_did.is_some()
    {
        let reg_state = state.clone();
        let reg_svc = svc.clone();
        tokio::spawn(async move {
            keep_registered(reg_state, reg_svc).await;
        });
    }

    // 5a. Reconcile against the control plane's listing every interval — the
    //     backstop behind the outbox, and what the staleness bound measures.
    let (reconcile_shutdown_tx, reconcile_shutdown_rx) = watch::channel(false);
    if let Some(svc) = didcomm_service {
        let rec_state = state.clone();
        let rec_svc = svc.clone();
        tokio::spawn(async move {
            crate::replication::run_reconcile_loop(rec_state, rec_svc, reconcile_shutdown_rx).await;
        });
    } else {
        warn!(
            "no messaging service: this edge cannot reconcile with its control plane, and will \
             report degraded once the staleness bound passes"
        );
    }

    // 5b. Keep a TSP relationship with the control plane established. Under Rev 3
    //     §7.2.2 the control plane's application pushes (sync/health) are dropped
    //     unless this edge holds a relationship with it, and forming one is the
    //     edge's job — the control plane persists its half and does not re-invite.
    //     The ensure is idempotent (skips when already Bidirectional) and MUST
    //     fire on every (re)connect, not just first boot: a control-plane restart,
    //     or a handshake that never completed, is repaired on the edge's next
    //     connect. See docs/tsp-transport.md.
    if let (Some(svc), Some(control_did)) = (didcomm_service, state.config.control_did.as_ref())
        && state.config.features.tsp
    {
        let svc = svc.clone();
        let control_did = control_did.clone();
        tokio::spawn(async move {
            ensure_control_tsp_relationship(svc, control_did).await;
        });
    }

    // 6. Spawn stats sync task — TSP or DIDComm, per the control DID (main runtime)
    let stats_sync_shutdown = CancellationToken::new();
    let didcomm_sync_interval = state.config.stats.sync_interval_secs;
    if let (Some(svc), Some(control_did), Some(server_did)) = (
        didcomm_service,
        state.config.control_did.as_ref(),
        state.config.server_did.as_ref(),
    ) && didcomm_sync_interval > 0
    {
        let token = stats_sync_shutdown.clone();
        let svc = svc.clone();
        let control_did = control_did.clone();
        let server_did = server_did.clone();
        let collector = stats_collector.clone();
        let sync_state = state.clone();
        tokio::spawn(async move {
            let mut timer =
                tokio::time::interval(Duration::from_secs(didcomm_sync_interval.max(1)));
            timer.tick().await; // skip first tick
            loop {
                tokio::select! {
                    _ = timer.tick() => {
                        stats::sync_to_control_messaging(
                            &svc,
                            &sync_state,
                            &server_did,
                            &control_did,
                            &collector,
                        ).await;
                    }
                    _ = token.cancelled() => break,
                }
            }
        });
    }

    // 6b. Reconnect to any mediator we rotated *away* from but whose grace period
    // has not elapsed. Peers holding a stale DID document are still delivering
    // there, and a restart part-way through the window must not abandon that
    // queue. No-op unless a rotation actually changed the mediator.
    crate::identity_rotation::resume_mediator_drains(&state);

    // 7. Spawn the identity sweep. Expires generations whose grace period has
    // elapsed (so a superseded key stops decrypting) and backstops identity
    // changes that never came through our own publish or sync path — one
    // applied out-of-band, or while this process was down.
    let (identity_shutdown_tx, identity_shutdown_rx) = watch::channel(false);
    let identity_state = state.clone();
    let identity_handle = tokio::spawn(async move {
        crate::identity_rotation::run_identity_sweep_loop(identity_state, identity_shutdown_rx)
            .await;
    });

    // Wait for shutdown signal
    init::shutdown_signal().await;

    // Ordered shutdown: identity → stats sync → DIDComm → REST → Storage
    let mut any_panic = false;

    let _ = identity_shutdown_tx.send(true);
    if let Err(e) = identity_handle.await {
        warn!("identity sweep task didn't shut down cleanly: {e}");
    }

    let _ = reconcile_shutdown_tx.send(true);
    stats_sync_shutdown.cancel();
    didcomm_shutdown.cancel();
    if let Some(svc) = didcomm_service {
        svc.shutdown().await;
        info!("DIDComm service stopped");
    }

    let _ = rest_shutdown_tx.send(true);
    {
        let handle = rest_handle;
        match tokio::time::timeout(
            Duration::from_secs(30),
            tokio::task::spawn_blocking(move || handle.join()),
        )
        .await
        {
            Ok(Ok(Ok(()))) => info!("REST thread stopped"),
            Ok(Ok(Err(_panic))) => {
                error!("REST thread panicked");
                any_panic = true;
            }
            Ok(Err(e)) => {
                error!("failed to join REST thread: {e}");
                any_panic = true;
            }
            Err(_) => {
                error!("REST thread shutdown timed out (30s)");
                any_panic = true;
            }
        }
    }

    let _ = storage_shutdown_tx.send(true);
    match tokio::time::timeout(
        Duration::from_secs(30),
        tokio::task::spawn_blocking(move || storage_handle.join()),
    )
    .await
    {
        Ok(Ok(Ok(()))) => info!("storage thread stopped"),
        Ok(Ok(Err(_panic))) => {
            error!("storage thread panicked");
            any_panic = true;
        }
        Ok(Err(e)) => {
            error!("failed to join storage thread: {e}");
            any_panic = true;
        }
        Err(_) => {
            error!("storage thread shutdown timed out (30s)");
            any_panic = true;
        }
    }

    if any_panic {
        return Err(AppError::Internal("one or more threads panicked".into()));
    }

    info!("server shut down");
    Ok(())
}

// ---------------------------------------------------------------------------
// DIDComm service startup
// ---------------------------------------------------------------------------

/// The listener id [`start_didcomm_service`] registers for the edge's mediator
/// connection. Must match, or the relationship ensure targets a listener that
/// does not exist.
const SERVER_LISTENER_ID: &str = "server";

/// Establish, and keep re-establishing, this edge's TSP relationship with its
/// control plane.
///
/// Under Rev 3 §7.2.2 the control plane's application pushes are dropped unless
/// this edge holds a relationship with it. Forming one is a control message the
/// edge must actually *send* — a relationship never forms as a side effect of an
/// application send, so this has to run explicitly, and on **every** (re)connect
/// before the first send, not just first boot: the durable store keeps a formed
/// relationship across our own restarts, but a control-plane restart, or a
/// handshake that never completed, is only repaired when the edge re-invites on
/// its next connection.
///
/// [`DIDCommService::tsp_ensure_relationship`] is idempotent (it skips when the
/// relationship already admits application messages), so re-running it every
/// connect is cheap and a duplicate for the initial connect is harmless.
/// How often an edge re-registers with its control plane even when nothing
/// else prompts it. Registration reports the DIDs this edge holds, and the
/// control plane answers with whatever it is missing or behind on (and deletes
/// for what it should no longer serve) — so this bounds how long an edge can
/// serve stale state after missing pushes.
pub const RESYNC_INTERVAL: Duration = Duration::from_secs(10 * 60);

/// Register with the control plane at startup, again on every mediator
/// reconnect, and every [`RESYNC_INTERVAL`] — each registration is a delta
/// re-sync request.
async fn keep_registered(state: AppState, svc: DIDCommService) {
    let mut events = svc.subscribe();
    control_register::register_via_didcomm(&state, &svc).await;
    let mut timer = tokio::time::interval(RESYNC_INTERVAL);
    timer.tick().await; // the first tick is immediate; registration just ran
    loop {
        tokio::select! {
            _ = timer.tick() => {}
            event = events.recv() => match event {
                Ok(ListenerEvent::Connected { listener_id }) if listener_id == SERVER_LISTENER_ID => {}
                Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            },
        }
        control_register::register_via_didcomm(&state, &svc).await;
    }
}

async fn ensure_control_tsp_relationship(svc: DIDCommService, control_did: String) {
    const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);

    // Subscribe before the initial wait so a reconnect that races the first
    // ensure is not missed.
    let mut events = svc.subscribe();

    // The initial connection may have completed before `subscribe`, so its
    // `Connected` event can be absent from the stream — cover it explicitly.
    if svc
        .wait_connected(SERVER_LISTENER_ID, CONNECT_TIMEOUT)
        .await
        .is_ok()
    {
        ensure_control_tsp_relationship_once(&svc, &control_did).await;
    }

    loop {
        match events.recv().await {
            Ok(ListenerEvent::Connected { listener_id }) if listener_id == SERVER_LISTENER_ID => {
                ensure_control_tsp_relationship_once(&svc, &control_did).await;
            }
            Ok(_) => {}
            // A lagged receiver only means we missed some events; the next
            // `Connected` still re-ensures, and the relationship is idempotent.
            Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
            // The service shut down; nothing more to ensure.
            Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
        }
    }
}

async fn ensure_control_tsp_relationship_once(svc: &DIDCommService, control_did: &str) {
    match svc
        .tsp_ensure_relationship(SERVER_LISTENER_ID, control_did)
        .await
    {
        Ok(()) => {
            debug!(control = %control_did, "TSP: relationship with control plane ensured")
        }
        Err(e) => warn!(
            control = %control_did,
            error = %e,
            "TSP: failed to ensure relationship with control plane — control-plane pushes may be \
             dropped until the next reconnect"
        ),
    }
}

pub async fn start_didcomm_service(
    state: &AppState,
    shutdown: CancellationToken,
) -> Result<Option<DIDCommService>, AppError> {
    let identity = match state.identity.as_ref() {
        Some(identity) => identity,
        None => {
            info!("DIDComm not configured — server_did not set");
            return Ok(None);
        }
    };

    let mediator_did = match identity.mediator_did() {
        Some(did) => did,
        None => {
            info!("mediator_did not configured — DIDComm messaging disabled");
            return Ok(None);
        }
    };

    // Block until the mediator DID document is resolvable. On a cold start
    // the mediator DID may be hosted by a control plane / did-hosting-server that
    // has not yet published its log, so we retry instead of starting the
    // listener against an unreachable mediator (which surfaces as a cryptic
    // "No Mediator is configured for this Profile" later).
    wait_for_did_resolution(&mediator_did, "mediator", &identity.did_resolver, &shutdown).await?;

    // Carries the key material of every live generation, keyed on the kids the
    // DID document actually resolved to — the same kids the secrets resolver
    // was seeded with.
    let profile = build_tdk_profile_for_identity("server", identity, Some(&mediator_did)).await?;

    // Transport selection — DIDComm and/or TSP ride the same mediator
    // socket. Inbound TSP frames (sync/domain pushes from the control
    // plane's outbox) are routed to the `ServerTspHandler` when TSP is on.
    // "neither flag set but a mediator is configured" defaults to
    // DIDComm-only for back-compat.
    //
    // The union across live generations, not just the current one's config
    // flags — a generation retiring out of DIDComm still has peers delivering
    // to it until it expires, and the single listener has to carry both.
    //
    // Then reconcile against what our own DID document actually advertises: the
    // document is authoritative for how peers reach us, so the listener must
    // carry every transport it advertises. Without this, a node whose config
    // disables a transport its DID still advertises (e.g. `features.tsp = false`
    // while the DID carries `TSPTransport`, since `did-host-didcomm` advertises
    // both) silently drops — and acks — every inbound frame on that transport.
    let advertised = advertised_protocols(&identity.did, Some(&identity.did_resolver)).await;
    let transports = reconcile_listener_protocols(identity.protocols(), advertised, &identity.did);
    let didcomm_enabled = transports.didcomm;
    let tsp_enabled = transports.tsp;
    let protocols = match (didcomm_enabled, tsp_enabled) {
        (true, true) => Protocols::BOTH,
        (false, true) => Protocols::TSP_ONLY,
        _ => Protocols::DIDCOMM_ONLY,
    };

    let mut listener = ListenerConfig {
        id: "server".into(),
        profile,
        restart_policy: RestartPolicy::Always {
            backoff: RetryConfig::default(),
        },
        auto_delete: true,
        protocols,
        ..Default::default()
    };
    // Persist this listener's TSP relationships (Rev 3 §7.2.2) across restarts.
    // The default store is in-memory, so without this a restart forgets every
    // peer and then silently drops each peer's application traffic — the peers
    // still hold the relationship the restarted node forgot — until a
    // re-handshake. Only meaningful when TSP is on. See docs/tsp-transport.md.
    if tsp_enabled {
        listener.relationship_store = Some(
            did_hosting_common::server::tsp_relationship_store::build_relationship_store(
                &state.store,
            )?,
        );
    }

    let router = messaging::build_server_router(state.clone())
        .map_err(|e| AppError::Internal(format!("failed to build DIDComm router: {e}")))?;

    let config = DIDCommServiceConfig {
        listeners: vec![listener],
    };

    let svc = if tsp_enabled {
        info!("starting messaging service with DIDComm + TSP on the mediator connection");
        DIDCommService::start_with_tsp(
            config,
            router,
            crate::tsp::ServerTspHandler::new(state.clone()),
            shutdown,
        )
        .await
    } else {
        info!("starting DIDComm service with mediator connection");
        DIDCommService::start(config, router, shutdown).await
    }
    .map_err(|e| AppError::Internal(format!("failed to start messaging service: {e}")))?;

    info!(
        tsp = tsp_enabled,
        server_did = %identity.did,
        "messaging service started"
    );
    Ok(Some(svc))
}

// ---------------------------------------------------------------------------
// REST thread
// ---------------------------------------------------------------------------

fn run_rest_thread(
    std_listener: std::net::TcpListener,
    state: AppState,
    upload_body_limit: usize,
    shutdown_rx: &mut watch::Receiver<bool>,
    ready_tx: oneshot::Sender<()>,
) {
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)
        .enable_all()
        .build()
        .expect("failed to build REST runtime");

    rt.block_on(async {
        let listener = tokio::net::TcpListener::from_std(std_listener)
            .expect("failed to convert std TcpListener to tokio TcpListener");

        // Public DID resolution and the Trust Task listener's HTTPS binding.
        // There is no management surface to switch on.
        info!("HTTP thread started (public DID serving + Trust Task listener)");
        let base_router = routes::router(upload_body_limit);

        // Cloned before `with_state` consumes `state`: the health route is
        // added below the router's state type has already collapsed to
        // `()` (deliberately, so it sits outside the CORS/security-header
        // layers), so it closes over its own `AppState` rather than using
        // axum's `State` extractor.
        let health_state = state.clone();
        let app = base_router
            .with_state(state)
            .layer(
                TraceLayer::new_for_http()
                    .make_span_with(DefaultMakeSpan::new().level(Level::DEBUG))
                    .on_response(
                        DefaultOnResponse::new()
                            .level(Level::DEBUG)
                            .latency_unit(tower_http::LatencyUnit::Millis),
                    ),
            )
            .layer(axum::middleware::from_fn(
                did_hosting_common::server::security_headers,
            ))
            // Allow browser-based resolvers to fetch public DID documents
            // cross-origin. Read-only, unauthenticated, wildcard origin.
            .layer(did_hosting_common::server::public_resolution_cors())
            .route(
                "/api/health",
                get(move || {
                    let health_state = health_state.clone();
                    async move { routes::health::health(health_state).await }
                }),
            );

        // Signal that REST is ready to serve
        let _ = ready_tx.send(());

        let shutdown_rx = shutdown_rx.clone();
        axum::serve(
            listener,
            app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
        )
        .with_graceful_shutdown(async move {
            let mut rx = shutdown_rx;
            let _ = rx.changed().await;
        })
        .await
        .expect("axum serve failed");

        info!("REST thread shutting down");
    });
}

// ---------------------------------------------------------------------------
// Storage thread
// ---------------------------------------------------------------------------

/// Parameters for the background storage thread.
struct StorageThreadParams {
    store: Store,
    dids_ks: KeyspaceHandle,
    auth_config: AuthConfig,
    collector: Arc<stats::StatsCollector>,
    stats_config: crate::config::StatsConfig,
}

fn run_storage_thread(params: StorageThreadParams, shutdown_rx: &mut watch::Receiver<bool>) {
    let StorageThreadParams {
        store,
        dids_ks,
        auth_config,
        collector,
        stats_config,
    } = params;
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("failed to build storage runtime");

    rt.block_on(async {
        info!(
            flush_interval = stats_config.flush_interval_secs,
            "storage thread started"
        );

        let did_ttl_seconds = auth_config.cleanup_ttl_minutes * 60;
        let did_interval = Duration::from_secs(did_ttl_seconds.max(60));
        let mut did_timer = tokio::time::interval(did_interval);

        // First tick completes immediately; skip so cleanup doesn't run at startup
        did_timer.tick().await;

        loop {
            tokio::select! {
                _ = did_timer.tick() => {
                    match cleanup_empty_dids(&dids_ks, did_ttl_seconds).await {
                        Ok(0) => {}
                        Ok(n) => {
                            info!(count = n, "cleaned up empty DID records");
                            for _ in 0..n {
                                collector.decrement_total_dids();
                            }
                        }
                        Err(e) => warn!("DID cleanup error: {e}"),
                    }
                }
                _ = shutdown_rx.changed() => {
                    info!("storage thread shutting down");
                    break;
                }
            }
        }

        // Persist store before closing
        if let Err(e) = store.persist().await {
            error!("failed to persist store on shutdown: {e}");
        } else {
            info!("store persisted");
        }
    });
}

// ---------------------------------------------------------------------------
// Auto-bootstrap
// ---------------------------------------------------------------------------

/// Extract the mnemonic (path) from a `did:webvh` DID string.
///
/// Re-exported from `did_hosting_common::server::identity`, which needs the same
/// parse to answer "was the DID just published our own?" — the question that
/// gates the rotation trigger. One copy, so the two cannot drift.
use did_hosting_common::server::identity::mnemonic_from_did;

/// First-boot multi-domain init for the standalone server.
///
/// Mirrors the daemon's `did-hosting-daemon/src/main.rs:544-622` block
/// so a standalone `did-hosting-server` deployment gets the same
/// effective state on first boot. Three steps, in order:
///
/// 1. **Domain catalog seed** (`seed_domains_first_boot`) — populates
///    `KS_DOMAINS` from `[hosting] bootstrap_domains` or the legacy
///    `public_url` host. Without it, the resolve-side safety check
///    (`assert_resolution_allowed`) takes its permissive skip branch
///    on every resolve and emits a per-request warn-log.
/// 2. **Assignment seed** (`seed_assignments_first_boot`) — same tier
///    chain, populates `KS_ASSIGNMENTS`. Matters once a control plane
///    starts driving `replica/domain/assign` / unassign messages.
/// 3. **Migration runner** (`MigrationRunner::run_pending`) — runs
///    `m01_tag_did_records_with_domain` which fills
///    `DidRecord.domain` for legacy records by parsing the embedded
///    `did_id` host. Required for any write path that keys on
///    `domain` once domains are configured.
///
/// All three are idempotent — subsequent boots short-circuit
/// (existing keyspace entries / migration markers in `meta`) so this
/// is cheap to call on every startup.
///
/// Failures log loudly but do **not** abort startup. The standalone
/// server's design treats domain seeding as best-effort metadata: a
/// legacy single-domain deployment without `bootstrap_domains`
/// configured boots into the permissive-skip state (with warnings)
/// rather than refusing to serve. The daemon's stricter
/// `std::process::exit(1)` is appropriate there because the daemon
/// owns the control plane and *must* have a domain catalog to make
/// admin write paths work; the standalone server has no such
/// guarantee.
pub async fn run_first_boot_init(store: &Store, config: &AppConfig) {
    // 1. Domain catalog
    match did_hosting_common::server::domain::seed_domains_first_boot(
        store,
        &config.hosting.bootstrap_domains,
        config.public_url.as_deref(),
    )
    .await
    {
        Ok(outcome) => info!(
            tier = ?outcome.tier,
            count = outcome.final_count,
            default = ?outcome.default,
            "first-boot domain seed"
        ),
        Err(e) => warn!(error = %e, "first-boot domain seed failed"),
    }

    // 2. Per-server assignments
    let now = did_hosting_common::server::auth::session::now_epoch();
    match did_hosting_common::server::assignment_seed::seed_assignments_first_boot(
        store,
        &config.hosting.bootstrap_domains,
        config.public_url.as_deref(),
        now,
    )
    .await
    {
        Ok(outcome) => info!(
            tier = ?outcome.tier,
            count = outcome.final_count,
            "first-boot assignment seed"
        ),
        Err(e) => warn!(error = %e, "first-boot assignment seed failed"),
    }

    // 3. Storage migrations (T2 runner + T13 M-01)
    let runner = did_hosting_common::server::migrations::MigrationRunner::new(
        did_hosting_common::server::migrations::registry(),
    );
    match runner.run_pending(store).await {
        Ok(summary) => info!(
            applied = ?summary.applied,
            skipped = ?summary.skipped,
            "migration runner complete"
        ),
        Err(e) => warn!(error = %e, "migration runner failed"),
    }
}

/// Verify DIDs in the store and log their status.
///
/// The server is a read-only edge node — it does not auto-create DIDs.
/// The setup wizard or `bootstrap-did` CLI command creates the server's
/// identity DID. This function only verifies and logs what's present.
pub async fn auto_bootstrap_dids(
    config: AppConfig,
    _store: &Store,
    dids_ks: &KeyspaceHandle,
    _secrets: &ServerSecrets,
) -> AppConfig {
    // Verify server_did exists in store (if configured)
    if let Some(ref server_did) = config.server_did
        && let Some(mnemonic) = mnemonic_from_did(server_did)
    {
        let exists = dids_ks
            .contains_key(format!("did:{mnemonic}"))
            .await
            .unwrap_or(false);
        if exists {
            info!(path = %mnemonic, "server DID loaded from store");
        } else {
            error!(
                did = %server_did,
                path = %mnemonic,
                "server DID not found in store — DID resolution will fail"
            );
            error!("  To create it, run:  did-hosting-server bootstrap-did --path {mnemonic}");
            error!("  Then update server_did in config.toml with the new DID");
        }
    }

    // Log all DIDs present in the store
    match dids_ks.prefix_iter_raw("did:").await {
        Ok(entries) => {
            let count = entries.len();
            if count == 0 {
                warn!("no DIDs in store — run `did-hosting-server bootstrap-did` to create one");
            } else {
                info!(count, "DIDs in store:");
                for (key, _) in &entries {
                    let mnemonic = String::from_utf8_lossy(key)
                        .strip_prefix("did:")
                        .unwrap_or("?")
                        .to_string();
                    info!(path = %mnemonic, "  DID loaded");
                }
            }
        }
        Err(e) => warn!("failed to list DIDs: {e}"),
    }

    config
}

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

    #[test]
    fn mnemonic_from_did_simple() {
        let did =
            "did:webvh:QmaErmPvnHUDaaiM4phkDgrK58T49cxgmCUtKon9gwyWtJ:webvh.storm.ws:webvh:server1";
        assert_eq!(mnemonic_from_did(did).unwrap(), "webvh/server1");
    }

    #[test]
    fn mnemonic_from_did_single_path() {
        let did = "did:webvh:QmABC:example.com:my-did";
        assert_eq!(mnemonic_from_did(did).unwrap(), "my-did");
    }

    #[test]
    fn mnemonic_from_did_deep_path() {
        let did = "did:webvh:QmABC:example.com:people:staff:glenn";
        assert_eq!(mnemonic_from_did(did).unwrap(), "people/staff/glenn");
    }

    /// A standalone edge with no control plane is not a supported shape: the
    /// server refuses to start rather than run with nothing able to write to it.
    #[test]
    fn an_edge_without_a_control_plane_does_not_start() {
        let config = |control: Option<&str>, server: Option<&str>| {
            let mut c: AppConfig = toml::from_str("").expect("defaults parse");
            c.control_did = control.map(String::from);
            c.server_did = server.map(String::from);
            c
        };
        for (control, server) in [
            (None, Some("did:key:edge")),
            (Some(""), Some("did:key:edge")),
            (Some("did:key:control"), None),
        ] {
            let err = require_control_plane(&config(control, server)).unwrap_err();
            assert!(matches!(err, AppError::Config(_)), "{err:?}");
        }
        let err = require_control_plane(&config(None, Some("did:key:edge"))).unwrap_err();
        assert!(err.to_string().contains("did-hosting-daemon"), "{err}");
        require_control_plane(&config(Some("did:key:control"), Some("did:key:edge")))
            .expect("an edge with a control plane starts");
    }

    #[test]
    fn mnemonic_from_did_invalid() {
        assert!(mnemonic_from_did("did:web:example.com").is_none());
        assert!(mnemonic_from_did("not-a-did").is_none());
    }
}