//! HTTP management adapter for [`wasmeld_runtime`].
//!
//! The console persists component artifacts and manifests on the local
//! filesystem. Service metadata, invocation metrics, and events are persisted
//! to libSQL through Toasty. It also hosts an immutable filesystem-backed WIT
//! package Registry. Actor memory remains process-local and is never presented
//! as durable state.

mod persistence;
mod wit_registry;

use std::{
    collections::{BTreeMap, VecDeque},
    fs, io,
    path::{Path, PathBuf},
    sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard},
    time::{Instant, SystemTime, UNIX_EPOCH},
};

use axum::{
    Json, Router,
    body::{Body, Bytes},
    extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State, rejection::BytesRejection},
    http::{HeaderMap, HeaderValue, Method, StatusCode, header},
    response::{IntoResponse, Response},
    routing::{get, post},
};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use serde::{Deserialize, Serialize};
use serde_json::json;
use thiserror::Error;
use tower_http::{
    cors::CorsLayer,
    trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer},
};
use tracing::Level;
use wasmeld_package::{
    ComponentPackage, PackageError, read_package,
    wit_package::{WitPackageError, WitPackageMetadata},
};
use wasmeld_runtime::{
    CapabilityDescriptor, ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey,
    ServiceManifest,
};

use persistence::{Persistence, StoredDeployment, StoredEvent, StoredService};
use wit_registry::WitRegistry;
pub use wit_registry::WitRegistryError;

const MAX_EVENTS: usize = 256;
const DEFAULT_MAX_ARTIFACT_BYTES: usize = 64 * 1024 * 1024;
const DEFAULT_MAX_WIT_PACKAGE_BYTES: usize = 4 * 1024 * 1024;
type PersistenceSnapshot = (Vec<StoredService>, Vec<StoredEvent>, Vec<StoredDeployment>);

/// Filesystem, persistence, upload, and Runtime policy used by [`Console`].
#[derive(Clone, Debug)]
pub struct ConsoleConfig {
    /// Directory for registered Component binaries and generated manifests.
    pub artifact_dir: PathBuf,
    /// Directory for immutable binary WIT Registry artifacts.
    pub wit_registry_dir: PathBuf,
    /// Local libSQL database file.
    pub database_path: PathBuf,
    /// Maximum accepted `.wasmpkg` and uncompressed Component size.
    pub max_artifact_bytes: usize,
    /// Maximum accepted binary WIT package size.
    pub max_wit_package_bytes: usize,
    /// Platform-owned limits injected into every registered service.
    pub registration_limits: ResourceLimits,
    /// Wasmtime Engine and epoch settings.
    pub runtime: RuntimeConfig,
}

impl Default for ConsoleConfig {
    fn default() -> Self {
        Self {
            artifact_dir: PathBuf::from("var/wasmeld/components"),
            wit_registry_dir: PathBuf::from("var/wasmeld/wit-packages"),
            database_path: PathBuf::from("var/wasmeld/console.db"),
            max_artifact_bytes: DEFAULT_MAX_ARTIFACT_BYTES,
            max_wit_package_bytes: DEFAULT_MAX_WIT_PACKAGE_BYTES,
            registration_limits: ResourceLimits::default(),
            runtime: RuntimeConfig::default(),
        }
    }
}

/// Management state that owns one optional Runtime and its durable control data.
///
/// Runtime stop/start operations replace the in-process Wasmtime state while
/// registered service metadata, artifacts, metrics, and events remain managed
/// by the Console.
pub struct Console {
    runtime: RwLock<Option<Runtime>>,
    runtime_config: RuntimeConfig,
    artifact_dir: PathBuf,
    wit_registry: WitRegistry,
    max_artifact_bytes: usize,
    registration_limits: ResourceLimits,
    lifecycle: Mutex<()>,
    state: Mutex<ConsoleState>,
    persistence: Persistence,
    persistence_sync: tokio::sync::Mutex<()>,
}

struct ConsoleState {
    runtime_started_at: Option<Instant>,
    next_event_id: u64,
    services: BTreeMap<ServiceKey, ServiceRecord>,
    deployments: BTreeMap<String, DeploymentRecord>,
    events: VecDeque<EventView>,
}

#[derive(Clone)]
struct ServiceRecord {
    manifest: ServiceManifest,
    capabilities: Vec<CapabilityView>,
    status: ServiceStatus,
    updated_at_ms: u64,
    calls: u64,
    errors: u64,
    last_latency_ms: Option<u64>,
}

#[derive(Clone)]
struct DeploymentRecord {
    active_revision: String,
    updated_at_ms: u64,
}

/// Lifecycle status exposed for one registered service version.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ServiceStatus {
    Running,
    Stopped,
    Faulted,
}

/// API projection of one public service deployment.
#[derive(Clone, Debug, Serialize)]
pub struct DeploymentView {
    pub service_id: String,
    pub active_revision: String,
    pub status: ServiceStatus,
    pub updated_at_ms: u64,
}

/// Exact versioned host interface imported by a registered Component.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct CapabilityView {
    pub interface: String,
    pub package: String,
    pub name: String,
    pub version: String,
}

impl From<CapabilityDescriptor> for CapabilityView {
    fn from(capability: CapabilityDescriptor) -> Self {
        Self {
            interface: capability.interface().to_owned(),
            package: capability.package().to_owned(),
            name: capability.name().to_owned(),
            version: capability.version().to_owned(),
        }
    }
}

/// API projection of a registered service and its current metrics.
#[derive(Clone, Debug, Serialize)]
pub struct ServiceView {
    pub id: String,
    pub revision: String,
    pub artifact: String,
    pub world: String,
    pub capabilities: Vec<CapabilityView>,
    pub status: ServiceStatus,
    pub updated_at_ms: u64,
    pub limits: ResourceLimits,
    pub calls: u64,
    pub errors: u64,
    pub last_latency_ms: Option<u64>,
}

impl ServiceRecord {
    fn view(&self) -> ServiceView {
        ServiceView {
            id: self.manifest.id.clone(),
            revision: self.manifest.revision.clone(),
            artifact: self.manifest.component.clone(),
            world: self.manifest.world.clone(),
            capabilities: self.capabilities.clone(),
            status: self.status,
            updated_at_ms: self.updated_at_ms,
            limits: self.manifest.limits.clone(),
            calls: self.calls,
            errors: self.errors,
            last_latency_ms: self.last_latency_ms,
        }
    }
}

/// One persisted Console lifecycle or invocation event.
#[derive(Clone, Debug, Serialize)]
pub struct EventView {
    pub id: u64,
    pub timestamp_ms: u64,
    pub kind: EventKind,
    pub service_id: Option<String>,
    pub revision: Option<String>,
    pub message: String,
}

/// Event category serialized by the management API.
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EventKind {
    Registered,
    Deployed,
    Started,
    Stopped,
    Invoked,
    Failed,
}

impl EventKind {
    fn as_database_value(self) -> &'static str {
        match self {
            Self::Registered => "registered",
            Self::Deployed => "deployed",
            Self::Started => "started",
            Self::Stopped => "stopped",
            Self::Invoked => "invoked",
            Self::Failed => "failed",
        }
    }

    fn from_database_value(value: &str) -> Result<Self, ConsoleError> {
        match value {
            "registered" => Ok(Self::Registered),
            "deployed" => Ok(Self::Deployed),
            "started" => Ok(Self::Started),
            "stopped" => Ok(Self::Stopped),
            "invoked" => Ok(Self::Invoked),
            "failed" => Ok(Self::Failed),
            _ => Err(ConsoleError::InvalidDatabaseData(format!(
                "unsupported event kind {value:?}"
            ))),
        }
    }
}

impl TryFrom<StoredEvent> for EventView {
    type Error = ConsoleError;

    fn try_from(event: StoredEvent) -> Result<Self, Self::Error> {
        Ok(Self {
            id: event.id,
            timestamp_ms: event.timestamp_ms,
            kind: EventKind::from_database_value(&event.kind)?,
            service_id: event.service_id,
            revision: event.revision,
            message: event.message,
        })
    }
}

#[derive(Debug, Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct StartRequest {
    init_config_base64: Option<String>,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct InvokeRequest {
    input_base64: String,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct ActivateDeploymentRequest {
    revision: String,
    init_config_base64: Option<String>,
}

#[derive(Debug, Serialize)]
struct ServiceList {
    services: Vec<ServiceView>,
}

#[derive(Debug, Serialize)]
struct DeploymentList {
    deployments: Vec<DeploymentView>,
}

#[derive(Debug, Serialize)]
struct EventList {
    events: Vec<EventView>,
}

#[derive(Debug, Serialize)]
struct WitPackageList {
    packages: Vec<WitPackageMetadata>,
}

#[derive(Debug, Serialize)]
struct RuntimeView {
    status: RuntimeStatus,
    uptime_ms: u64,
    managed_services: usize,
    registered_services: usize,
    running_services: usize,
}

#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
enum RuntimeStatus {
    Running,
    Stopped,
}

#[derive(Debug, Serialize)]
struct InvokeResponse {
    output_base64: String,
    output_bytes: usize,
    latency_ms: u64,
}

struct InvocationResult {
    output: Vec<u8>,
    latency_ms: u64,
}

struct GatewayInvocation {
    revision: String,
    result: InvocationResult,
}

/// Errors raised by Console state, persistence, Registry, or Runtime operations.
#[derive(Debug, Error)]
pub enum ConsoleError {
    #[error("{0}")]
    InvalidRequest(String),

    #[error("service revision {0} is not managed by Wasmeld")]
    ServiceNotFound(ServiceKey),

    #[error("service {0} has no active deployment")]
    DeploymentNotFound(String),

    #[error("Wasmeld Runtime is not running")]
    RuntimeNotRunning,

    #[error("artifact exceeds the {limit}-byte limit")]
    ArtifactTooLarge { limit: usize },

    #[error(transparent)]
    Package(#[from] PackageError),

    #[error(transparent)]
    WitRegistry(#[from] WitRegistryError),

    #[error("failed to access {path}: {source}")]
    Storage {
        path: PathBuf,
        #[source]
        source: io::Error,
    },

    #[error("failed to serialize manifest: {0}")]
    ManifestSerialize(#[from] toml::ser::Error),

    #[error("database operation failed: {0}")]
    Database(#[from] toasty::Error),

    #[error("database contains invalid console data: {0}")]
    InvalidDatabaseData(String),

    #[error(transparent)]
    Runtime(#[from] RuntimeError),

    #[error("blocking runtime task failed: {0}")]
    Join(#[from] tokio::task::JoinError),

    #[error("console state lock was poisoned")]
    LockPoisoned,
}

impl Console {
    /// Opens durable stores, restores control data, and starts a fresh Runtime.
    ///
    /// Actors selected by durable deployments are recreated with empty init
    /// configuration. Their previous Store and linear memory are not restored.
    pub async fn new(config: ConsoleConfig) -> Result<Self, ConsoleError> {
        config.registration_limits.validate()?;
        fs::create_dir_all(&config.artifact_dir).map_err(|source| ConsoleError::Storage {
            path: config.artifact_dir.clone(),
            source,
        })?;
        let artifact_dir =
            fs::canonicalize(&config.artifact_dir).map_err(|source| ConsoleError::Storage {
                path: config.artifact_dir.clone(),
                source,
            })?;
        if let Some(parent) = config.database_path.parent() {
            fs::create_dir_all(parent).map_err(|source| ConsoleError::Storage {
                path: parent.to_path_buf(),
                source,
            })?;
        }
        let persistence = Persistence::open(&config.database_path).await?;
        let wit_registry =
            WitRegistry::open(&config.wit_registry_dir, config.max_wit_package_bytes)?;
        let (stored_services, mut stored_events, stored_deployments) = persistence.load().await?;
        stored_events.sort_unstable_by_key(|event| std::cmp::Reverse(event.id));
        stored_events.truncate(MAX_EVENTS);
        let next_event_id = stored_events
            .first()
            .map(|event| event.id.saturating_add(1))
            .unwrap_or(1);
        let events = stored_events
            .into_iter()
            .map(EventView::try_from)
            .collect::<Result<VecDeque<_>, _>>()?;

        let runtime = Runtime::new(config.runtime.clone())?;
        let console = Self {
            runtime: RwLock::new(Some(runtime)),
            runtime_config: config.runtime,
            artifact_dir,
            wit_registry,
            max_artifact_bytes: config.max_artifact_bytes,
            registration_limits: config.registration_limits,
            lifecycle: Mutex::new(()),
            state: Mutex::new(ConsoleState {
                runtime_started_at: Some(Instant::now()),
                next_event_id,
                services: BTreeMap::new(),
                deployments: BTreeMap::new(),
                events,
            }),
            persistence,
            persistence_sync: tokio::sync::Mutex::new(()),
        };
        // The database is authoritative for metrics. Filesystem manifests are
        // then used to recover artifacts that predate or missed a DB snapshot.
        console.load_stored_services(stored_services)?;
        console.load_manifest_services()?;
        console.load_stored_deployments(stored_deployments)?;
        console.restore_deployments()?;
        console.persist_snapshot().await?;
        Ok(console)
    }

    /// Returns the configured component upload limit.
    pub fn max_artifact_bytes(&self) -> usize {
        self.max_artifact_bytes
    }

    /// Returns the configured binary WIT package upload limit.
    pub fn max_wit_package_bytes(&self) -> usize {
        self.wit_registry.max_package_bytes()
    }

    /// Returns the maximum raw request size accepted by the Gateway.
    pub fn max_gateway_input_bytes(&self) -> usize {
        self.registration_limits.max_input_bytes
    }

    /// Returns registered service versions in stable key order.
    pub fn services(&self) -> Result<Vec<ServiceView>, ConsoleError> {
        let state = self.state()?;
        Ok(state.services.values().map(ServiceRecord::view).collect())
    }

    /// Returns recent events from newest to oldest.
    pub fn events(&self) -> Result<Vec<EventView>, ConsoleError> {
        let state = self.state()?;
        Ok(state.events.iter().cloned().collect())
    }

    /// Returns active deployments in stable service id order.
    pub fn deployments(&self) -> Result<Vec<DeploymentView>, ConsoleError> {
        let state = self.state()?;
        state
            .deployments
            .iter()
            .map(|(service_id, deployment)| deployment_view(&state, service_id, deployment))
            .collect()
    }

    /// Returns the active deployment for one public service id.
    pub fn deployment(&self, service_id: &str) -> Result<DeploymentView, ConsoleError> {
        let state = self.state()?;
        let deployment = state
            .deployments
            .get(service_id)
            .ok_or_else(|| ConsoleError::DeploymentNotFound(service_id.to_owned()))?;
        deployment_view(&state, service_id, deployment)
    }

    fn runtime_view(&self) -> Result<RuntimeView, ConsoleError> {
        let runtime = self.runtime()?;
        let state = self.state()?;
        match runtime.as_ref() {
            Some(runtime) => {
                let stats = runtime.stats()?;
                Ok(RuntimeView {
                    status: RuntimeStatus::Running,
                    uptime_ms: state
                        .runtime_started_at
                        .map(|started_at| duration_ms(started_at.elapsed()))
                        .unwrap_or_default(),
                    managed_services: state.services.len(),
                    registered_services: stats.registered_services,
                    running_services: stats.running_actors,
                })
            }
            None => Ok(RuntimeView {
                status: RuntimeStatus::Stopped,
                uptime_ms: 0,
                managed_services: state.services.len(),
                registered_services: 0,
                running_services: 0,
            }),
        }
    }

    fn start_runtime(&self) -> Result<RuntimeView, ConsoleError> {
        let _lifecycle = self
            .lifecycle
            .lock()
            .map_err(|_| ConsoleError::LockPoisoned)?;
        let mut runtime = self
            .runtime
            .write()
            .map_err(|_| ConsoleError::LockPoisoned)?;

        if runtime.is_some() {
            drop(runtime);
            return self.runtime_view();
        }

        match self.build_runtime() {
            Ok(started) => {
                *runtime = Some(started);
                drop(runtime);
                self.mark_runtime_started()?;
                self.push_event(
                    EventKind::Started,
                    None,
                    "Wasmeld Runtime started".to_owned(),
                )?;
                self.runtime_view()
            }
            Err(error) => {
                drop(runtime);
                self.push_event(
                    EventKind::Failed,
                    None,
                    format!("Wasmeld Runtime failed to start: {error}"),
                )?;
                Err(error)
            }
        }
    }

    fn stop_runtime(&self) -> Result<RuntimeView, ConsoleError> {
        let _lifecycle = self
            .lifecycle
            .lock()
            .map_err(|_| ConsoleError::LockPoisoned)?;
        let mut runtime = self
            .runtime
            .write()
            .map_err(|_| ConsoleError::LockPoisoned)?;
        let Some(running) = runtime.take() else {
            drop(runtime);
            return self.runtime_view();
        };
        let stopped = running.stop_all();
        drop(runtime);
        self.mark_runtime_stopped()?;

        match stopped {
            Ok(()) => {
                self.push_event(
                    EventKind::Stopped,
                    None,
                    "Wasmeld Runtime stopped".to_owned(),
                )?;
                self.runtime_view()
            }
            Err(error) => {
                self.push_event(
                    EventKind::Failed,
                    None,
                    format!("Wasmeld Runtime stopped with an actor shutdown error: {error}"),
                )?;
                Err(error.into())
            }
        }
    }

    fn restart_runtime(&self) -> Result<RuntimeView, ConsoleError> {
        let _lifecycle = self
            .lifecycle
            .lock()
            .map_err(|_| ConsoleError::LockPoisoned)?;
        let mut runtime = self
            .runtime
            .write()
            .map_err(|_| ConsoleError::LockPoisoned)?;

        if let Some(running) = runtime.take()
            && let Err(error) = running.stop_all()
        {
            drop(runtime);
            self.mark_runtime_stopped()?;
            self.push_event(
                EventKind::Failed,
                None,
                format!("Wasmeld Runtime failed to restart while stopping actors: {error}"),
            )?;
            return Err(error.into());
        }

        match self.build_runtime() {
            Ok(started) => {
                *runtime = Some(started);
                drop(runtime);
                self.mark_runtime_started()?;
                self.push_event(
                    EventKind::Started,
                    None,
                    "Wasmeld Runtime restarted".to_owned(),
                )?;
                self.runtime_view()
            }
            Err(error) => {
                drop(runtime);
                self.mark_runtime_stopped()?;
                self.push_event(
                    EventKind::Failed,
                    None,
                    format!("Wasmeld Runtime failed to restart: {error}"),
                )?;
                Err(error)
            }
        }
    }

    fn build_runtime(&self) -> Result<Runtime, ConsoleError> {
        let state = self.state()?;
        let manifests = state
            .services
            .values()
            .map(|service| service.manifest.clone())
            .collect::<Vec<_>>();
        let deployments = state
            .deployments
            .iter()
            .map(|(service_id, deployment)| {
                ServiceKey::new(service_id.clone(), deployment.active_revision.clone())
            })
            .collect::<Result<Vec<_>, _>>()?;
        drop(state);

        let runtime = Runtime::new(self.runtime_config.clone())?;
        for manifest in manifests {
            runtime.register_from_file(manifest)?;
        }
        for key in deployments {
            runtime.start(&key, Vec::new())?;
        }
        Ok(runtime)
    }

    fn mark_runtime_started(&self) -> Result<(), ConsoleError> {
        let mut state = self.state()?;
        let deployed = state
            .deployments
            .iter()
            .map(|(service_id, deployment)| {
                ServiceKey::new(service_id.clone(), deployment.active_revision.clone())
            })
            .collect::<Result<Vec<_>, _>>()?;
        let updated_at_ms = unix_time_ms();
        state.runtime_started_at = Some(Instant::now());
        for (key, service) in &mut state.services {
            let status = if deployed.contains(key) {
                ServiceStatus::Running
            } else {
                ServiceStatus::Stopped
            };
            if service.status != status {
                service.status = status;
                service.updated_at_ms = updated_at_ms;
            }
        }
        Ok(())
    }

    fn mark_runtime_stopped(&self) -> Result<(), ConsoleError> {
        let mut state = self.state()?;
        let updated_at_ms = unix_time_ms();
        state.runtime_started_at = None;
        for service in state.services.values_mut() {
            if !matches!(service.status, ServiceStatus::Stopped) {
                service.status = ServiceStatus::Stopped;
                service.updated_at_ms = updated_at_ms;
            }
        }
        Ok(())
    }

    fn register_package(&self, package_bytes: Vec<u8>) -> Result<ServiceView, ConsoleError> {
        if package_bytes.len() > self.max_artifact_bytes {
            return Err(ConsoleError::ArtifactTooLarge {
                limit: self.max_artifact_bytes,
            });
        }
        let ComponentPackage {
            manifest: package_manifest,
            component: component_bytes,
        } = read_package(&package_bytes, self.max_artifact_bytes)?;

        validate_path_segment("id", &package_manifest.id)?;
        validate_path_segment("revision", &package_manifest.revision)?;
        let key = ServiceKey::new(
            package_manifest.id.clone(),
            package_manifest.revision.clone(),
        )?;
        let _lifecycle = self
            .lifecycle
            .lock()
            .map_err(|_| ConsoleError::LockPoisoned)?;
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;

        if self.state()?.services.contains_key(&key) {
            return Err(RuntimeError::ServiceAlreadyRegistered(key).into());
        }

        let basename = format!("{}@{}", package_manifest.id, package_manifest.revision);
        let component_path = self.artifact_dir.join(format!("{basename}.wasm"));
        let manifest_path = self.artifact_dir.join(format!("{basename}.toml"));
        let manifest = ServiceManifest {
            id: package_manifest.id,
            revision: package_manifest.revision,
            component: component_path.display().to_string(),
            world: package_manifest.world,
            limits: self.registration_limits.clone(),
        };
        manifest.validate()?;

        write_atomic(&component_path, &component_bytes)?;
        let manifest_toml = toml::to_string_pretty(&manifest)?;
        if let Err(error) = write_atomic(&manifest_path, manifest_toml.as_bytes()) {
            let _ = fs::remove_file(&component_path);
            return Err(error);
        }

        if let Err(error) = runtime.register(manifest.clone(), &component_bytes) {
            let _ = fs::remove_file(&component_path);
            let _ = fs::remove_file(&manifest_path);
            return Err(error.into());
        }
        let capabilities = runtime_capabilities(runtime, &key)?;

        let view = self.insert_service(manifest, capabilities, ServiceStatus::Stopped)?;
        self.push_event(
            EventKind::Registered,
            Some(&key),
            "component package registered".to_owned(),
        )?;
        Ok(view)
    }

    fn start(&self, key: &ServiceKey, init_config: Vec<u8>) -> Result<ServiceView, ConsoleError> {
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
        self.ensure_managed(key)?;
        runtime.start(key, init_config)?;
        let view = self.update_status(key, ServiceStatus::Running)?;
        self.push_event(EventKind::Started, Some(key), "actor started".to_owned())?;
        Ok(view)
    }

    fn stop(&self, key: &ServiceKey) -> Result<ServiceView, ConsoleError> {
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
        self.ensure_managed(key)?;
        runtime.stop(key)?;
        let view = self.update_status(key, ServiceStatus::Stopped)?;
        self.push_event(EventKind::Stopped, Some(key), "actor stopped".to_owned())?;
        Ok(view)
    }

    fn restart(&self, key: &ServiceKey, init_config: Vec<u8>) -> Result<ServiceView, ConsoleError> {
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
        self.ensure_managed(key)?;
        match runtime.stop(key) {
            Ok(()) | Err(RuntimeError::ActorUnavailable(_)) => {}
            Err(error) => return Err(error.into()),
        }
        runtime.start(key, init_config)?;
        let view = self.update_status(key, ServiceStatus::Running)?;
        self.push_event(EventKind::Started, Some(key), "actor restarted".to_owned())?;
        Ok(view)
    }

    fn activate_deployment(
        &self,
        service_id: String,
        revision: String,
        init_config: Vec<u8>,
    ) -> Result<DeploymentView, ConsoleError> {
        validate_path_segment("id", &service_id)?;
        validate_path_segment("revision", &revision)?;
        let key = ServiceKey::new(service_id.clone(), revision.clone())?;
        let _lifecycle = self
            .lifecycle
            .lock()
            .map_err(|_| ConsoleError::LockPoisoned)?;
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
        self.ensure_managed(&key)?;

        // Starting is idempotent. The old revision stays resident so calls
        // that resolved immediately before this switch can finish safely and
        // rollback does not require recompilation.
        runtime.start(&key, init_config)?;

        let updated_at_ms = unix_time_ms();
        let view = {
            let mut state = self.state()?;
            let service = state
                .services
                .get_mut(&key)
                .ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?;
            service.status = ServiceStatus::Running;
            service.updated_at_ms = updated_at_ms;
            let deployment = DeploymentRecord {
                active_revision: revision,
                updated_at_ms,
            };
            state
                .deployments
                .insert(service_id.clone(), deployment.clone());
            deployment_view(&state, &service_id, &deployment)?
        };
        self.push_event(
            EventKind::Deployed,
            Some(&key),
            "deployment activated".to_owned(),
        )?;
        Ok(view)
    }

    fn invoke(&self, key: &ServiceKey, input: Vec<u8>) -> Result<InvocationResult, ConsoleError> {
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
        self.ensure_managed(key)?;
        let started_at = Instant::now();
        let result = runtime.invoke(key, input);
        let latency_ms = duration_ms(started_at.elapsed());

        let mut state = self.state()?;
        let record = state
            .services
            .get_mut(key)
            .ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?;
        record.calls = record.calls.saturating_add(1);
        record.last_latency_ms = Some(latency_ms);
        record.updated_at_ms = unix_time_ms();

        match result {
            Ok(output) => {
                drop(state);
                self.push_event(
                    EventKind::Invoked,
                    Some(key),
                    format!("invoke completed in {latency_ms} ms"),
                )?;
                Ok(InvocationResult { output, latency_ms })
            }
            Err(error) => {
                record.errors = record.errors.saturating_add(1);
                if matches!(error, RuntimeError::ActorFault { .. }) {
                    record.status = ServiceStatus::Faulted;
                }
                drop(state);
                self.push_event(EventKind::Failed, Some(key), error.to_string())?;
                Err(error.into())
            }
        }
    }

    fn gateway_invoke(
        &self,
        service_id: &str,
        input: Vec<u8>,
    ) -> Result<GatewayInvocation, ConsoleError> {
        let revision = {
            let state = self.state()?;
            state
                .deployments
                .get(service_id)
                .map(|deployment| deployment.active_revision.clone())
                .ok_or_else(|| ConsoleError::DeploymentNotFound(service_id.to_owned()))?
        };
        let key = ServiceKey::new(service_id.to_owned(), revision.clone())?;
        let result = self.invoke(&key, input)?;
        Ok(GatewayInvocation { revision, result })
    }

    fn load_stored_services(
        &self,
        stored_services: Vec<StoredService>,
    ) -> Result<(), ConsoleError> {
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
        for stored in stored_services {
            let manifest = ServiceManifest::from_toml(&stored.manifest_toml)?;
            validate_path_segment("id", &manifest.id)?;
            validate_path_segment("revision", &manifest.revision)?;
            let key = manifest.key()?;
            if stored.service_key != key.to_string() {
                return Err(ConsoleError::InvalidDatabaseData(format!(
                    "service key {:?} does not match manifest key {key}",
                    stored.service_key
                )));
            }
            runtime.register_from_file(manifest.clone())?;
            let capabilities = runtime_capabilities(runtime, &key)?;
            self.state()?.services.insert(
                key,
                ServiceRecord {
                    manifest,
                    capabilities,
                    status: ServiceStatus::Stopped,
                    updated_at_ms: stored.updated_at_ms,
                    calls: stored.calls,
                    errors: stored.errors,
                    last_latency_ms: stored.last_latency_ms,
                },
            );
        }
        Ok(())
    }

    fn load_manifest_services(&self) -> Result<(), ConsoleError> {
        let runtime = self.runtime()?;
        let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
        let mut manifests = fs::read_dir(&self.artifact_dir)
            .map_err(|source| ConsoleError::Storage {
                path: self.artifact_dir.clone(),
                source,
            })?
            .filter_map(Result::ok)
            .map(|entry| entry.path())
            .filter(|path| {
                path.extension()
                    .is_some_and(|extension| extension == "toml")
            })
            .collect::<Vec<_>>();
        manifests.sort();

        for manifest_path in manifests {
            let source =
                fs::read_to_string(&manifest_path).map_err(|source| ConsoleError::Storage {
                    path: manifest_path.clone(),
                    source,
                })?;
            let manifest = ServiceManifest::from_toml(&source)?;
            validate_path_segment("id", &manifest.id)?;
            validate_path_segment("revision", &manifest.revision)?;
            let key = manifest.key()?;
            if self.state()?.services.contains_key(&key) {
                continue;
            }
            runtime.register_from_file(manifest.clone())?;
            let capabilities = runtime_capabilities(runtime, &key)?;
            self.insert_service(manifest, capabilities, ServiceStatus::Stopped)?;
        }

        Ok(())
    }

    fn load_stored_deployments(
        &self,
        stored_deployments: Vec<StoredDeployment>,
    ) -> Result<(), ConsoleError> {
        let mut state = self.state()?;
        for stored in stored_deployments {
            validate_path_segment("id", &stored.service_id)?;
            validate_path_segment("revision", &stored.active_revision)?;
            let key = ServiceKey::new(stored.service_id.clone(), stored.active_revision.clone())?;
            if !state.services.contains_key(&key) {
                return Err(ConsoleError::InvalidDatabaseData(format!(
                    "deployment for {:?} references unmanaged revision {key}",
                    stored.service_id
                )));
            }
            state.deployments.insert(
                stored.service_id,
                DeploymentRecord {
                    active_revision: stored.active_revision,
                    updated_at_ms: stored.updated_at_ms,
                },
            );
        }
        Ok(())
    }

    fn restore_deployments(&self) -> Result<(), ConsoleError> {
        let deployments = {
            let state = self.state()?;
            state
                .deployments
                .iter()
                .map(|(service_id, deployment)| {
                    ServiceKey::new(service_id.clone(), deployment.active_revision.clone())
                })
                .collect::<Result<Vec<_>, _>>()?
        };
        {
            let runtime = self.runtime()?;
            let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
            for key in &deployments {
                runtime.start(key, Vec::new())?;
            }
        }

        let mut state = self.state()?;
        for key in deployments {
            let service = state
                .services
                .get_mut(&key)
                .ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?;
            service.status = ServiceStatus::Running;
        }
        Ok(())
    }

    async fn persist_snapshot(&self) -> Result<(), ConsoleError> {
        // Serialize snapshots so a slower earlier write cannot overwrite a
        // newer state captured by a concurrent HTTP operation.
        let _sync = self.persistence_sync.lock().await;
        let (services, events, deployments) = self.persistence_snapshot()?;
        self.persistence.sync(services, events, deployments).await?;
        Ok(())
    }

    fn persistence_snapshot(&self) -> Result<PersistenceSnapshot, ConsoleError> {
        let state = self.state()?;
        let services = state
            .services
            .iter()
            .map(|(key, service)| {
                Ok(StoredService {
                    service_key: key.to_string(),
                    manifest_toml: toml::to_string_pretty(&service.manifest)?,
                    updated_at_ms: service.updated_at_ms,
                    calls: service.calls,
                    errors: service.errors,
                    last_latency_ms: service.last_latency_ms,
                })
            })
            .collect::<Result<Vec<_>, ConsoleError>>()?;
        let events = state
            .events
            .iter()
            .map(|event| StoredEvent {
                id: event.id,
                timestamp_ms: event.timestamp_ms,
                kind: event.kind.as_database_value().to_owned(),
                service_id: event.service_id.clone(),
                revision: event.revision.clone(),
                message: event.message.clone(),
            })
            .collect();
        let deployments = state
            .deployments
            .iter()
            .map(|(service_id, deployment)| StoredDeployment {
                service_id: service_id.clone(),
                active_revision: deployment.active_revision.clone(),
                updated_at_ms: deployment.updated_at_ms,
            })
            .collect();
        Ok((services, events, deployments))
    }

    fn insert_service(
        &self,
        manifest: ServiceManifest,
        capabilities: Vec<CapabilityView>,
        status: ServiceStatus,
    ) -> Result<ServiceView, ConsoleError> {
        let key = manifest.key()?;
        let mut state = self.state()?;
        let record = ServiceRecord {
            manifest,
            capabilities,
            status,
            updated_at_ms: unix_time_ms(),
            calls: 0,
            errors: 0,
            last_latency_ms: None,
        };
        let view = record.view();
        state.services.insert(key, record);
        Ok(view)
    }

    fn update_status(
        &self,
        key: &ServiceKey,
        status: ServiceStatus,
    ) -> Result<ServiceView, ConsoleError> {
        let mut state = self.state()?;
        let record = state
            .services
            .get_mut(key)
            .ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?;
        record.status = status;
        record.updated_at_ms = unix_time_ms();
        Ok(record.view())
    }

    fn ensure_managed(&self, key: &ServiceKey) -> Result<(), ConsoleError> {
        if self.state()?.services.contains_key(key) {
            Ok(())
        } else {
            Err(ConsoleError::ServiceNotFound(key.clone()))
        }
    }

    fn push_event(
        &self,
        kind: EventKind,
        key: Option<&ServiceKey>,
        message: String,
    ) -> Result<(), ConsoleError> {
        let mut state = self.state()?;
        let event = EventView {
            id: state.next_event_id,
            timestamp_ms: unix_time_ms(),
            kind,
            service_id: key.map(|key| key.id().to_owned()),
            revision: key.map(|key| key.revision().to_owned()),
            message,
        };
        state.next_event_id = state.next_event_id.saturating_add(1);
        state.events.push_front(event);
        state.events.truncate(MAX_EVENTS);
        Ok(())
    }

    fn state(&self) -> Result<MutexGuard<'_, ConsoleState>, ConsoleError> {
        self.state.lock().map_err(|_| ConsoleError::LockPoisoned)
    }

    fn runtime(&self) -> Result<RwLockReadGuard<'_, Option<Runtime>>, ConsoleError> {
        self.runtime.read().map_err(|_| ConsoleError::LockPoisoned)
    }
}

fn runtime_capabilities(
    runtime: &Runtime,
    key: &ServiceKey,
) -> Result<Vec<CapabilityView>, ConsoleError> {
    Ok(runtime
        .capabilities(key)?
        .into_iter()
        .map(Into::into)
        .collect())
}

fn deployment_view(
    state: &ConsoleState,
    service_id: &str,
    deployment: &DeploymentRecord,
) -> Result<DeploymentView, ConsoleError> {
    let key = ServiceKey::new(service_id.to_owned(), deployment.active_revision.clone())?;
    let service = state
        .services
        .get(&key)
        .ok_or_else(|| ConsoleError::ServiceNotFound(key))?;
    Ok(DeploymentView {
        service_id: service_id.to_owned(),
        active_revision: deployment.active_revision.clone(),
        status: service.status,
        updated_at_ms: deployment.updated_at_ms,
    })
}

/// Builds the Axum management API around a shared [`Console`].
///
/// The returned router exposes Runtime control, component registration,
/// invocation, events, and WIT Registry endpoints.
pub fn app(console: Arc<Console>, allowed_origins: Vec<HeaderValue>) -> Router {
    let max_body_bytes = console
        .max_artifact_bytes()
        .max(console.max_wit_package_bytes())
        .saturating_add(1024 * 1024);
    let mut cors = CorsLayer::new()
        .allow_methods([Method::GET, Method::POST])
        .allow_headers([header::CONTENT_TYPE]);
    if !allowed_origins.is_empty() {
        cors = cors.allow_origin(allowed_origins);
    }

    Router::new()
        .route("/healthz", get(health))
        .route("/api/v1/runtime", get(runtime_status))
        .route("/api/v1/runtime/start", post(start_runtime))
        .route("/api/v1/runtime/stop", post(stop_runtime))
        .route("/api/v1/runtime/restart", post(restart_runtime))
        .route("/api/v1/services", get(list_services).post(register))
        .route("/api/v1/deployments", get(list_deployments))
        .route("/api/v1/deployments/{id}", get(get_deployment))
        .route(
            "/api/v1/deployments/{id}/activate",
            post(activate_deployment),
        )
        .route(
            "/api/v1/wit/packages",
            get(list_wit_packages).post(publish_wit_package),
        )
        .route(
            "/api/v1/wit/packages/{namespace}/{name}/{version}",
            get(get_wit_package),
        )
        .route(
            "/api/v1/wit/packages/{namespace}/{name}/{version}/content",
            get(download_wit_package),
        )
        .route(
            "/api/v1/services/{id}/{revision}/start",
            post(start_service),
        )
        .route("/api/v1/services/{id}/{revision}/stop", post(stop_service))
        .route(
            "/api/v1/services/{id}/{revision}/restart",
            post(restart_service),
        )
        .route(
            "/api/v1/services/{id}/{revision}/invoke",
            post(invoke_service),
        )
        .route("/api/v1/events", get(list_events))
        .layer(DefaultBodyLimit::max(max_body_bytes))
        .layer(cors)
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(DefaultMakeSpan::new().level(Level::INFO))
                .on_response(DefaultOnResponse::new().level(Level::INFO)),
        )
        .with_state(console)
}

/// Builds the public data-plane API around the same resident Runtime.
///
/// This router intentionally contains no registration, lifecycle, Registry,
/// or deployment management routes.
pub fn gateway_app(console: Arc<Console>) -> Router {
    let max_input_bytes = console.max_gateway_input_bytes();
    Router::new()
        .route("/healthz", get(health))
        .route("/v1/services/{id}/invoke", post(gateway_invoke))
        .layer(DefaultBodyLimit::max(max_input_bytes))
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(DefaultMakeSpan::new().level(Level::INFO))
                .on_response(DefaultOnResponse::new().level(Level::INFO)),
        )
        .with_state(console)
}

async fn health() -> Json<serde_json::Value> {
    Json(json!({ "status": "ok" }))
}

async fn gateway_invoke(
    State(console): State<Arc<Console>>,
    AxumPath(service_id): AxumPath<String>,
    headers: HeaderMap,
    body: Result<Bytes, BytesRejection>,
) -> Result<Response, GatewayError> {
    validate_path_segment("id", &service_id)?;
    let content_type = headers
        .get(header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        .map(str::trim);
    if !content_type.is_some_and(|value| value.eq_ignore_ascii_case("application/octet-stream")) {
        return Err(GatewayError::UnsupportedMediaType);
    }
    let input = body.map_err(|_| GatewayError::PayloadTooLarge {
        limit: console.max_gateway_input_bytes(),
    })?;
    let invocation = run_blocking(console, move |console| {
        console.gateway_invoke(&service_id, input.to_vec())
    })
    .await?;

    let mut response_headers = HeaderMap::new();
    response_headers.insert(
        header::CONTENT_TYPE,
        HeaderValue::from_static("application/octet-stream"),
    );
    response_headers.insert(
        "x-wasmeld-revision",
        HeaderValue::from_str(&invocation.revision).map_err(|_| {
            ConsoleError::InvalidRequest("revision is not a valid header".to_owned())
        })?,
    );
    response_headers.insert(
        "x-wasmeld-latency-ms",
        HeaderValue::from_str(&invocation.result.latency_ms.to_string()).map_err(|_| {
            ConsoleError::InvalidRequest("latency is not a valid header".to_owned())
        })?,
    );
    Ok((response_headers, invocation.result.output).into_response())
}

async fn runtime_status(
    State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
    Ok(Json(console.runtime_view()?))
}

async fn start_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
    Ok(Json(
        run_blocking(console, |console| console.start_runtime()).await?,
    ))
}

async fn stop_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
    Ok(Json(
        run_blocking(console, |console| console.stop_runtime()).await?,
    ))
}

async fn restart_runtime(
    State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
    Ok(Json(
        run_blocking(console, |console| console.restart_runtime()).await?,
    ))
}

async fn list_services(State(console): State<Arc<Console>>) -> Result<Json<ServiceList>, ApiError> {
    Ok(Json(ServiceList {
        services: console.services()?,
    }))
}

async fn list_deployments(
    State(console): State<Arc<Console>>,
) -> Result<Json<DeploymentList>, ApiError> {
    Ok(Json(DeploymentList {
        deployments: console.deployments()?,
    }))
}

async fn get_deployment(
    State(console): State<Arc<Console>>,
    AxumPath(service_id): AxumPath<String>,
) -> Result<Json<DeploymentView>, ApiError> {
    validate_path_segment("id", &service_id)?;
    Ok(Json(console.deployment(&service_id)?))
}

async fn activate_deployment(
    State(console): State<Arc<Console>>,
    AxumPath(service_id): AxumPath<String>,
    Json(request): Json<ActivateDeploymentRequest>,
) -> Result<Json<DeploymentView>, ApiError> {
    validate_path_segment("id", &service_id)?;
    validate_path_segment("revision", &request.revision)?;
    let init_config = decode_optional_base64(request.init_config_base64.as_deref())?;
    Ok(Json(
        run_blocking(console, move |console| {
            console.activate_deployment(service_id, request.revision, init_config)
        })
        .await?,
    ))
}

async fn list_events(State(console): State<Arc<Console>>) -> Result<Json<EventList>, ApiError> {
    Ok(Json(EventList {
        events: console.events()?,
    }))
}

async fn list_wit_packages(
    State(console): State<Arc<Console>>,
) -> Result<Json<WitPackageList>, ApiError> {
    Ok(Json(WitPackageList {
        packages: console.wit_registry.list()?,
    }))
}

async fn get_wit_package(
    State(console): State<Arc<Console>>,
    AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Json<WitPackageMetadata>, ApiError> {
    Ok(Json(
        console
            .wit_registry
            .metadata(&format!("{namespace}:{name}"), &version)?,
    ))
}

async fn download_wit_package(
    State(console): State<Arc<Console>>,
    AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Response, ApiError> {
    let (metadata, bytes) = console
        .wit_registry
        .download(&format!("{namespace}:{name}"), &version)?;
    Response::builder()
        .status(StatusCode::OK)
        .header(header::CONTENT_TYPE, "application/wasm")
        .header(header::ETAG, format!("\"{}\"", metadata.sha256))
        .body(Body::from(bytes))
        .map_err(|error| {
            ConsoleError::InvalidRequest(format!("failed to build package response: {error}"))
                .into()
        })
}

async fn publish_wit_package(
    State(console): State<Arc<Console>>,
    mut multipart: Multipart,
) -> Result<(StatusCode, Json<WitPackageMetadata>), ApiError> {
    let mut package = None;
    while let Some(field) = multipart
        .next_field()
        .await
        .map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))?
    {
        match field.name() {
            Some("package") => {
                if package.is_some() {
                    return Err(ConsoleError::InvalidRequest(
                        "package field must occur exactly once".to_owned(),
                    )
                    .into());
                }
                package = Some(
                    field
                        .bytes()
                        .await
                        .map_err(|error| {
                            ConsoleError::InvalidRequest(format!(
                                "invalid WIT package field: {error}"
                            ))
                        })?
                        .to_vec(),
                );
            }
            Some(name) => {
                return Err(ConsoleError::InvalidRequest(format!(
                    "unsupported multipart field {name}"
                ))
                .into());
            }
            None => {}
        }
    }
    let package = package
        .ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?;
    if package.len() > console.max_wit_package_bytes() {
        return Err(WitRegistryError::TooLarge {
            limit: console.max_wit_package_bytes(),
        }
        .into());
    }
    let metadata = run_blocking(console, move |console| {
        Ok(console.wit_registry.publish(&package)?)
    })
    .await?;
    Ok((StatusCode::CREATED, Json(metadata)))
}

async fn register(
    State(console): State<Arc<Console>>,
    mut multipart: Multipart,
) -> Result<(StatusCode, Json<ServiceView>), ApiError> {
    let mut package = None;

    while let Some(field) = multipart
        .next_field()
        .await
        .map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))?
    {
        match field.name() {
            Some("package") => {
                if package.is_some() {
                    return Err(ConsoleError::InvalidRequest(
                        "package field must occur exactly once".to_owned(),
                    )
                    .into());
                }
                package = Some(
                    field
                        .bytes()
                        .await
                        .map_err(|error| {
                            ConsoleError::InvalidRequest(format!("invalid package field: {error}"))
                        })?
                        .to_vec(),
                );
            }
            Some(name) => {
                return Err(ConsoleError::InvalidRequest(format!(
                    "unsupported multipart field {name}"
                ))
                .into());
            }
            None => {}
        }
    }

    let package = package
        .ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?;
    let service = run_blocking(console, move |console| console.register_package(package)).await?;
    Ok((StatusCode::CREATED, Json(service)))
}

async fn start_service(
    State(console): State<Arc<Console>>,
    AxumPath((id, revision)): AxumPath<(String, String)>,
    request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, ApiError> {
    let key = api_key(id, revision)?;
    let init_config = decode_optional_base64(
        request
            .as_ref()
            .and_then(|Json(request)| request.init_config_base64.as_deref()),
    )?;
    Ok(Json(
        run_blocking(console, move |console| console.start(&key, init_config)).await?,
    ))
}

async fn stop_service(
    State(console): State<Arc<Console>>,
    AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, ApiError> {
    let key = api_key(id, revision)?;
    Ok(Json(
        run_blocking(console, move |console| console.stop(&key)).await?,
    ))
}

async fn restart_service(
    State(console): State<Arc<Console>>,
    AxumPath((id, revision)): AxumPath<(String, String)>,
    request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, ApiError> {
    let key = api_key(id, revision)?;
    let init_config = decode_optional_base64(
        request
            .as_ref()
            .and_then(|Json(request)| request.init_config_base64.as_deref()),
    )?;
    Ok(Json(
        run_blocking(console, move |console| console.restart(&key, init_config)).await?,
    ))
}

async fn invoke_service(
    State(console): State<Arc<Console>>,
    AxumPath((id, revision)): AxumPath<(String, String)>,
    Json(request): Json<InvokeRequest>,
) -> Result<Json<InvokeResponse>, ApiError> {
    let key = api_key(id, revision)?;
    let input = BASE64.decode(request.input_base64).map_err(|error| {
        ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}"))
    })?;
    let result = run_blocking(console, move |console| console.invoke(&key, input)).await?;
    Ok(Json(InvokeResponse {
        output_base64: BASE64.encode(&result.output),
        output_bytes: result.output.len(),
        latency_ms: result.latency_ms,
    }))
}

async fn run_blocking<T, F>(console: Arc<Console>, operation: F) -> Result<T, ConsoleError>
where
    T: Send + 'static,
    F: FnOnce(Arc<Console>) -> Result<T, ConsoleError> + Send + 'static,
{
    // Wasmtime compilation and synchronous Actor calls must not occupy a Tokio
    // async worker. Persist afterward even on failure because an operation may
    // already have recorded metrics or an event before returning its error.
    let operation_console = Arc::clone(&console);
    let result = tokio::task::spawn_blocking(move || operation(operation_console))
        .await
        .map_err(ConsoleError::from)
        .and_then(|result| result);
    let persisted = console.persist_snapshot().await;

    match result {
        Ok(value) => {
            persisted?;
            Ok(value)
        }
        Err(error) => {
            if let Err(persistence_error) = persisted {
                tracing::error!(
                    error = %persistence_error,
                    "failed to persist console state after operation error"
                );
            }
            Err(error)
        }
    }
}

fn api_key(id: String, revision: String) -> Result<ServiceKey, ConsoleError> {
    validate_path_segment("id", &id)?;
    validate_path_segment("revision", &revision)?;
    ServiceKey::new(id, revision).map_err(Into::into)
}

fn validate_path_segment(name: &str, value: &str) -> Result<(), ConsoleError> {
    let valid = !value.is_empty()
        && value.len() <= 128
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'));
    if valid {
        Ok(())
    } else {
        Err(ConsoleError::InvalidRequest(format!(
            "{name} must contain only ASCII letters, digits, '.', '_' or '-' and be at most 128 bytes"
        )))
    }
}

fn decode_optional_base64(value: Option<&str>) -> Result<Vec<u8>, ConsoleError> {
    match value {
        Some(value) if !value.is_empty() => BASE64.decode(value).map_err(|error| {
            ConsoleError::InvalidRequest(format!("init_config_base64 is invalid: {error}"))
        }),
        _ => Ok(Vec::new()),
    }
}

fn write_atomic(path: &Path, bytes: &[u8]) -> Result<(), ConsoleError> {
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("artifact");
    let temporary = path.with_file_name(format!(".{file_name}.tmp-{}", std::process::id()));
    fs::write(&temporary, bytes).map_err(|source| ConsoleError::Storage {
        path: temporary.clone(),
        source,
    })?;
    if let Err(source) = fs::rename(&temporary, path) {
        let _ = fs::remove_file(&temporary);
        return Err(ConsoleError::Storage {
            path: path.to_path_buf(),
            source,
        });
    }
    Ok(())
}

fn unix_time_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(duration_ms)
        .unwrap_or_default()
}

fn duration_ms(duration: std::time::Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

enum GatewayError {
    Console(ConsoleError),
    UnsupportedMediaType,
    PayloadTooLarge { limit: usize },
}

impl From<ConsoleError> for GatewayError {
    fn from(error: ConsoleError) -> Self {
        Self::Console(error)
    }
}

impl IntoResponse for GatewayError {
    fn into_response(self) -> Response {
        let (status, code, message) = match self {
            Self::UnsupportedMediaType => (
                StatusCode::UNSUPPORTED_MEDIA_TYPE,
                "unsupported_media_type",
                "Content-Type must be application/octet-stream".to_owned(),
            ),
            Self::PayloadTooLarge { limit } => (
                StatusCode::PAYLOAD_TOO_LARGE,
                "payload_too_large",
                format!("request body exceeds the {limit}-byte limit"),
            ),
            Self::Console(error) => {
                let (status, code, message) = match &error {
                    ConsoleError::InvalidRequest(_) => (
                        StatusCode::BAD_REQUEST,
                        "invalid_request",
                        error.to_string(),
                    ),
                    ConsoleError::DeploymentNotFound(_)
                    | ConsoleError::ServiceNotFound(_)
                    | ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => (
                        StatusCode::NOT_FOUND,
                        "service_not_found",
                        "service is not deployed".to_owned(),
                    ),
                    ConsoleError::RuntimeNotRunning
                    | ConsoleError::Runtime(RuntimeError::ActorUnavailable(_))
                    | ConsoleError::Runtime(RuntimeError::ActorStopped(_)) => (
                        StatusCode::SERVICE_UNAVAILABLE,
                        "service_unavailable",
                        "service is temporarily unavailable".to_owned(),
                    ),
                    ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => (
                        StatusCode::TOO_MANY_REQUESTS,
                        "service_overloaded",
                        "service is overloaded".to_owned(),
                    ),
                    ConsoleError::Runtime(RuntimeError::InputTooLarge { .. }) => (
                        StatusCode::PAYLOAD_TOO_LARGE,
                        "payload_too_large",
                        error.to_string(),
                    ),
                    ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => (
                        StatusCode::GATEWAY_TIMEOUT,
                        "deadline_exceeded",
                        "service execution deadline exceeded".to_owned(),
                    ),
                    ConsoleError::Runtime(
                        RuntimeError::OutputTooLarge { .. }
                        | RuntimeError::ActorFault { .. }
                        | RuntimeError::ComponentError { .. },
                    ) => (StatusCode::BAD_GATEWAY, "service_error", error.to_string()),
                    ConsoleError::ArtifactTooLarge { .. }
                    | ConsoleError::Package(_)
                    | ConsoleError::WitRegistry(_)
                    | ConsoleError::Storage { .. }
                    | ConsoleError::ManifestSerialize(_)
                    | ConsoleError::Database(_)
                    | ConsoleError::InvalidDatabaseData(_)
                    | ConsoleError::Runtime(_)
                    | ConsoleError::Join(_)
                    | ConsoleError::LockPoisoned => (
                        StatusCode::INTERNAL_SERVER_ERROR,
                        "internal_error",
                        "internal gateway error".to_owned(),
                    ),
                };
                (status, code, message)
            }
        };

        (
            status,
            Json(json!({
                "error": {
                    "code": code,
                    "message": message,
                }
            })),
        )
            .into_response()
    }
}

struct ApiError(ConsoleError);

impl From<ConsoleError> for ApiError {
    fn from(error: ConsoleError) -> Self {
        Self(error)
    }
}

impl From<WitRegistryError> for ApiError {
    fn from(error: WitRegistryError) -> Self {
        Self(error.into())
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let (status, code) = match &self.0 {
            ConsoleError::InvalidRequest(_) => (StatusCode::BAD_REQUEST, "invalid_request"),
            ConsoleError::ServiceNotFound(_) => (StatusCode::NOT_FOUND, "service_not_found"),
            ConsoleError::DeploymentNotFound(_) => (StatusCode::NOT_FOUND, "deployment_not_found"),
            ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"),
            ConsoleError::ArtifactTooLarge { .. }
            | ConsoleError::Package(PackageError::ComponentTooLarge { .. })
            | ConsoleError::WitRegistry(WitRegistryError::TooLarge { .. })
            | ConsoleError::Runtime(RuntimeError::InputTooLarge { .. })
            | ConsoleError::Runtime(RuntimeError::OutputTooLarge { .. }) => {
                (StatusCode::PAYLOAD_TOO_LARGE, "payload_too_large")
            }
            ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => {
                (StatusCode::NOT_FOUND, "service_not_registered")
            }
            ConsoleError::Runtime(RuntimeError::ServiceAlreadyRegistered(_))
            | ConsoleError::Runtime(RuntimeError::ActorUnavailable(_))
            | ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => {
                (StatusCode::CONFLICT, "runtime_conflict")
            }
            ConsoleError::WitRegistry(WitRegistryError::AlreadyExists(_)) => {
                (StatusCode::CONFLICT, "wit_package_exists")
            }
            ConsoleError::WitRegistry(WitRegistryError::NotFound(_)) => {
                (StatusCode::NOT_FOUND, "wit_package_not_found")
            }
            ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => {
                (StatusCode::REQUEST_TIMEOUT, "deadline_exceeded")
            }
            ConsoleError::Runtime(
                RuntimeError::InvalidServiceKey
                | RuntimeError::InvalidManifest(_)
                | RuntimeError::ManifestParse(_)
                | RuntimeError::ComponentCompilation(_)
                | RuntimeError::UnsupportedImport(_)
                | RuntimeError::ActorInitialization(_)
                | RuntimeError::ComponentError { .. },
            ) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_component"),
            ConsoleError::Package(_) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_package"),
            ConsoleError::WitRegistry(WitRegistryError::InvalidPackage(
                WitPackageError::Source { .. }
                | WitPackageError::Encode(_)
                | WitPackageError::Decode(_)
                | WitPackageError::MissingVersion(_),
            )) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_wit_package"),
            ConsoleError::Storage { .. }
            | ConsoleError::ManifestSerialize(_)
            | ConsoleError::Database(_)
            | ConsoleError::InvalidDatabaseData(_)
            | ConsoleError::Runtime(_)
            | ConsoleError::WitRegistry(WitRegistryError::InvalidStorage(_))
            | ConsoleError::WitRegistry(WitRegistryError::Storage { .. })
            | ConsoleError::WitRegistry(WitRegistryError::LockPoisoned)
            | ConsoleError::Join(_)
            | ConsoleError::LockPoisoned => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"),
        };

        (
            status,
            Json(json!({
                "error": {
                    "code": code,
                    "message": self.0.to_string(),
                }
            })),
        )
            .into_response()
    }
}