epics-bridge-rs 0.18.2

//! Top-level gateway server.
//!
//! Ties together [`PvCache`], [`UpstreamManager`], [`DownstreamServer`],
//! [`PvList`], [`AccessConfig`], [`Stats`] into a single async daemon.
//!
//! ## Main event loop
//!
//! ```text
//! loop {
//!     tokio::select! {
//!         _ = downstream.run()    => break,    // CaServer drives downstream
//!         _ = cleanup_tick.tick() => cache.cleanup() + upstream.sweep_orphaned()
//!         _ = stats_tick.tick()   => stats.refresh() + publish to gateway:* PVs
//!         _ = heartbeat_tick.tick() => heartbeat counter ++
//!         _ = signal_handler      => reload pvlist / dump report
//!     }
//! }
//! ```

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use arc_swap::ArcSwap;
use epics_base_rs::server::database::PvDatabase;
use tokio::sync::RwLock;

use crate::error::BridgeResult;

use super::access::AccessConfig;
use super::beacon::BeaconAnomaly;
use super::cache::{CacheTimeouts, PvCache};
// Used only inside the cfg(unix) signal handler below.
#[cfg(unix)]
use super::command::CommandHandler;
use super::downstream::DownstreamServer;
use super::putlog::PutLog;
use super::pvlist::PvList;
use super::stats::Stats;
use super::upstream::{UpstreamManager, UpstreamManagerConfig};

/// Configuration for [`GatewayServer`].
///
/// `Debug` is implemented manually (see below) rather than derived:
/// the `ca-gateway-tls` `upstream_tls` field holds an
/// `epics_ca_rs::tls::TlsConfig`, which does not implement `Debug`.
/// The manual impl redacts the two TLS fields to a presence marker.
#[derive(Clone)]
pub struct GatewayConfig {
    /// Path to `.pvlist` file.
    pub pvlist_path: Option<PathBuf>,
    /// Inline pvlist content (alternative to file).
    pub pvlist_content: Option<String>,
    /// Path to `.access` (ACF) file.
    pub access_path: Option<PathBuf>,
    /// Optional path to put-event log file.
    pub putlog_path: Option<PathBuf>,
    /// Optional path to a command file processed on SIGUSR1 (Unix only).
    /// Each non-comment line is a [`super::command::GatewayCommand`].
    pub command_path: Option<PathBuf>,
    /// Optional path to a file containing literal upstream PV names to
    /// pre-subscribe (one per line). When set, the gateway pre-fetches
    /// each name on startup. Used because lazy resolution is not yet
    /// supported (see `downstream.rs` doc comment).
    pub preload_path: Option<PathBuf>,
    /// CA server port (downstream side). 0 = use EPICS default.
    pub server_port: u16,
    /// Cache timeouts.
    pub timeouts: CacheTimeouts,
    /// Statistics PV prefix (e.g. `"gateway:"`). Empty disables stats PVs.
    pub stats_prefix: String,
    /// Cleanup sweep interval.
    pub cleanup_interval: Duration,
    /// Statistics refresh interval.
    pub stats_interval: Duration,
    /// Heartbeat increment interval. `None` disables the heartbeat PV.
    pub heartbeat_interval: Option<Duration>,
    /// Read-only mode: rejects all puts.
    pub read_only: bool,
    /// Optional TLS server config for downstream connections.
    /// Available with the `ca-gateway-tls` feature.
    #[cfg(feature = "ca-gateway-tls")]
    pub tls: Option<std::sync::Arc<epics_ca_rs::tls::ServerConfig>>,
    /// Optional TLS client config for the gateway's *upstream*
    /// connections to the real IOC (B10). Independent of the
    /// downstream [`Self::tls`] termination: a site can run plaintext
    /// downstream + TLS upstream, TLS both ends, or any mix. When
    /// `Some`, the upstream `CaClient` wraps every TCP virtual circuit
    /// to the IOC in TLS. `None` keeps upstream traffic plaintext.
    /// Available with the `ca-gateway-tls` feature.
    #[cfg(feature = "ca-gateway-tls")]
    pub upstream_tls: Option<epics_ca_rs::tls::TlsConfig>,
    /// Override SNI / cert-hostname-verification name for the upstream
    /// TLS connections. Forwarded to `CaClientConfig::tls_server_name`.
    /// When `None`, the upstream client falls back to the IOC's IP
    /// literal (which only validates IP-bound certs). Set this to the
    /// DNS name embedded in the upstream IOC's server certificate.
    /// Available with the `ca-gateway-tls` feature.
    #[cfg(feature = "ca-gateway-tls")]
    pub upstream_tls_server_name: Option<String>,
}

impl Default for GatewayConfig {
    fn default() -> Self {
        Self {
            pvlist_path: None,
            pvlist_content: None,
            access_path: None,
            putlog_path: None,
            command_path: None,
            preload_path: None,
            server_port: 0,
            timeouts: CacheTimeouts::default(),
            stats_prefix: "gateway:".to_string(),
            cleanup_interval: Duration::from_secs(10),
            stats_interval: Duration::from_secs(10),
            heartbeat_interval: Some(Duration::from_secs(1)),
            read_only: false,
            #[cfg(feature = "ca-gateway-tls")]
            tls: None,
            #[cfg(feature = "ca-gateway-tls")]
            upstream_tls: None,
            #[cfg(feature = "ca-gateway-tls")]
            upstream_tls_server_name: None,
        }
    }
}

// Manual `Debug` — `epics_ca_rs::tls::TlsConfig` (the `upstream_tls`
// field type) does not implement `Debug`, so the derive cannot be
// used. The TLS server config (`tls`) and upstream client config
// (`upstream_tls`) are redacted to a presence marker; certificate
// material has no business in a `Debug` dump anyway.
impl std::fmt::Debug for GatewayConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut d = f.debug_struct("GatewayConfig");
        d.field("pvlist_path", &self.pvlist_path)
            .field("pvlist_content", &self.pvlist_content)
            .field("access_path", &self.access_path)
            .field("putlog_path", &self.putlog_path)
            .field("command_path", &self.command_path)
            .field("preload_path", &self.preload_path)
            .field("server_port", &self.server_port)
            .field("timeouts", &self.timeouts)
            .field("stats_prefix", &self.stats_prefix)
            .field("cleanup_interval", &self.cleanup_interval)
            .field("stats_interval", &self.stats_interval)
            .field("heartbeat_interval", &self.heartbeat_interval)
            .field("read_only", &self.read_only);
        #[cfg(feature = "ca-gateway-tls")]
        {
            d.field("tls", &self.tls.as_ref().map(|_| "<ServerConfig>"));
            d.field(
                "upstream_tls",
                &self.upstream_tls.as_ref().map(|_| "<TlsConfig>"),
            );
            d.field("upstream_tls_server_name", &self.upstream_tls_server_name);
        }
        d.finish()
    }
}

/// The CA gateway server.
///
/// Construct via [`GatewayServer::build`], then call [`GatewayServer::run`]
/// to start the daemon.
pub struct GatewayServer {
    config: GatewayConfig,
    /// `ArcSwap` so live reload (SIGUSR1 `PVL`) can swap the pvlist
    /// atomically without taking a write lock against the put-hot-path.
    /// Every WriteHook reads via `load_full()` (wait-free).
    pvlist: Arc<ArcSwap<PvList>>,
    /// Same hot-reload pattern as `pvlist` — SIGUSR1 `AS` command
    /// stores a fresh `Arc<AccessConfig>` here; in-flight puts that
    /// already loaded the previous one continue with the old rules
    /// (acceptable; matches the pvlist semantics and what C ca-gateway
    /// does on reload).
    access: Arc<ArcSwap<AccessConfig>>,
    cache: Arc<RwLock<PvCache>>,
    shadow_db: Arc<PvDatabase>,
    upstream: Arc<UpstreamManager>,
    downstream: Arc<DownstreamServer>,
    stats: Arc<Stats>,
    putlog: Option<Arc<PutLog>>,
    beacon_anomaly: Arc<BeaconAnomaly>,
}

impl GatewayServer {
    /// Build the gateway from configuration.
    ///
    /// Loads pvlist + access files, initializes cache + upstream client,
    /// constructs downstream CA server. Does not start any I/O — call
    /// [`GatewayServer::run`] for that.
    pub async fn build(config: GatewayConfig) -> BridgeResult<Self> {
        // Load .pvlist
        let pvlist = if let Some(path) = &config.pvlist_path {
            super::pvlist::parse_pvlist_file(path)?
        } else if let Some(content) = &config.pvlist_content {
            super::pvlist::parse_pvlist(content)?
        } else {
            // Empty pvlist allows nothing
            PvList::new()
        };
        let pvlist = Arc::new(ArcSwap::from_pointee(pvlist));

        // Load .access (optional). `ArcSwap` for the same lock-free
        // hot-reload pattern as `pvlist`.
        let access = if let Some(path) = &config.access_path {
            AccessConfig::from_file(path)?
        } else {
            AccessConfig::allow_all()
        };
        let access = Arc::new(ArcSwap::from_pointee(access));

        // Cache + shadow database
        let cache = Arc::new(RwLock::new(PvCache::new()));
        let shadow_db = Arc::new(PvDatabase::new());

        // Stats — needed before UpstreamManager so per-PV WriteHook
        // closures can capture the same Arc.
        let stats = Arc::new(Stats::new(config.stats_prefix.clone()));

        // Put-event logger (optional) — also captured by every WriteHook.
        let putlog = config
            .putlog_path
            .as_ref()
            .map(|p| Arc::new(PutLog::new(p.clone())));

        // Beacon anomaly throttle — constructed BEFORE UpstreamManager
        // so every per-PV WriteHookEnv captures it (B-G9: the
        // forwarding task fires `request()` on upstream reconnect to
        // tell other gateway-aware downstream clients to re-search).
        // The pulse handle (CaServer beacon-reset Notify) is wired in
        // below once the downstream server has been constructed, so
        // honored `request()` calls actually emit a beacon — without
        // this the throttle just tracked timestamps and
        // `generateBeaconAnomaly` was silent on the wire.
        let beacon_anomaly = Arc::new(BeaconAnomaly::new());

        // Upstream manager — receives the full WriteHook environment so
        // every PV's hook can enforce read_only / ACL / host-deny / putlog
        // before forwarding the put to upstream.
        let upstream = UpstreamManager::new(UpstreamManagerConfig {
            cache: cache.clone(),
            shadow_db: shadow_db.clone(),
            access: access.clone(),
            pvlist: pvlist.clone(),
            putlog: putlog.clone(),
            stats: stats.clone(),
            read_only: config.read_only,
            beacon_anomaly: beacon_anomaly.clone(),
            // B10: forward the upstream-side TLS config so the
            // gateway's CaClient to the real IOC can also use TLS,
            // independently of downstream TLS termination.
            #[cfg(feature = "ca-gateway-tls")]
            upstream_tls: config.upstream_tls.clone(),
            #[cfg(feature = "ca-gateway-tls")]
            upstream_tls_server_name: config.upstream_tls_server_name.clone(),
        })
        .await?;
        let upstream = Arc::new(upstream);

        // Downstream server — wrap each accepted client in TLS when
        // configured. Upstream traffic to the IOC is encrypted
        // independently via `GatewayConfig::upstream_tls` (B10,
        // wired into `UpstreamManager::new` above).
        let downstream = Arc::new({
            #[cfg(feature = "ca-gateway-tls")]
            {
                if let Some(ref tls) = config.tls {
                    DownstreamServer::new_with_tls(
                        shadow_db.clone(),
                        config.server_port,
                        tls.clone(),
                    )
                } else {
                    DownstreamServer::new(shadow_db.clone(), config.server_port)
                }
            }
            #[cfg(not(feature = "ca-gateway-tls"))]
            {
                DownstreamServer::new(shadow_db.clone(), config.server_port)
            }
        });

        // Now that the downstream CaServer is built, snapshot its
        // beacon-reset handle and install it on the throttle so honored
        // `request()` calls actually emit a beacon. Captured BEFORE
        // `downstream.run()` consumes the inner CaServer.
        if let Some(pulse) = downstream.beacon_anomaly_handle().await {
            beacon_anomaly.install_pulse(pulse);
        }

        let server = Self {
            config,
            pvlist,
            access,
            cache,
            shadow_db,
            upstream,
            downstream,
            stats,
            putlog,
            beacon_anomaly,
        };

        // Pre-register stats PVs in shadow database so downstream can read them
        server.stats.publish_initial(&server.shadow_db).await;

        // Install lazy search resolver: when an unknown name is searched
        // for, check the .pvlist and (if allowed) subscribe upstream.
        server.install_search_resolver().await;

        Ok(server)
    }

    /// Install the lazy search resolver into the shadow PvDatabase.
    ///
    /// This implements the equivalent of C++ ca-gateway's
    /// `gateServer::pvExistTest()` (gateServer.cc:1484), but at a
    /// different layer: C++ overrides `caServer::pvExistTest`, while
    /// epics-rs hooks `PvDatabase::set_search_resolver`. The effect is
    /// the same — when a downstream client searches for an unknown
    /// name, the gateway is given a chance to consult the `.pvlist`,
    /// subscribe upstream, and report whether the name became
    /// resolvable.
    ///
    /// Called once during build().
    async fn install_search_resolver(&self) {
        let pvlist = self.pvlist.clone();
        let upstream = self.upstream.clone();
        let stats = self.stats.clone();
        let beacon_anomaly = self.beacon_anomaly.clone();

        let resolver: epics_base_rs::server::database::SearchResolver = std::sync::Arc::new(
            move |name: String| -> std::pin::Pin<
                Box<dyn std::future::Future<Output = bool> + Send>,
            > {
                let pvlist = pvlist.clone();
                let upstream = upstream.clone();
                let stats = stats.clone();
                let beacon_anomaly = beacon_anomaly.clone();
                Box::pin(async move {
                    // 1. Check pvlist
                    let m = {
                        let pvlist = pvlist.load_full();
                        pvlist.match_name(&name)
                    };
                    let m = match m {
                        Some(m) => m,
                        None => return false,
                    };

                    // 2. Subscribe upstream — this also adds the PV to the
                    //    shadow database via UpstreamManager::ensure_subscribed.
                    //    Pass the matched ASG/ASL through so the per-PV
                    //    WriteHook can do the right ACL check.
                    if upstream
                        .ensure_subscribed(&m.resolved_name, m.asg.clone(), m.asl.unwrap_or(0))
                        .await
                        .is_err()
                    {
                        return false;
                    }

                    // 3. B-G9: trigger a beacon anomaly so other
                    //    gateway-aware downstream clients re-search
                    //    and discover this gateway as the server for
                    //    the just-added PV. Mirrors C++ ca-gateway
                    //    `gateServer::generateBeaconAnomaly` on the
                    //    add-PV path.
                    beacon_anomaly.request();

                    // 4. Stats: count this resolution
                    stats.record_event();
                    true
                })
            },
        );

        self.shadow_db.set_search_resolver(resolver).await;
    }

    /// Pre-subscribe to upstream PVs from the preload file.
    pub async fn preload_pvs(&self) -> BridgeResult<usize> {
        let path = match &self.config.preload_path {
            Some(p) => p,
            None => return Ok(0),
        };
        let content = std::fs::read_to_string(path)?;
        let mut count = 0;

        for line in content.lines() {
            let name = line.trim();
            if name.is_empty() || name.starts_with('#') {
                continue;
            }

            // Resolve through pvlist (alias or allow check)
            let m = {
                let pvlist = self.pvlist.load_full();
                pvlist.match_name(name)
            };
            let m = match m {
                Some(m) => m,
                None => continue, // Denied or not in list
            };

            self.upstream
                .ensure_subscribed(&m.resolved_name, m.asg.clone(), m.asl.unwrap_or(0))
                .await?;
            count += 1;
        }

        Ok(count)
    }

    /// Access the shadow database (for stats publication, testing).
    pub fn shadow_database(&self) -> &Arc<PvDatabase> {
        &self.shadow_db
    }

    /// Access the cache (for stats, introspection).
    pub fn cache(&self) -> &Arc<RwLock<PvCache>> {
        &self.cache
    }

    /// Access the pvlist slot (`ArcSwap` for atomic hot reload).
    pub fn pvlist(&self) -> &Arc<ArcSwap<PvList>> {
        &self.pvlist
    }

    /// Access the access-security config slot. SIGUSR1 `AS`
    /// (RELOAD_ACCESS) `store`s a fresh `Arc<AccessConfig>` here;
    /// in-flight puts that already loaded the previous one continue
    /// with the old rules, while later puts pick up the new ones.
    pub fn access(&self) -> &Arc<ArcSwap<AccessConfig>> {
        &self.access
    }

    /// Access stats.
    pub fn stats(&self) -> &Arc<Stats> {
        &self.stats
    }

    /// Access the put-event logger (if configured).
    pub fn putlog(&self) -> Option<&Arc<PutLog>> {
        self.putlog.as_ref()
    }

    /// Access the beacon anomaly throttle.
    pub fn beacon_anomaly(&self) -> &Arc<BeaconAnomaly> {
        &self.beacon_anomaly
    }

    /// Run the gateway daemon. Blocks until shutdown.
    pub async fn run(self) -> BridgeResult<()> {
        // Pre-load configured upstream PVs
        let preloaded = self.preload_pvs().await?;
        tracing::info!(preloaded, "ca-gateway-rs: preloaded upstream PVs");

        let downstream = self.downstream.clone();
        let cache = self.cache.clone();
        let upstream = self.upstream.clone();
        let stats = self.stats.clone();
        let shadow_db = self.shadow_db.clone();
        let timeouts = self.config.timeouts;
        let cleanup_interval = self.config.cleanup_interval;
        let stats_interval = self.config.stats_interval;
        let heartbeat_interval = self.config.heartbeat_interval;

        // Cleanup task
        let cache_for_cleanup = cache.clone();
        let upstream_for_cleanup = upstream.clone();
        let stats_for_cleanup = stats.clone();
        let cleanup_handle = tokio::spawn(async move {
            let mut tick = tokio::time::interval(cleanup_interval);
            tick.tick().await; // first tick is immediate, skip
            loop {
                tick.tick().await;
                // B5 RATE_STATS: count one gateway run-loop iteration.
                // The cleanup tick is the gateway's canonical periodic
                // maintenance loop — the tokio analogue of the C++
                // fdManager event-loop pass that drives gateServer::
                // loopCount.
                stats_for_cleanup.record_loop();
                let removed = cache_for_cleanup.write().await.cleanup(&timeouts).await;
                if !removed.is_empty() {
                    upstream_for_cleanup.sweep_orphaned().await;
                    tracing::info!(evicted = removed.len(), "ca-gateway-rs: cache eviction");
                }
            }
        });

        // Stats refresh task
        let cache_for_stats = cache.clone();
        let upstream_for_stats = upstream.clone();
        let stats_for_refresh = stats.clone();
        let db_for_stats = shadow_db.clone();
        let stats_handle = tokio::spawn(async move {
            let mut tick = tokio::time::interval(stats_interval);
            tick.tick().await;
            loop {
                tick.tick().await;
                let cache_size = cache_for_stats.read().await.len();
                let upstream_count = upstream_for_stats.subscription_count();
                stats_for_refresh
                    .refresh(&cache_for_stats, &db_for_stats, cache_size, upstream_count)
                    .await;
            }
        });

        // Heartbeat task
        let heartbeat_handle = if let Some(period) = heartbeat_interval {
            let stats_hb = stats.clone();
            let db_hb = shadow_db.clone();
            Some(tokio::spawn(async move {
                let mut tick = tokio::time::interval(period);
                tick.tick().await;
                loop {
                    tick.tick().await;
                    stats_hb.heartbeat_tick(&db_hb).await;
                }
            }))
        } else {
            None
        };

        // SIGUSR1 → command file processing (Unix only)
        let signal_handle = self.spawn_signal_handler();

        // Connection event subscriber.
        //
        // - `Connected`/`Disconnected`: per-host stats tracking (matches
        //   the C ca-gateway "connected client count" diagnostic PV).
        // - `ChannelCreated`/`ChannelCleared`: per-PV subscriber tracking
        //   for the cache FSM. A channel-create flips the corresponding
        //   `GwPvEntry` from `Inactive` → `Active`; a channel-clear
        //   reverses the transition once subscribers drop to zero.
        //   Without this wiring the `Active` state is unreachable and
        //   the C-gateway parity is incomplete (see review §3).
        //
        // Subscriber-id passed into `add_subscriber`/`remove_subscriber`
        // is a synthetic hash of `(peer, pv_name, cid)`. Including the
        // CA cid is critical: a single client can open multiple
        // channels to the same PV (camonitor + caget loop, etc.), and
        // hashing only `(peer, pv_name)` would collapse N channels
        // into one refcount slot — `Active` would flip back to
        // `Inactive` on the first CLEAR even with channels still open.
        //
        // `Lagged` is handled by replay (B11): `connection_events`
        // returns a `ReplayingReceiver` that, on a broadcast lag,
        // recovers the exact missed events from a bounded ring buffer
        // before resuming the live stream. The consumer below never
        // sees a silent gap, so the per-PV refcounts stay correct.
        // The only residual lossy case — a lag that overflows the
        // replay log — surfaces as `ConnEventRecv::GapTruncated` and
        // is logged.
        let conn_rx = downstream.connection_events().await;
        let conn_handle = if let Some(mut rx) = conn_rx {
            let stats_for_conn = stats.clone();
            let cache_for_conn = self.cache.clone();
            Some(tokio::spawn(async move {
                use super::downstream::ConnEventRecv;
                use epics_ca_rs::server::ServerConnectionEvent;
                use std::collections::HashMap;
                use std::collections::hash_map::DefaultHasher;
                use std::hash::{Hash, Hasher};
                fn synthetic_sid(peer: std::net::SocketAddr, pv: &str, cid: u32) -> u32 {
                    let mut h = DefaultHasher::new();
                    peer.hash(&mut h);
                    pv.hash(&mut h);
                    cid.hash(&mut h);
                    h.finish() as u32
                }
                // Per-peer subscription registry. On a hard close
                // (TCP RST, process kill) the underlying CaServer may
                // emit only `Disconnected(addr)` without the matching
                // ChannelCleared events — leaving cache subscriber
                // refcounts inflated and entries stuck in the Active
                // state. We mirror every Created here and drain the
                // peer's entries on Disconnected as a safety net.
                let mut peer_channels: HashMap<std::net::SocketAddr, Vec<(String, u32)>> =
                    HashMap::new();
                loop {
                    let event = match rx.recv().await {
                        ConnEventRecv::Event(ev) => ev,
                        ConnEventRecv::GapTruncated { missed } => {
                            // A lag overflowed the replay ring buffer
                            // — the only case where events are
                            // genuinely unrecoverable. Far rarer than
                            // the channel-depth lag that replay
                            // covers; warn so the operator notices.
                            tracing::warn!(
                                missed,
                                "ca-gateway-rs: connection-event lag exceeded the \
                                 replay log — per-PV refcount may be transiently \
                                 off until the next CREATE/CLEAR cycle"
                            );
                            continue;
                        }
                        ConnEventRecv::Closed => break,
                    };
                    match event {
                        ServerConnectionEvent::Connected(addr) => {
                            stats_for_conn.record_host(&addr.ip().to_string()).await;
                        }
                        ServerConnectionEvent::Disconnected(addr) => {
                            stats_for_conn.forget_host(&addr.ip().to_string()).await;
                            if let Some(channels) = peer_channels.remove(&addr) {
                                let cache = cache_for_conn.read().await;
                                for (pv_name, sid) in channels {
                                    if let Some(entry) = cache.get(&pv_name) {
                                        entry.write().await.remove_subscriber(sid);
                                    }
                                }
                            }
                        }
                        ServerConnectionEvent::ChannelCreated { peer, pv_name, cid } => {
                            let sid = synthetic_sid(peer, &pv_name, cid);
                            if let Some(entry) = cache_for_conn.read().await.get(&pv_name) {
                                entry.write().await.add_subscriber(sid);
                            }
                            peer_channels.entry(peer).or_default().push((pv_name, sid));
                        }
                        ServerConnectionEvent::ChannelCleared { peer, pv_name, cid } => {
                            let sid = synthetic_sid(peer, &pv_name, cid);
                            if let Some(entry) = cache_for_conn.read().await.get(&pv_name) {
                                entry.write().await.remove_subscriber(sid);
                            }
                            if let Some(channels) = peer_channels.get_mut(&peer) {
                                channels.retain(|(p, s)| !(p == &pv_name && *s == sid));
                                if channels.is_empty() {
                                    peer_channels.remove(&peer);
                                }
                            }
                        }
                        _ => {}
                    }
                }
            }))
        } else {
            None
        };

        // Run downstream CaServer until either the server returns
        // (fatal I/O) or SIGINT/SIGTERM arrives (B-G15: graceful
        // shutdown). On signal we tear down upstream subscriptions
        // first so the upstream IOC sees a clean disconnect, then
        // abort the auxiliary tasks.
        let downstream_result = {
            let downstream_run = downstream.run();
            tokio::pin!(downstream_run);
            let ctrl_c = tokio::signal::ctrl_c();
            tokio::pin!(ctrl_c);
            tokio::select! {
                r = &mut downstream_run => r,
                _ = &mut ctrl_c => {
                    tracing::info!("ca-gateway-rs: SIGINT received — shutting down");
                    self.upstream.shutdown().await;
                    Ok(())
                }
            }
        };

        // Cleanup
        cleanup_handle.abort();
        stats_handle.abort();
        if let Some(h) = heartbeat_handle {
            h.abort();
        }
        if let Some(h) = signal_handle {
            h.abort();
        }
        if let Some(h) = conn_handle {
            h.abort();
        }
        // B11: stop the connection-event forwarder task spawned by
        // `connection_events()` so it does not outlive the server.
        downstream.stop_connection_events().await;

        downstream_result
    }

    /// Spawn a Unix SIGUSR1 watcher that re-reads the command file.
    /// Returns None on non-Unix or when no command file is configured.
    #[cfg(unix)]
    fn spawn_signal_handler(&self) -> Option<tokio::task::JoinHandle<()>> {
        let cmd_path = self.config.command_path.clone()?;
        let pvlist_path = self.config.pvlist_path.clone();
        let access_path = self.config.access_path.clone();
        let cache = self.cache.clone();
        let pvlist = self.pvlist.clone();
        let access = self.access.clone();
        let upstream = self.upstream.clone();

        Some(tokio::spawn(async move {
            use tokio::signal::unix::{SignalKind, signal};
            let mut sigusr1 = match signal(SignalKind::user_defined1()) {
                Ok(s) => s,
                Err(e) => {
                    tracing::error!(error = %e, "ca-gateway-rs: failed to install SIGUSR1 handler");
                    return;
                }
            };
            let handler = CommandHandler::new(cache, pvlist, access, pvlist_path, access_path)
                .with_upstream(upstream);
            tracing::info!(
                command_file = %cmd_path.display(),
                "ca-gateway-rs: SIGUSR1 handler armed"
            );
            loop {
                if sigusr1.recv().await.is_none() {
                    break;
                }
                tracing::info!("ca-gateway-rs: SIGUSR1 received — processing command file");
                match handler.process_file(&cmd_path).await {
                    Ok(out) => {
                        if !out.is_empty() {
                            tracing::info!(output = %out.trim_end(), "ca-gateway-rs: command output");
                        }
                    }
                    Err(e) => {
                        tracing::warn!(error = %e, "ca-gateway-rs: command file error");
                    }
                }
            }
        }))
    }

    /// Stub for non-Unix platforms (no SIGUSR1).
    #[cfg(not(unix))]
    fn spawn_signal_handler(&self) -> Option<tokio::task::JoinHandle<()>> {
        None
    }
}

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

    #[tokio::test]
    async fn build_with_minimal_config() {
        let config = GatewayConfig {
            pvlist_content: Some("".to_string()),
            ..Default::default()
        };
        let server = GatewayServer::build(config).await;
        assert!(server.is_ok(), "build failed: {:?}", server.err());
    }

    #[tokio::test]
    async fn build_with_inline_pvlist() {
        let config = GatewayConfig {
            pvlist_content: Some(
                r#"
                EVALUATION ORDER ALLOW, DENY
                Beam:.* ALLOW BeamGroup 1
                test.* DENY
                "#
                .to_string(),
            ),
            ..Default::default()
        };
        let server = GatewayServer::build(config).await.unwrap();
        let pvlist = server.pvlist().load_full();
        assert!(pvlist.match_name("Beam:current").is_some());
        assert!(pvlist.match_name("test:foo").is_none());
    }

    /// B10: `GatewayServer::build` must succeed when an upstream TLS
    /// client config is supplied. The config flows
    /// `GatewayConfig::upstream_tls` → `UpstreamManagerConfig` →
    /// `CaClient::new_with_config`. No upstream IOC is contacted at
    /// build time, so this exercises the plumbing end to end.
    #[cfg(feature = "ca-gateway-tls")]
    #[tokio::test]
    async fn build_with_upstream_tls() {
        use epics_ca_rs::tls::{Roots, TlsConfig};
        let config = GatewayConfig {
            pvlist_content: Some("".to_string()),
            upstream_tls: Some(TlsConfig::client_from_roots(Roots::empty())),
            upstream_tls_server_name: Some("ioc.example.com".to_string()),
            ..Default::default()
        };
        let server = GatewayServer::build(config).await;
        assert!(
            server.is_ok(),
            "build with upstream TLS failed: {:?}",
            server.err()
        );
    }

    #[tokio::test]
    async fn build_unknown_acf_path_returns_error() {
        let config = GatewayConfig {
            access_path: Some(PathBuf::from("/nonexistent/file.acf")),
            ..Default::default()
        };
        let result = GatewayServer::build(config).await;
        assert!(result.is_err());
    }
}