=== leveva/src/main.rs ===
//! `leveva` — the idiomatic-Rust IRC daemon binary.
//!
//! Boot path:
//!
//! 1. parse arguments (clap),
//! 2. read + validate the KDL config ([`leveva::config::Config::from_kdl`]),
//! 3. build a hot-reloadable TLS endpoint for each TLS listener ([`tls`]),
//! 4. bind every non-deferred listener and run an accept loop per socket,
//! 5. watch TLS cert/key files for changes (and `SIGHUP`) and reload them live.
//!
//! Accepted client connections (after TLS termination, if applicable) are served
//! through the registration state machine ([`Session`]) and receive the welcome
//! burst. Server-only ports are served through [`serve_server`] (inbound S2S links),
//! and operator [`CONNECT`](leveva::command) requests are dialed out through
//! [`dial_server`] — both share the `PASS`/`SERVER` handshake reader
//! ([`read_link_handshake`]) and the post-establishment lifecycle ([`serve_linked_peer`])
//! of the S2S link protocol ([`leveva::s2s`]). Authentication (ident, today) runs
//! **in-process** per connection via [`leveva_iauth`] — no child daemon — and its
//! verdict is folded into registration before the welcome burst.

use std::net::SocketAddr;
use std::path::PathBuf;
use std::process::ExitCode;
use std::sync::Arc;
use std::time::Duration;

use clap::Parser;
use leveva::config::{Config, WebsocketConfig};
use leveva::control::{self, ControlAction};
use leveva::heartbeat::{default_ping_freq, tick_interval};
use leveva::registry::Envelope;
use leveva::server::ServerContext;
use leveva::session::Session;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tracing::{error, info, warn};
use tracing_subscriber::EnvFilter;

use leveva::tls::{self, TlsEndpoint};

/// `--version` output: the crate version, the "From Within" tagline, and the
/// source repository.
const LONG_VERSION: &str = concat!(
    env!("CARGO_PKG_VERSION"),
    "\nFrom Within — https://tangled.org/xeiaso.net/leveva"
);

/// Command-line arguments.
#[derive(Parser, Debug)]
#[command(name = "leveva", version = LONG_VERSION, about = "An idiomatic-Rust IRC daemon")]
struct Cli {
    /// Path to the KDL configuration file.
    #[arg(short, long, default_value = "ircd.kdl")]
    config: PathBuf,

    /// Parse and validate the config, report warnings, then exit without binding.
    #[arg(long)]
    check: bool,

    /// Optional MOTD file; its lines are sent as 372/375/376 on registration.
    #[arg(long)]
    motd: Option<PathBuf>,

    /// Log verbosity: a level (error|warn|info|debug|trace) or a full RUST_LOG-style filter.
    #[arg(long, default_value = "info")]
    log_level: String,
}

/// A bound listening socket plus what the accept loop needs to serve it.
struct Bound {
    listener: TcpListener,
    addr: String,
    server_only: bool,
    /// `Some` for a TLS port; the accept loop wraps each connection with it.
    tls: Option<Arc<TlsEndpoint>>,
    ctx: std::sync::Arc<ServerContext>,
    /// The keepalive interval (the default connection class's `ping-freq`) applied
    /// to each client accepted here.
    ping_freq: Duration,
    /// `Some` for a WebSocket port: the accept loop runs the axum `/socket` server
    /// ([`leveva::websocket::server::serve_websocket`]) instead of the raw line framer.
    websocket: Option<WebsocketConfig>,
}

#[tokio::main]
async fn main() -> ExitCode {
    let cli = Cli::parse();

    // Initialize tracing first so even config-read/parse errors log through the
    // subscriber. `RUST_LOG` (if set) wins; otherwise the `--log-level` flag is the
    // default filter.
    let filter = EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new(&cli.log_level));
    tracing_subscriber::fmt().with_env_filter(filter).init();

    // Record where the config came from so `REHASH` can re-read it.
    control::set_config_path(cli.config.clone());

    let src = match std::fs::read_to_string(&cli.config) {
        Ok(s) => s,
        Err(e) => {
            error!(config = %cli.config.display(), error = %e, "cannot read config");
            return ExitCode::FAILURE;
        }
    };

    let config = match Config::from_kdl(&src) {
        Ok(c) => c,
        Err(e) => {
            error!(config = %cli.config.display(), error = %e, "invalid config");
            return ExitCode::FAILURE;
        }
    };

    for w in config.warnings() {
        warn!("{w}");
    }

    info!(
        server = config.server.name.as_str(),
        sid = config.server.sid.as_str(),
        "loaded config",
    );

    info!(
        ident = config.iauth.ident,
        "iauth configured (in-process)",
    );

    if cli.check {
        info!(config = %cli.config.display(), "config ok");
        return ExitCode::SUCCESS;
    }

    let created = format_created();
    // CLI `--motd` wins; otherwise fall back to the config's `options.motd`.
    let motd_path = cli
        .motd
        .clone()
        .or_else(|| config.options.motd.as_ref().map(PathBuf::from));
    let motd = motd_path
        .as_ref()
        .map(|p| control::load_motd_path(p));
    // Record the resolved path + seed the live MOTD store so `REHASH` can re-read the same
    // file (slice 65); the running server's MOTD / welcome burst read the live store.
    if let Some(p) = &motd_path {
        control::set_motd_path(p.clone());
    }
    control::seed_live_motd(motd.clone());
    let ctx = ServerContext::from_config(&config, created, motd);

    // The server-side keepalive interval. leveva does not yet match clients against
    // `allow → class`, so every client uses the default (first) class's `ping-freq`
    // (or the 120s built-in default if no classes are defined). When allow-matching
    // lands this moves into the session, keyed by the matched class.
    let ping_freq = default_ping_freq(&config.classes);

    // Build a TLS endpoint per TLS listener, then bind. The cert/key paths are
    // guaranteed present by config validation, but loading can still fail (missing
    // file, bad PEM) — that's a fatal boot error.
    let mut bound = Vec::new();
    let mut tls_endpoints = Vec::new();
    for l in &config.listeners {
        if l.delayed {
            continue;
        }

        let tls = if l.tls {
            let cert = PathBuf::from(l.tls_cert.as_deref().expect("validated present"));
            let key = PathBuf::from(l.tls_key.as_deref().expect("validated present"));
            match TlsEndpoint::new(cert, key) {
                Ok(ep) => {
                    let ep = Arc::new(ep);
                    tls_endpoints.push(Arc::clone(&ep));
                    Some(ep)
                }
                Err(e) => {
                    error!(port = l.port, error = %e, "TLS listener load failed");
                    return ExitCode::FAILURE;
                }
            }
        } else {
            None
        };

        let host = l.bind.as_deref().unwrap_or("0.0.0.0");
        let addr = format!("{host}:{}", l.port);
        match TcpListener::bind(&addr).await {
            Ok(listener) => {
                info!(
                    %addr,
                    tls = l.tls,
                    server_only = l.server_only,
                    "listening",
                );
                bound.push(Bound {
                    listener,
                    addr,
                    server_only: l.server_only,
                    tls,
                    ctx: Arc::clone(&ctx),
                    ping_freq,
                    websocket: l.websocket.clone(),
                });
            }
            Err(e) => {
                error!(%addr, error = %e, "cannot bind listener");
                return ExitCode::FAILURE;
            }
        }
    }

    if bound.is_empty() {
        error!("no listeners bound; nothing to do");
        return ExitCode::FAILURE;
    }

    // One accept loop per socket. A WebSocket port runs the axum `/socket` server; every
    // other port runs the raw IRC line framer.
    for b in bound {
        match b.websocket.clone() {
            Some(cfg) => {
                tokio::spawn(leveva::websocket::server::serve_websocket(
                    b.listener,
                    b.addr,
                    b.tls,
                    b.ctx,
                    b.ping_freq,
                    cfg,
                ));
            }
            None => {
                tokio::spawn(serve(b));
            }
        }
    }

    // Drain the outbound link plane: each `CONNECT` queues an `OutboundLink`, which this
    // task dials (one spawned task per link, so a slow handshake doesn't stall others).
    if let Some(mut rx) = leveva::link::take_receiver() {
        let dial_ctx = Arc::clone(&ctx);
        tokio::spawn(async move {
            while let Some(req) = rx.recv().await {
                tokio::spawn(dial_server(req, Arc::clone(&dial_ctx)));
            }
        });
    }

    // The cert/key watcher: reloads TLS endpoints on file change or SIGHUP.
    if !tls_endpoints.is_empty() {
        tokio::spawn(async move {
            if let Err(e) = tls::watch(tls_endpoints).await {
                warn!(error = %e, "TLS watcher stopped");
            }
        });
    }

    info!("running; Ctrl-C to stop");

    // Shut down on either Ctrl-C (treated as DIE) or an in-band control-plane request
    // (`DIE`/`RESTART` from an operator — see [`leveva::control`]).
    let action = tokio::select! {
        r = tokio::signal::ctrl_c() => {
            if let Err(e) = r {
                error!(error = %e, "failed to listen for shutdown signal");
                return ExitCode::FAILURE;
            }
            info!("Ctrl-C received");
            ControlAction::Die
        }
        a = control::wait() => a,
    };

    // The DIE/RESTART farewell NOTICEs are queued on each connection's mailbox as eject
    // frames; give them a brief window to flush + close before the runtime tears down.
    let grace = Duration::from_millis(200);
    match action {
        ControlAction::Die => {
            info!("shutting down");
            tokio::time::sleep(grace).await;
            ExitCode::SUCCESS
        }
        ControlAction::Restart => {
            info!("restarting");
            tokio::time::sleep(grace).await;
            reexec()
        }
    }
}

/// Re-exec the current binary with its original arguments (the `RESTART` action). On
/// success `exec` replaces this process image and never returns; any return is a failure.
/// The listening sockets are `CLOEXEC`, so the fresh image re-binds cleanly.
#[cfg(unix)]
fn reexec() -> ExitCode {
    use std::os::unix::process::CommandExt;
    let exe = match std::env::current_exe() {
        Ok(p) => p,
        Err(e) => {
            error!(error = %e, "restart: cannot find current executable");
            return ExitCode::FAILURE;
        }
    };
    let args: Vec<String> = std::env::args().skip(1).collect();
    let err = std::process::Command::new(exe).args(args).exec();
    error!(error = %err, "restart: exec failed");
    ExitCode::FAILURE
}

#[cfg(not(unix))]
fn reexec() -> ExitCode {
    error!("restart: re-exec is only supported on Unix");
    ExitCode::FAILURE
}

/// Accept connections on one socket forever, handing each to [`handle`].
async fn serve(b: Bound) {
    loop {
        match b.listener.accept().await {
            Ok((stream, peer)) => {
                let tls = b.tls.clone();
                let server_only = b.server_only;
                tokio::spawn(handle(
                    stream,
                    peer,
                    tls,
                    server_only,
                    b.ctx.clone(),
                    b.ping_freq,
                ));
            }
            Err(e) => {
                // A persistent accept error (e.g. fd exhaustion) shouldn't spin.
                warn!(addr = %b.addr, error = %e, "accept error");
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
        }
    }
}

/// Handle one accepted connection: complete TLS if applicable, then serve the
/// IRC protocol. Server-only ports do not yet speak S2S, so they are dropped.
async fn handle(
    stream: TcpStream,
    peer: SocketAddr,
    tls: Option<Arc<TlsEndpoint>>,
    server_only: bool,
    ctx: Arc<ServerContext>,
    ping_freq: Duration,
) {
    let host = peer.ip().to_string();
    if server_only {
        // A server-only port speaks the S2S link protocol, not the client protocol.
        match tls {
            Some(ep) => match ep.acceptor.accept(stream).await {
                Ok(tls_stream) => serve_server(tls_stream, host, ctx).await,
                Err(e) => warn!(%peer, error = %e, "TLS handshake failed"),
            },
            None => serve_server(stream, host, ctx).await,
        }
        return;
    }
    // The ident lookup needs both endpoints: the client's IP + source port (its query's
    // first field) and our local IP + listener port (the second). Captured here from the
    // raw socket, before any TLS wrap loses the addresses. A failed `local_addr` leaves a
    // blank/zero local end — the ident reply's port pair then won't match, so the lookup
    // simply declines (the `~user` fallback), which is the safe degenerate behaviour.
    let local = stream.local_addr().ok();
    let auth_ctx = leveva_iauth::ModuleCtx::new(
        peer.ip().to_string(),
        peer.port(),
        local.map(|a| a.ip().to_string()).unwrap_or_default(),
        local.map(|a| a.port()).unwrap_or(0),
    );
    match tls {
        Some(ep) => match ep.acceptor.accept(stream).await {
            Ok(tls_stream) => serve_client(tls_stream, host, ctx, ping_freq, auth_ctx).await,
            Err(e) => warn!(%peer, error = %e, "TLS handshake failed"),
        },
        None => serve_client(stream, host, ctx, ping_freq, auth_ctx).await,
    }
}

/// Drive one client connection until close, multiplexing two sources:
///
/// - **inbound socket bytes** → [`Session::feed`] → write the sender's own replies;
/// - **the mailbox** → wire bytes other connections delivered here (e.g. a
///   `PRIVMSG`) → write straight to the socket.
///
/// A third `select!` arm fires the **keepalive timer** ([`Session::check_ping`]):
/// an idle connection is pinged once and, if it stays silent through a second
/// `ping_freq`, dropped with a ping-timeout.
///
/// The session keeps its own mailbox sender, so the receiver never closes while the
/// connection lives; the `select!` branch stays live for the whole loop.
async fn serve_client<S>(
    mut stream: S,
    host: String,
    ctx: Arc<ServerContext>,
    ping_freq: Duration,
    auth_ctx: leveva_iauth::ModuleCtx,
) where
    S: AsyncRead + AsyncWrite + Unpin,
{
    info!(%host, "client connected");
    let mut session = Session::new(host.clone());
    let mut mailbox = session
        .take_mailbox()
        .expect("a fresh session always has its mailbox receiver");
    session.start_heartbeat(ping_freq, leveva::clock::unixnano());
    // Logged once, when the connection transitions to fully registered.
    let mut announced_registration = false;

    // When any auth module is configured, run the pipeline concurrently with the
    // NICK/USER handshake (it needs only the addresses, known already). Its verdict is
    // folded in via `Session::resolve_auth`, which finalizes registration if it was
    // waiting (and can refuse it if a denying module — dnsbl — fired). With no modules
    // there is no task and registration finalizes synchronously. The modules are shared
    // (`Arc`), so this clones the handles rather than rebuilding the resolver per client.
    let mut auth_task = if ctx.auth.any() {
        let pipeline = leveva_iauth::AuthPipeline::new(ctx.auth.modules.clone(), ctx.auth.timeout);
        Some(tokio::spawn(async move { pipeline.run(auth_ctx).await }))
    } else {
        None
    };

    // The keepalive check runs on a coarse timer; `check_ping` reads the real clock
    // each tick, so the tick cadence only bounds detection latency, not correctness.
    let mut ticker = tokio::time::interval(tick_interval(ping_freq));
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    ticker.tick().await; // the first interval tick is immediate — consume it

    let mut buf = [0u8; 4096];
    loop {
        tokio::select! {
            read = stream.read(&mut buf) => {
                let n = match read {
                    Ok(0) | Err(_) => break,
                    Ok(n) => n,
                };
                // Any inbound bytes count as liveness (resets the ping timer).
                session.record_activity(leveva::clock::unixnano());
                let out = session.feed(&buf[..n], &ctx);
                if !announced_registration && session.is_registered() {
                    announced_registration = true;
                    let nick = session.current_nick().unwrap_or("*");
                    let uid = session.current_uid().map(ToString::to_string).unwrap_or_default();
                    info!(uid, nick, %host, "client registered");
                }
                if !out.is_empty() && stream.write_all(&out).await.is_err() {
                    break;
                }
                if session.should_close() {
                    let _ = stream.flush().await;
                    break;
                }
            }
            Some(item) = mailbox.recv() => {
                match item {
                    Envelope::Wire(bytes) => {
                        if stream.write_all(&bytes).await.is_err() {
                            break;
                        }
                    }
                    // A KILL from another connection: write the final line, then close,
                    // relaying QUIT :reason to channel co-members via release().
                    Envelope::Eject { wire, reason } => {
                        let _ = stream.write_all(&wire).await;
                        let _ = stream.flush().await;
                        session.force_close(reason);
                        break;
                    }
                }
            }
            _ = ticker.tick() => {
                let out = session.check_ping(leveva::clock::unixnano(), &ctx);
                if !out.is_empty() && stream.write_all(&out).await.is_err() {
                    break;
                }
                if session.should_close() {
                    let _ = stream.flush().await;
                    break;
                }
            }
            // The auth pipeline reported back (ident finished, declined, or timed out).
            // Fires at most once — `auth_task` is cleared so the completed handle is
            // never awaited again. A panicked/cancelled task yields a blank verdict.
            v = async { auth_task.as_mut().unwrap().await }, if auth_task.is_some() => {
                auth_task = None;
                let verdict = v.unwrap_or_default();
                let out = session.resolve_auth(verdict, &ctx);
                if !announced_registration && session.is_registered() {
                    announced_registration = true;
                    let nick = session.current_nick().unwrap_or("*");
                    let uid = session.current_uid().map(ToString::to_string).unwrap_or_default();
                    info!(uid, nick, %host, "client registered");
                }
                if !out.is_empty() && stream.write_all(&out).await.is_err() {
                    break;
                }
                if session.should_close() {
                    let _ = stream.flush().await;
                    break;
                }
            }
        }
    }
    match (session.current_uid(), session.current_nick()) {
        (Some(uid), Some(nick)) => info!(%uid, nick, %host, "client disconnected"),
        _ => info!(%host, "client disconnected (unregistered)"),
    }
    session.release(&ctx);
}

/// Read the `PASS` + `SERVER` link handshake off `stream` until it resolves.
///
/// `\n`/`\r\n`-tolerant line framing, identical to the client framer. A `PASS` is stashed
/// in the [`Handshake`](leveva::s2s::Handshake); a `SERVER` line authenticates the whole
/// thing against the matching [`Connect`](leveva::config::Connect) block. On success this
/// returns the [`Accepted`](leveva::s2s::Accepted) plus any bytes already buffered *past*
/// the `SERVER` line (the start of the peer's burst, if it pipelined it). On a rejected
/// handshake it writes the `ERROR :…` close line and returns `None`; on EOF/socket error
/// it returns `None` with nothing written.
///
/// This is the side-agnostic core shared by [`serve_server`] (we were dialed — the caller
/// writes our `PASS`/`SERVER` reply *after* this returns) and [`dial_server`] (we dialed —
/// the caller already sent our greeting *before* calling this, so it discards
/// `Accepted::reply`). The peer's `PASS`/`SERVER` is validated identically in both
/// directions, so only one reader is needed.
async fn read_link_handshake<S>(
    stream: &mut S,
    host: &str,
    ctx: &Arc<ServerContext>,
) -> Option<(leveva::s2s::Accepted, Vec<u8>)>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    use leveva::s2s::{Handshake, HandshakeOutcome};

    let mut buf = [0u8; 4096];
    let mut pending: Vec<u8> = Vec::new();
    let mut handshake = Handshake::new(host.to_string());

    loop {
        while let Some(pos) = pending.iter().position(|&b| b == b'\n') {
            let mut line: Vec<u8> = pending.drain(..=pos).collect();
            line.pop();
            if line.last() == Some(&b'\r') {
                line.pop();
            }
            let text = String::from_utf8_lossy(&line);
            let msg = match leveva::Message::parse(&text) {
                Ok(m) => m,
                Err(_) => continue,
            };
            match msg.command().to_ascii_uppercase().as_str() {
                "PASS" => {
                    if let Err(err) = handshake.handle_pass(&msg) {
                        let _ = stream.write_all(&err).await;
                        let _ = stream.flush().await;
                        return None;
                    }
                }
                "SERVER" => {
                    // `handle_server` consumes the handshake; take it (a fresh default is
                    // left behind, unused) so the borrow checker is happy on either arm.
                    let hs = std::mem::take(&mut handshake);
                    match hs.handle_server(&msg, ctx) {
                        HandshakeOutcome::Accepted(accepted) => {
                            return Some((*accepted, pending));
                        }
                        HandshakeOutcome::Rejected(rejected) => {
                            warn!(%host, reason = %rejected.reason, "S2S handshake rejected");
                            let _ = stream.write_all(&rejected.reply).await;
                            let _ = stream.flush().await;
                            return None;
                        }
                    }
                }
                // Any other verb before establishment is out of sequence; ignore it (the
                // oracle's pre-registration server parse drops unknown verbs).
                _ => {}
            }
        }
        let n = match stream.read(&mut buf).await {
            Ok(0) | Err(_) => return None,
            Ok(n) => n,
        };
        pending.extend_from_slice(&buf[..n]);
    }
}

/// Drive one **inbound** linked-peer connection (a `server_only` port): a peer dialed us.
/// Run the [`Handshake`](leveva::s2s::Handshake); on success write our `PASS`/`SERVER`
/// reply (the passive side replies *after* validating the peer), then hand off to
/// [`serve_linked_peer`] for the burst + steady-state dispatch.
async fn serve_server<S>(mut stream: S, host: String, ctx: Arc<ServerContext>)
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    info!(%host, "S2S link accepted; awaiting PASS/SERVER");

    match read_link_handshake(&mut stream, &host, &ctx).await {
        Some((accepted, pending)) => {
            // Passive side: write our PASS/SERVER reply before bursting.
            if stream.write_all(&accepted.reply).await.is_err() {
                return;
            }
            info!(
                %host,
                server = %accepted.remote.name,
                sid = %accepted.remote.sid,
                "S2S link established",
            );
            serve_linked_peer(
                stream,
                host,
                ctx,
                accepted.link,
                accepted.mailbox_rx,
                pending,
            )
            .await;
        }
        None => info!(%host, "S2S link closed"),
    }
}

/// Initiate an **outbound** server link (the `CONNECT` active side). Re-resolve the
/// `connect{}` block by name (so a config change between command and dial is honoured),
/// dial it, present our `PASS`/`SERVER` greeting *first* (the active side speaks first),
/// then validate the peer's reply through the shared [`read_link_handshake`] and hand off
/// to [`serve_linked_peer`]. We already sent our greeting, so the [`Accepted::reply`] the
/// reader produces is discarded.
async fn dial_server(req: leveva::link::OutboundLink, ctx: Arc<ServerContext>) {
    let connect = match ctx
        .stats_conf
        .links
        .iter()
        .find(|c| c.name.as_irc_str() == req.name.as_irc_str())
        .cloned()
    {
        Some(c) => c,
        None => {
            warn!(server = %req.name, "CONNECT: no matching connect block (config changed?)");
            return;
        }
    };

    info!(server = %req.name, host = %req.host, port = req.port, "CONNECT dialing");
    let mut stream = match TcpStream::connect((req.host.as_str(), req.port)).await {
        Ok(s) => s,
        Err(e) => {
            warn!(
                server = %req.name,
                host = %req.host,
                port = req.port,
                error = %e,
                "CONNECT failed",
            );
            return;
        }
    };

    // The active side speaks first: our PASS + SERVER greeting.
    let greeting = leveva::s2s::build_link_greeting(&ctx, &connect);
    if stream.write_all(&greeting).await.is_err() {
        warn!(server = %req.name, "CONNECT: failed to send greeting");
        return;
    }

    let host = req.host.clone();
    match read_link_handshake(&mut stream, &host, &ctx).await {
        Some((accepted, pending)) => {
            // We greeted first, so `accepted.reply` is redundant here — discard it.
            info!(
                server = %accepted.remote.name,
                sid = %accepted.remote.sid,
                "CONNECT established",
            );
            serve_linked_peer(
                stream,
                host,
                ctx,
                accepted.link,
                accepted.mailbox_rx,
                pending,
            )
            .await;
        }
        None => warn!(server = %req.name, "CONNECT: did not complete PASS/SERVER handshake"),
    }
}

/// The shared post-establishment lifecycle for a linked peer, regardless of which side
/// dialed: burst our state onto the link, register the peer in the routing table, then
/// multiplex two sources until the socket closes — finally SQUIT the peer's subtree.
///
/// - **inbound socket bytes** → [`leveva::s2s::dispatch`] (burst absorption + steady-state
///   relay) → write any reply (e.g. `EOBACK`) back to the peer;
/// - **the link mailbox** → wire frames other connections routed to this peer (our burst,
///   then any relayed `PRIVMSG`/`JOIN`/`MODE`/… a local client's action produced) → write
///   straight to the socket.
///
/// `pending` is any bytes the handshake reader buffered past the `SERVER` line (an
/// already-pipelined burst), drained before each `select!`. The peer is registered in
/// [`ServerContext::peers`] for the life of the connection so the relay senders can reach it,
/// and unlinked on teardown.
async fn serve_linked_peer<S>(
    mut stream: S,
    host: String,
    ctx: Arc<ServerContext>,
    link: leveva::s2s::PeerLink,
    mut mailbox_rx: tokio::sync::mpsc::UnboundedReceiver<Envelope>,
    mut pending: Vec<u8>,
) where
    S: AsyncRead + AsyncWrite + Unpin,
{
    // Burst our state onto the mailbox (drained to the socket by the loop below), then
    // register the peer so local handlers can route relayed frames to it.
    leveva::s2s::send_burst(&link, &ctx);
    ctx.peers.link(link.sid.clone(), link.mailbox_tx.clone());

    let mut link = link;
    let mut buf = [0u8; 4096];
    'conn: loop {
        // Drain any complete inbound lines buffered so far (the pipelined burst first, then
        // whatever the socket reads below appends).
        while let Some(pos) = pending.iter().position(|&b| b == b'\n') {
            let mut line: Vec<u8> = pending.drain(..=pos).collect();
            line.pop();
            if line.last() == Some(&b'\r') {
                line.pop();
            }
            let text = String::from_utf8_lossy(&line);
            if let Ok(msg) = leveva::Message::parse(&text) {
                let out = leveva::s2s::dispatch(&mut link, &msg, &ctx);
                if !out.is_empty() && stream.write_all(&out).await.is_err() {
                    break 'conn;
                }
            }
        }
        tokio::select! {
            read = stream.read(&mut buf) => {
                match read {
                    Ok(0) | Err(_) => break,
                    Ok(n) => pending.extend_from_slice(&buf[..n]),
                }
            }
            Some(env) = mailbox_rx.recv() => {
                match env {
                    Envelope::Wire(bytes) => {
                        if stream.write_all(&bytes).await.is_err() {
                            break;
                        }
                    }
                    // A peer link is never the target of a KILL, but handle an eject
                    // defensively by writing the line and closing.
                    Envelope::Eject { wire, .. } => {
                        let _ = stream.write_all(&wire).await;
                        let _ = stream.flush().await;
                        break;
                    }
                }
            }
        }
    }

    // Unlink the peer from the routing table, then SQUIT its subtree: an abrupt socket drop
    // tears down every remote user behind it (a QUIT relayed to each of their local channel
    // co-members), frees their registry nicks, and drops the server nodes from the mirror.
    ctx.peers.unlink(&link.sid);
    let removed = leveva::s2s::squit::squit_subtree(&ctx, &link.sid, "Connection reset by peer");
    if !removed.is_empty() {
        info!(
            %host,
            sid = %link.sid,
            removed = removed.len(),
            "S2S link split",
        );
    }
    drop(mailbox_rx);
    info!(%host, "S2S link closed");
}

/// A volatile human-readable boot timestamp for RPL_CREATED (masked in goldens).
fn format_created() -> String {
    use std::time::{SystemTime, UNIX_EPOCH};
    let secs = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    format!("at unix time {secs}")
}
=== cargo build --bin check ===
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.12s
=== Dockerfile ===
# Build the `leveva` IRC daemon binary.
#
# leveva is a workspace member, so the whole workspace is copied in: cargo needs
# every member manifest to resolve the lockfile, even though `-p leveva` only
# compiles leveva and its path dependency (leveva-iauth). The build is pure Rust
# (rustls/ring for TLS, no system OpenSSL), so the runtime image needs nothing
# beyond glibc.

FROM rust:1-bookworm AS build
WORKDIR /src

# Copy the full workspace (path deps + lockfile) and build only leveva.
COPY . .
RUN cargo build --release --locked -p leveva --bin leveva

# ---

FROM debian:bookworm-slim AS runtime
RUN useradd --system --user-group --home-dir /var/lib/leveva --create-home leveva

COPY --from=build /src/target/release/leveva /usr/local/bin/leveva

# Config lives in /etc/leveva; runtime state (none today) under /var/lib/leveva.
WORKDIR /var/lib/leveva
USER leveva

# 6667 = clients, 6697 = TLS clients, 6067 = server-only (S2S) links.
EXPOSE 6667 6697 6067

CMD ["/usr/local/bin/leveva"]
