//! Host-side ownership and state machine for resident Component resources.
//!
//! # Architecture
//!
//! A resident Component is not a native daemon and never owns a socket, file
//! descriptor, timer task, broker consumer, or async runtime. Those objects
//! belong to a Host driver. The driver translates external activity into
//! [`ResidentEvent`](crate::ResidentEvent) values and applies the validated
//! [`ResidentOperation`] values returned by this module.
//!
//! ```text
//! TCP / UDP / Unix / timer / broker / custom driver
//!                         |
//!                         v
//!                 ResidentSession
//!               resource + state checks
//!                         |
//!                         v
//!               bounded Actor mailbox
//!                         |
//!                         v
//!                    Wasm Component
//!                         |
//!                    raw effects
//!                         |
//!                         v
//!                 ResidentSession
//!               ownership + policy checks
//!                         |
//!                         v
//!                 driver operations
//! ```
//!
//! One session is pinned to one [`ServiceKey`]. It must not be reused when a
//! deployment switches revisions. The deployment manager should start a new
//! Actor and session, attach new drivers, drain the old session, deliver
//! `shutdown`, close its Host resources, and finally stop the old Actor.
//!
//! Driver event loops may run concurrently, but calls into a session must be
//! serialized by their owner (normally one supervisor task). The Actor mailbox
//! is the final serialization boundary for Component memory.

use std::{
    collections::HashMap,
    fs,
    net::SocketAddr,
    path::{Component as PathComponent, Path, PathBuf},
};

use serde::{Deserialize, Serialize};
use thiserror::Error;

use crate::{
    ActorHandle, ComponentExecution, ResidentEffect, ResidentEvent, ResourceId, RuntimeError,
    ServiceKey, StreamCloseReason,
};

/// Network exposure allowed for one transport direction.
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum NetworkScope {
    /// No endpoint of this class may be registered.
    #[default]
    Disabled,
    /// Only numeric loopback addresses are accepted.
    Loopback,
    /// Any numeric IP address is accepted.
    Any,
}

impl NetworkScope {
    fn allows(self, address: SocketAddr) -> bool {
        match self {
            Self::Disabled => false,
            Self::Loopback => address.ip().is_loopback(),
            Self::Any => true,
        }
    }
}

/// Platform policy applied before a Host driver binds an endpoint.
///
/// The default is deny-all. Unix roots must already exist so their canonical
/// paths can be checked before a socket is created. A Unix driver must perform
/// the same check again immediately before bind to prevent path replacement
/// between validation and use.
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
pub struct ResidentPolicy {
    pub tcp_listen: NetworkScope,
    pub udp_bind: NetworkScope,
    pub unix_listen_roots: Vec<PathBuf>,
}

/// A management-defined endpoint. Components cannot create listeners.
///
/// Using `SocketAddr` instead of a hostname avoids DNS changes bypassing the
/// configured network scope. Protocols such as HTTP, WebSocket, gRPC, TLS, and
/// framed streams are implemented by Host drivers above these transports.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "transport", rename_all = "snake_case")]
pub enum ResidentEndpoint {
    Tcp { name: String, bind: SocketAddr },
    Udp { name: String, bind: SocketAddr },
    Unix { name: String, path: PathBuf },
}

impl ResidentEndpoint {
    fn name(&self) -> &str {
        match self {
            Self::Tcp { name, .. } | Self::Udp { name, .. } | Self::Unix { name, .. } => name,
        }
    }
}

/// Stable category exposed in management resource snapshots.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ResidentResourceKind {
    TcpEndpoint,
    UdpEndpoint,
    UnixEndpoint,
    Stream,
    Timer,
    MessageSubscription,
    Source,
}

/// Observable lifecycle state without exposing driver handles.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ResidentResourceState {
    Ready,
    Open,
    ReadPaused,
    PeerHalfClosed,
    Closing,
}

/// Management-safe snapshot of one Host-owned resource.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct ResidentResourceInfo {
    pub id: ResourceId,
    pub name: String,
    pub kind: ResidentResourceKind,
    pub state: ResidentResourceState,
    pub metadata: ResidentResourceMetadata,
}

/// Kind-specific metadata required to correlate a resource with its driver.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ResidentResourceMetadata {
    Endpoint { endpoint: ResidentEndpoint },
    Stream { endpoint_id: ResourceId },
    Timer { armed: bool },
    MessageSubscription,
    Source { source_kind: String },
}

/// An operation that passed payload, ownership, kind, and state validation.
///
/// This is the only output a network or source driver should execute. Raw
/// [`ResidentEffect`] values are intentionally kept inside [`ResidentSession`].
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ResidentOperation {
    WriteStream {
        stream_id: ResourceId,
        bytes: Vec<u8>,
    },
    CloseStream {
        stream_id: ResourceId,
    },
    PauseStream {
        stream_id: ResourceId,
    },
    ResumeStream {
        stream_id: ResourceId,
    },
    SendDatagram {
        endpoint_id: ResourceId,
        peer: String,
        bytes: Vec<u8>,
    },
    ArmTimer {
        timer_id: ResourceId,
        delay_ms: u64,
        interval_ms: Option<u64>,
    },
    CancelTimer {
        timer_id: ResourceId,
    },
    AcknowledgeMessage {
        subscription_id: ResourceId,
        message_id: u64,
    },
    SourceCommand {
        source_id: ResourceId,
        command: String,
        payload: Vec<u8>,
    },
}

/// Host resource, policy, or Actor failure while processing resident work.
#[derive(Debug, Error)]
pub enum ResidentHostError {
    #[error(transparent)]
    Runtime(#[from] RuntimeError),

    #[error("resident session for {0} is shutting down")]
    ShuttingDown(ServiceKey),

    #[error("resident resource limit for {service} is {limit}")]
    ResourceLimit { service: ServiceKey, limit: usize },

    #[error("resident resource ids exhausted for {0}")]
    ResourceIdsExhausted(ServiceKey),

    #[error("resident resource name must not be empty")]
    EmptyResourceName,

    #[error("resident resource name {0} is already registered")]
    DuplicateResourceName(String),

    #[error("resident resource {resource_id} for {service} still owns active streams")]
    ResourceInUse {
        service: ServiceKey,
        resource_id: u64,
    },

    #[error("resident resource {resource_id} does not belong to {service}")]
    UnknownResource {
        service: ServiceKey,
        resource_id: u64,
    },

    #[error("resident resource {resource_id} for {service} is {actual:?}, expected {expected}")]
    WrongResourceKind {
        service: ServiceKey,
        resource_id: u64,
        actual: ResidentResourceKind,
        expected: &'static str,
    },

    #[error("resident resource {resource_id} for {service} is in invalid state {state:?}")]
    InvalidResourceState {
        service: ServiceKey,
        resource_id: u64,
        state: ResidentResourceState,
    },

    #[error("resident endpoint denied by policy: {0}")]
    EndpointDenied(String),
}

#[derive(Clone, Debug)]
enum ResourceEntry {
    Endpoint(ResidentEndpoint),
    Stream {
        name: String,
        endpoint_id: ResourceId,
        state: StreamState,
    },
    Timer {
        name: String,
        armed: bool,
    },
    MessageSubscription {
        name: String,
    },
    Source {
        name: String,
        source_kind: String,
    },
}

impl ResourceEntry {
    fn name(&self) -> &str {
        match self {
            Self::Endpoint(endpoint) => endpoint.name(),
            Self::Stream { name, .. }
            | Self::Timer { name, .. }
            | Self::MessageSubscription { name }
            | Self::Source { name, .. } => name,
        }
    }

    fn kind(&self) -> ResidentResourceKind {
        match self {
            Self::Endpoint(ResidentEndpoint::Tcp { .. }) => ResidentResourceKind::TcpEndpoint,
            Self::Endpoint(ResidentEndpoint::Udp { .. }) => ResidentResourceKind::UdpEndpoint,
            Self::Endpoint(ResidentEndpoint::Unix { .. }) => ResidentResourceKind::UnixEndpoint,
            Self::Stream { .. } => ResidentResourceKind::Stream,
            Self::Timer { .. } => ResidentResourceKind::Timer,
            Self::MessageSubscription { .. } => ResidentResourceKind::MessageSubscription,
            Self::Source { .. } => ResidentResourceKind::Source,
        }
    }

    fn state(&self) -> ResidentResourceState {
        match self {
            Self::Endpoint(_)
            | Self::Timer { .. }
            | Self::MessageSubscription { .. }
            | Self::Source { .. } => ResidentResourceState::Ready,
            Self::Stream { state, .. } => (*state).into(),
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum StreamState {
    Open,
    ReadPaused,
    PeerHalfClosed,
    PeerHalfClosedReadPaused,
    Closing,
}

impl From<StreamState> for ResidentResourceState {
    fn from(state: StreamState) -> Self {
        match state {
            StreamState::Open => Self::Open,
            StreamState::ReadPaused | StreamState::PeerHalfClosedReadPaused => Self::ReadPaused,
            StreamState::PeerHalfClosed => Self::PeerHalfClosed,
            StreamState::Closing => Self::Closing,
        }
    }
}

/// Revision-scoped supervisor for Host-owned resident resources.
///
/// This type intentionally performs no operating-system I/O. It makes driver
/// implementations replaceable and keeps Tokio or another async executor out
/// of the Wasmtime Actor thread.
pub struct ResidentSession {
    actor: ActorHandle,
    policy: ResidentPolicy,
    resources: HashMap<ResourceId, ResourceEntry>,
    next_resource_id: u64,
    shutting_down: bool,
}

impl ResidentSession {
    /// Creates a resource session for an Actor exporting resident WIT 0.1.0.
    pub fn new(actor: ActorHandle, policy: ResidentPolicy) -> Result<Self, ResidentHostError> {
        if actor.execution() != ComponentExecution::Resident {
            return Err(RuntimeError::NotResident(actor.key().clone()).into());
        }
        Ok(Self {
            actor,
            policy,
            resources: HashMap::new(),
            next_resource_id: 1,
            shutting_down: false,
        })
    }

    /// Returns the immutable service revision that owns every resource ID.
    pub fn owner(&self) -> &ServiceKey {
        self.actor.key()
    }

    /// Registers a configured listener or datagram endpoint after policy checks.
    pub fn register_endpoint(
        &mut self,
        endpoint: ResidentEndpoint,
    ) -> Result<ResourceId, ResidentHostError> {
        self.ensure_active()?;
        self.policy.validate_endpoint(&endpoint)?;
        self.insert_resource(ResourceEntry::Endpoint(endpoint), true)
    }

    /// Registers a Host timer source. The driver decides its initial schedule.
    pub fn register_timer(
        &mut self,
        name: impl Into<String>,
    ) -> Result<ResourceId, ResidentHostError> {
        self.ensure_active()?;
        self.insert_resource(
            ResourceEntry::Timer {
                name: name.into(),
                armed: false,
            },
            true,
        )
    }

    /// Registers a queue, broker, or internal mailbox subscription.
    pub fn register_message_subscription(
        &mut self,
        name: impl Into<String>,
    ) -> Result<ResourceId, ResidentHostError> {
        self.ensure_active()?;
        self.insert_resource(
            ResourceEntry::MessageSubscription { name: name.into() },
            true,
        )
    }

    /// Registers an extension driver such as a file watcher or serial device.
    ///
    /// `source_kind` identifies the separately versioned capability contract
    /// understood by both the driver and Component.
    pub fn register_source(
        &mut self,
        name: impl Into<String>,
        source_kind: impl Into<String>,
    ) -> Result<ResourceId, ResidentHostError> {
        self.ensure_active()?;
        let source_kind = source_kind.into();
        if source_kind.trim().is_empty() {
            return Err(ResidentHostError::EmptyResourceName);
        }
        self.insert_resource(
            ResourceEntry::Source {
                name: name.into(),
                source_kind,
            },
            true,
        )
    }

    /// Allocates a stream for a connection accepted by a TCP or Unix driver.
    pub fn accept_stream(
        &mut self,
        endpoint_id: ResourceId,
        peer: Option<String>,
    ) -> Result<(ResourceId, Vec<ResidentOperation>), ResidentHostError> {
        self.ensure_active()?;
        let endpoint = self.resource(endpoint_id)?;
        match endpoint {
            ResourceEntry::Endpoint(ResidentEndpoint::Tcp { .. })
            | ResourceEntry::Endpoint(ResidentEndpoint::Unix { .. }) => {}
            other => {
                return Err(self.wrong_kind(endpoint_id, other.kind(), "TCP or Unix endpoint"));
            }
        }

        let stream_id = self.insert_resource(
            ResourceEntry::Stream {
                name: format!("stream-{}", self.next_resource_id),
                endpoint_id,
                state: StreamState::Open,
            },
            false,
        )?;
        let result = self.dispatch(ResidentEvent::StreamOpened {
            endpoint_id,
            stream_id,
            peer,
        });
        if result.is_err() {
            self.resources.remove(&stream_id);
        }
        result.map(|operations| (stream_id, operations))
    }

    /// Delivers one bounded stream read. Paused and closing streams reject data.
    pub fn stream_data(
        &mut self,
        stream_id: ResourceId,
        bytes: Vec<u8>,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_stream_state(stream_id, &[StreamState::Open])?;
        self.dispatch(ResidentEvent::StreamData { stream_id, bytes })
    }

    /// Signals that a driver can accept writes again.
    pub fn stream_writable(
        &mut self,
        stream_id: ResourceId,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_stream_not_closing(stream_id)?;
        self.dispatch(ResidentEvent::StreamWritable { stream_id })
    }

    /// Records a peer read-half close while preserving the local write half.
    pub fn stream_half_closed(
        &mut self,
        stream_id: ResourceId,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        let state = self.stream_state_mut(stream_id)?;
        *state = match *state {
            StreamState::Open => StreamState::PeerHalfClosed,
            StreamState::ReadPaused => StreamState::PeerHalfClosedReadPaused,
            other => return Err(self.invalid_state(stream_id, other.into())),
        };
        self.dispatch(ResidentEvent::StreamHalfClosed { stream_id })
    }

    /// Delivers terminal stream state and releases the resource identity.
    pub fn stream_closed(
        &mut self,
        stream_id: ResourceId,
        reason: StreamCloseReason,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_kind(stream_id, ResidentResourceKind::Stream, "stream")?;
        let result = self.dispatch(ResidentEvent::StreamClosed { stream_id, reason });
        self.resources.remove(&stream_id);
        result
    }

    /// Delivers one UDP datagram to its configured endpoint.
    pub fn datagram(
        &mut self,
        endpoint_id: ResourceId,
        peer: String,
        bytes: Vec<u8>,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_kind(
            endpoint_id,
            ResidentResourceKind::UdpEndpoint,
            "UDP endpoint",
        )?;
        self.dispatch(ResidentEvent::Datagram {
            endpoint_id,
            peer,
            bytes,
        })
    }

    /// Delivers a timer occurrence.
    pub fn timer_fired(
        &mut self,
        timer_id: ResourceId,
        scheduled_at_ns: u64,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_kind(timer_id, ResidentResourceKind::Timer, "timer")?;
        if let ResourceEntry::Timer { armed, .. } = self.resource_mut(timer_id)? {
            *armed = false;
        }
        self.dispatch(ResidentEvent::Timer {
            timer_id,
            scheduled_at_ns,
        })
    }

    /// Delivers a broker or internal queue message.
    pub fn message(
        &mut self,
        subscription_id: ResourceId,
        message_id: u64,
        bytes: Vec<u8>,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_kind(
            subscription_id,
            ResidentResourceKind::MessageSubscription,
            "message subscription",
        )?;
        self.dispatch(ResidentEvent::Message {
            subscription_id,
            message_id,
            bytes,
        })
    }

    /// Delivers an event from a separately versioned extension driver.
    pub fn source_event(
        &mut self,
        source_id: ResourceId,
        event: String,
        payload: Vec<u8>,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_kind(source_id, ResidentResourceKind::Source, "source")?;
        self.dispatch(ResidentEvent::Source {
            source_id,
            event,
            payload,
        })
    }

    /// Gives the Component one final bounded callback and rejects later events.
    ///
    /// After applying the returned operations, the owner closes every driver
    /// resource and stops the Actor. Shutdown is not a drain barrier by itself.
    pub fn shutdown(&mut self) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        self.ensure_active()?;
        let result = self.dispatch(ResidentEvent::Shutdown);
        self.shutting_down = true;
        result
    }

    /// Removes a Host resource after its driver has released the actual handle.
    pub fn release_resource(
        &mut self,
        resource_id: ResourceId,
    ) -> Result<ResidentResourceInfo, ResidentHostError> {
        if self.resources.values().any(|entry| {
            matches!(
                entry,
                ResourceEntry::Stream { endpoint_id, .. } if *endpoint_id == resource_id
            )
        }) {
            return Err(ResidentHostError::ResourceInUse {
                service: self.owner().clone(),
                resource_id: resource_id.get(),
            });
        }
        let entry = self
            .resources
            .remove(&resource_id)
            .ok_or_else(|| self.unknown_resource(resource_id))?;
        Ok(resource_info(resource_id, &entry))
    }

    /// Returns one management-safe resource snapshot.
    pub fn resource_info(
        &self,
        resource_id: ResourceId,
    ) -> Result<ResidentResourceInfo, ResidentHostError> {
        self.resource(resource_id)
            .map(|entry| resource_info(resource_id, entry))
    }

    /// Returns all resource snapshots ordered by their non-reused IDs.
    pub fn resources(&self) -> Vec<ResidentResourceInfo> {
        let mut resources = self
            .resources
            .iter()
            .map(|(id, entry)| resource_info(*id, entry))
            .collect::<Vec<_>>();
        resources.sort_unstable_by_key(|resource| resource.id);
        resources
    }

    fn dispatch(
        &mut self,
        event: ResidentEvent,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        let effects = self.actor.dispatch_event(event)?;
        self.apply_effects(effects)
    }

    fn apply_effects(
        &mut self,
        effects: Vec<ResidentEffect>,
    ) -> Result<Vec<ResidentOperation>, ResidentHostError> {
        // Validate the whole batch against a copy. A bad later effect must not
        // leave pause, close, or timer state from an earlier effect committed.
        let mut next_resources = self.resources.clone();
        let mut operations = Vec::with_capacity(effects.len());
        for effect in effects {
            operations.push(self.validate_effect(&mut next_resources, effect)?);
        }
        self.resources = next_resources;
        Ok(operations)
    }

    fn validate_effect(
        &self,
        resources: &mut HashMap<ResourceId, ResourceEntry>,
        effect: ResidentEffect,
    ) -> Result<ResidentOperation, ResidentHostError> {
        match effect {
            ResidentEffect::WriteStream { stream_id, bytes } => {
                ensure_stream_not_closing(resources, self.owner(), stream_id)?;
                Ok(ResidentOperation::WriteStream { stream_id, bytes })
            }
            ResidentEffect::CloseStream { stream_id } => {
                *stream_state_mut(resources, self.owner(), stream_id)? = StreamState::Closing;
                Ok(ResidentOperation::CloseStream { stream_id })
            }
            ResidentEffect::PauseStream { stream_id } => {
                let state = stream_state_mut(resources, self.owner(), stream_id)?;
                *state = match *state {
                    StreamState::Open => StreamState::ReadPaused,
                    StreamState::PeerHalfClosed => StreamState::PeerHalfClosedReadPaused,
                    other => return Err(invalid_state(self.owner(), stream_id, other.into())),
                };
                Ok(ResidentOperation::PauseStream { stream_id })
            }
            ResidentEffect::ResumeStream { stream_id } => {
                let state = stream_state_mut(resources, self.owner(), stream_id)?;
                *state = match *state {
                    StreamState::ReadPaused => StreamState::Open,
                    StreamState::PeerHalfClosedReadPaused => StreamState::PeerHalfClosed,
                    other => return Err(invalid_state(self.owner(), stream_id, other.into())),
                };
                Ok(ResidentOperation::ResumeStream { stream_id })
            }
            ResidentEffect::SendDatagram {
                endpoint_id,
                peer,
                bytes,
            } => {
                ensure_kind(
                    resources,
                    self.owner(),
                    endpoint_id,
                    ResidentResourceKind::UdpEndpoint,
                    "UDP endpoint",
                )?;
                Ok(ResidentOperation::SendDatagram {
                    endpoint_id,
                    peer,
                    bytes,
                })
            }
            ResidentEffect::ArmTimer {
                timer_id,
                delay_ms,
                interval_ms,
            } => {
                let entry = resource_mut(resources, self.owner(), timer_id)?;
                let ResourceEntry::Timer { armed, .. } = entry else {
                    return Err(wrong_kind(self.owner(), timer_id, entry.kind(), "timer"));
                };
                *armed = true;
                Ok(ResidentOperation::ArmTimer {
                    timer_id,
                    delay_ms,
                    interval_ms,
                })
            }
            ResidentEffect::CancelTimer { timer_id } => {
                let entry = resource_mut(resources, self.owner(), timer_id)?;
                let ResourceEntry::Timer { armed, .. } = entry else {
                    return Err(wrong_kind(self.owner(), timer_id, entry.kind(), "timer"));
                };
                *armed = false;
                Ok(ResidentOperation::CancelTimer { timer_id })
            }
            ResidentEffect::AcknowledgeMessage {
                subscription_id,
                message_id,
            } => {
                ensure_kind(
                    resources,
                    self.owner(),
                    subscription_id,
                    ResidentResourceKind::MessageSubscription,
                    "message subscription",
                )?;
                Ok(ResidentOperation::AcknowledgeMessage {
                    subscription_id,
                    message_id,
                })
            }
            ResidentEffect::SourceCommand {
                source_id,
                command,
                payload,
            } => {
                ensure_kind(
                    resources,
                    self.owner(),
                    source_id,
                    ResidentResourceKind::Source,
                    "source",
                )?;
                Ok(ResidentOperation::SourceCommand {
                    source_id,
                    command,
                    payload,
                })
            }
        }
    }

    fn insert_resource(
        &mut self,
        entry: ResourceEntry,
        unique_name: bool,
    ) -> Result<ResourceId, ResidentHostError> {
        if entry.name().trim().is_empty() {
            return Err(ResidentHostError::EmptyResourceName);
        }
        if unique_name
            && self
                .resources
                .values()
                .any(|existing| existing.name() == entry.name())
        {
            return Err(ResidentHostError::DuplicateResourceName(
                entry.name().to_owned(),
            ));
        }
        let limit = self.actor.resident_resource_limit();
        if self.resources.len() >= limit {
            return Err(ResidentHostError::ResourceLimit {
                service: self.owner().clone(),
                limit,
            });
        }

        let id = ResourceId::new(self.next_resource_id)
            .ok_or_else(|| ResidentHostError::ResourceIdsExhausted(self.owner().clone()))?;
        self.next_resource_id = self
            .next_resource_id
            .checked_add(1)
            .ok_or_else(|| ResidentHostError::ResourceIdsExhausted(self.owner().clone()))?;
        self.resources.insert(id, entry);
        Ok(id)
    }

    fn ensure_active(&self) -> Result<(), ResidentHostError> {
        if self.shutting_down {
            Err(ResidentHostError::ShuttingDown(self.owner().clone()))
        } else {
            Ok(())
        }
    }

    fn resource(&self, id: ResourceId) -> Result<&ResourceEntry, ResidentHostError> {
        self.resources
            .get(&id)
            .ok_or_else(|| self.unknown_resource(id))
    }

    fn resource_mut(&mut self, id: ResourceId) -> Result<&mut ResourceEntry, ResidentHostError> {
        let owner = self.owner().clone();
        self.resources
            .get_mut(&id)
            .ok_or(ResidentHostError::UnknownResource {
                service: owner,
                resource_id: id.get(),
            })
    }

    fn ensure_kind(
        &self,
        id: ResourceId,
        expected_kind: ResidentResourceKind,
        expected: &'static str,
    ) -> Result<(), ResidentHostError> {
        let entry = self.resource(id)?;
        if entry.kind() == expected_kind {
            Ok(())
        } else {
            Err(self.wrong_kind(id, entry.kind(), expected))
        }
    }

    fn stream_state_mut(&mut self, id: ResourceId) -> Result<&mut StreamState, ResidentHostError> {
        let owner = self.owner().clone();
        stream_state_mut(&mut self.resources, &owner, id)
    }

    fn ensure_stream_state(
        &self,
        id: ResourceId,
        allowed: &[StreamState],
    ) -> Result<(), ResidentHostError> {
        let entry = self.resource(id)?;
        let ResourceEntry::Stream { state, .. } = entry else {
            return Err(self.wrong_kind(id, entry.kind(), "stream"));
        };
        if allowed.contains(state) {
            Ok(())
        } else {
            Err(self.invalid_state(id, (*state).into()))
        }
    }

    fn ensure_stream_not_closing(&self, id: ResourceId) -> Result<(), ResidentHostError> {
        ensure_stream_not_closing(&self.resources, self.owner(), id)
    }

    fn unknown_resource(&self, id: ResourceId) -> ResidentHostError {
        ResidentHostError::UnknownResource {
            service: self.owner().clone(),
            resource_id: id.get(),
        }
    }

    fn wrong_kind(
        &self,
        id: ResourceId,
        actual: ResidentResourceKind,
        expected: &'static str,
    ) -> ResidentHostError {
        wrong_kind(self.owner(), id, actual, expected)
    }

    fn invalid_state(&self, id: ResourceId, state: ResidentResourceState) -> ResidentHostError {
        invalid_state(self.owner(), id, state)
    }
}

impl ResidentPolicy {
    fn validate_endpoint(&self, endpoint: &ResidentEndpoint) -> Result<(), ResidentHostError> {
        match endpoint {
            ResidentEndpoint::Tcp { bind, .. } if !self.tcp_listen.allows(*bind) => {
                Err(ResidentHostError::EndpointDenied(format!(
                    "TCP listen address {bind} is outside {:?}",
                    self.tcp_listen
                )))
            }
            ResidentEndpoint::Udp { bind, .. } if !self.udp_bind.allows(*bind) => {
                Err(ResidentHostError::EndpointDenied(format!(
                    "UDP bind address {bind} is outside {:?}",
                    self.udp_bind
                )))
            }
            ResidentEndpoint::Unix { path, .. } => self.validate_unix_path(path),
            ResidentEndpoint::Tcp { .. } | ResidentEndpoint::Udp { .. } => Ok(()),
        }
    }

    fn validate_unix_path(&self, path: &Path) -> Result<(), ResidentHostError> {
        if !path.is_absolute()
            || path
                .components()
                .any(|component| matches!(component, PathComponent::ParentDir))
        {
            return Err(ResidentHostError::EndpointDenied(format!(
                "Unix socket path {} must be absolute and contain no parent traversal",
                path.display()
            )));
        }
        let parent = path.parent().ok_or_else(|| {
            ResidentHostError::EndpointDenied(format!(
                "Unix socket path {} has no parent",
                path.display()
            ))
        })?;
        let resolved_parent = fs::canonicalize(parent).map_err(|error| {
            ResidentHostError::EndpointDenied(format!(
                "Unix socket parent {} cannot be resolved: {error}",
                parent.display()
            ))
        })?;

        for root in &self.unix_listen_roots {
            let resolved_root = fs::canonicalize(root).map_err(|error| {
                ResidentHostError::EndpointDenied(format!(
                    "configured Unix root {} cannot be resolved: {error}",
                    root.display()
                ))
            })?;
            if resolved_parent.starts_with(resolved_root) {
                return Ok(());
            }
        }
        Err(ResidentHostError::EndpointDenied(format!(
            "Unix socket path {} is outside configured roots",
            path.display()
        )))
    }
}

fn resource_info(id: ResourceId, entry: &ResourceEntry) -> ResidentResourceInfo {
    ResidentResourceInfo {
        id,
        name: entry.name().to_owned(),
        kind: entry.kind(),
        state: entry.state(),
        metadata: match entry {
            ResourceEntry::Endpoint(endpoint) => ResidentResourceMetadata::Endpoint {
                endpoint: endpoint.clone(),
            },
            ResourceEntry::Stream { endpoint_id, .. } => ResidentResourceMetadata::Stream {
                endpoint_id: *endpoint_id,
            },
            ResourceEntry::Timer { armed, .. } => ResidentResourceMetadata::Timer { armed: *armed },
            ResourceEntry::MessageSubscription { .. } => {
                ResidentResourceMetadata::MessageSubscription
            }
            ResourceEntry::Source { source_kind, .. } => ResidentResourceMetadata::Source {
                source_kind: source_kind.clone(),
            },
        },
    }
}

fn resource_mut<'a>(
    resources: &'a mut HashMap<ResourceId, ResourceEntry>,
    owner: &ServiceKey,
    id: ResourceId,
) -> Result<&'a mut ResourceEntry, ResidentHostError> {
    resources
        .get_mut(&id)
        .ok_or_else(|| ResidentHostError::UnknownResource {
            service: owner.clone(),
            resource_id: id.get(),
        })
}

fn stream_state_mut<'a>(
    resources: &'a mut HashMap<ResourceId, ResourceEntry>,
    owner: &ServiceKey,
    id: ResourceId,
) -> Result<&'a mut StreamState, ResidentHostError> {
    let entry = resource_mut(resources, owner, id)?;
    let ResourceEntry::Stream { state, .. } = entry else {
        return Err(wrong_kind(owner, id, entry.kind(), "stream"));
    };
    Ok(state)
}

fn ensure_stream_not_closing(
    resources: &HashMap<ResourceId, ResourceEntry>,
    owner: &ServiceKey,
    id: ResourceId,
) -> Result<(), ResidentHostError> {
    let entry = resources
        .get(&id)
        .ok_or_else(|| ResidentHostError::UnknownResource {
            service: owner.clone(),
            resource_id: id.get(),
        })?;
    let ResourceEntry::Stream { state, .. } = entry else {
        return Err(wrong_kind(owner, id, entry.kind(), "stream"));
    };
    if *state == StreamState::Closing {
        Err(invalid_state(owner, id, (*state).into()))
    } else {
        Ok(())
    }
}

fn ensure_kind(
    resources: &HashMap<ResourceId, ResourceEntry>,
    owner: &ServiceKey,
    id: ResourceId,
    expected_kind: ResidentResourceKind,
    expected: &'static str,
) -> Result<(), ResidentHostError> {
    let entry = resources
        .get(&id)
        .ok_or_else(|| ResidentHostError::UnknownResource {
            service: owner.clone(),
            resource_id: id.get(),
        })?;
    if entry.kind() == expected_kind {
        Ok(())
    } else {
        Err(wrong_kind(owner, id, entry.kind(), expected))
    }
}

fn wrong_kind(
    owner: &ServiceKey,
    id: ResourceId,
    actual: ResidentResourceKind,
    expected: &'static str,
) -> ResidentHostError {
    ResidentHostError::WrongResourceKind {
        service: owner.clone(),
        resource_id: id.get(),
        actual,
        expected,
    }
}

fn invalid_state(
    owner: &ServiceKey,
    id: ResourceId,
    state: ResidentResourceState,
) -> ResidentHostError {
    ResidentHostError::InvalidResourceState {
        service: owner.clone(),
        resource_id: id.get(),
        state,
    }
}