polyc-tui 2026.8.3

//! In-process kube port-forward for the cockpit's localhost data endpoints.
//!
//! The cockpit talks to the control plane over two localhost sockets: the gRPC
//! `AgentService`/`ApprovalService` ([`AGENT_PORT`]) and the read-only forensics
//! HTTP server ([`FORENSICS_PORT`]). Rather than make the operator run `kubectl
//! port-forward` by hand and hand-set the `PC_TUI_*` env vars, the cockpit
//! forwards both ports itself — reusing the shared control-plane pod resolution
//! that `polychrome send` uses ([`polyc_controller::control_plane`]).
//!
//! [`start`] binds a localhost listener immediately — a local syscall with no
//! kube round-trip — so the cockpit's first paint never blocks on the apiserver.
//! It then spawns a SELF-HEALING accept loop: the control-plane pod is resolved
//! per inbound connection, so a pod restart or a dropped stream recovers on the
//! next dial without relaunching the cockpit (a dead upstream closes the local
//! socket, which makes the pooled forensics client reconnect → re-resolve).
//! Per-connection failures are surfaced on the action channel instead of being
//! swallowed.

use std::time::Duration;

use anyhow::{Context as _, Result};
use k8s_openapi::api::core::v1::Pod;
use kube::{Client, api::Api};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::UnboundedSender;
use tokio::task::JoinHandle;

use polyc_controller::control_plane;

use crate::action::Action;

/// The control-plane ports, re-exported so callers say `portforward::AGENT_PORT`
/// while the single source of truth lives in the shared control-plane module.
pub(crate) use polyc_controller::control_plane::{AGENT_PORT, FORENSICS_PORT};

/// Backoff after a hard `accept()` error before retrying, so a transiently
/// broken listener cannot busy-spin. The listener stays bound across the error,
/// so the cached localhost URL never goes dead.
const ACCEPT_RETRY_BACKOFF: Duration = Duration::from_secs(1);

/// A running port-forward: the localhost URL it serves plus the accept-loop
/// task. Dropping the guard aborts the accept loop (any in-flight per-connection
/// bridges end when their sockets close, or when the process exits); the `Model`
/// holds each `Forward` for the cockpit's lifetime.
pub(crate) struct Forward {
    /// `http://127.0.0.1:<local-port>` the forward listens on.
    pub url: String,
    /// The accept-loop driving the forward. Aborted on drop.
    task: JoinHandle<()>,
}

impl Drop for Forward {
    fn drop(&mut self) {
        self.task.abort();
    }
}

/// Bind a localhost listener and spawn the self-healing forward to
/// `remote_port` on the control-plane pod. `label` tags status-line errors
/// (e.g. `"agent"` / `"forensics"`); `tx` carries per-connection failures to the
/// loop so a broken forward is visible instead of silent.
///
/// Binding is a local syscall — no kube round-trip — so this returns instantly
/// and never blocks the cockpit's first paint. The pod is resolved lazily, per
/// connection, inside the accept loop.
///
/// # Errors
/// Returns an error only if the local listener cannot be bound or adopted.
pub(crate) fn start(
    client: Client,
    namespace: String,
    remote_port: u16,
    label: &'static str,
    tx: UnboundedSender<Action>,
) -> Result<Forward> {
    let std_listener =
        std::net::TcpListener::bind("127.0.0.1:0").context("bind local forward listener")?;
    std_listener
        .set_nonblocking(true)
        .context("set forward listener non-blocking")?;
    let local_port = std_listener.local_addr()?.port();
    let listener = TcpListener::from_std(std_listener).context("adopt forward listener")?;
    let task = tokio::spawn(accept_loop(
        client,
        namespace,
        listener,
        remote_port,
        label,
        tx,
    ));
    Ok(Forward {
        url: format!("http://127.0.0.1:{local_port}"),
        task,
    })
}

/// Accept connections for the cockpit's lifetime, bridging each onto its own
/// freshly-resolved kube port-forward. A hard `accept` error is reported and
/// retried after a short backoff rather than ending the loop, so the localhost
/// port stays bound and the cached URL never goes stale.
async fn accept_loop(
    client: Client,
    namespace: String,
    listener: TcpListener,
    remote_port: u16,
    label: &'static str,
    tx: UnboundedSender<Action>,
) {
    loop {
        match listener.accept().await {
            Ok((local, _peer)) => {
                tokio::spawn(bridge_conn(
                    client.clone(),
                    namespace.clone(),
                    local,
                    remote_port,
                    label,
                    tx.clone(),
                ));
            }
            Err(err) => {
                let _ = tx.send(Action::Error(format!("{label} forward accept: {err}")));
                tokio::time::sleep(ACCEPT_RETRY_BACKOFF).await;
            }
        }
    }
}

/// Resolve a Running control-plane pod, open a port-forward to `remote_port`,
/// and copy bytes until either side closes.
///
/// Resolving the pod PER CONNECTION is what makes the forward self-heal across
/// control-plane restarts: the next dial finds the new pod name. Setup failures
/// and port-forward channel errors are surfaced on the action channel — a plain
/// EOF from `copy_bidirectional` would otherwise hide them.
async fn bridge_conn(
    client: Client,
    namespace: String,
    mut local: TcpStream,
    remote_port: u16,
    label: &'static str,
    tx: UnboundedSender<Action>,
) {
    let pod = match control_plane::pick_control_plane_pod(&client, &namespace).await {
        Ok(pod) => pod,
        Err(err) => {
            let _ = tx.send(Action::Error(format!("{label} forward: {err}")));
            return;
        }
    };
    let pods: Api<Pod> = Api::namespaced(client, &namespace);
    let mut pf = match pods.portforward(&pod, &[remote_port]).await {
        Ok(pf) => pf,
        Err(err) => {
            let _ = tx.send(Action::Error(format!(
                "{label} port-forward to {pod}: {err}"
            )));
            return;
        }
    };
    let Some(mut upstream) = pf.take_stream(remote_port) else {
        let _ = tx.send(Action::Error(format!(
            "{label} port-forward returned no stream for :{remote_port}"
        )));
        return;
    };
    let error_fut = pf.take_error(remote_port);

    // Best-effort byte pump; a normal close returns Ok/EOF with no error frame.
    let _ = tokio::io::copy_bidirectional(&mut local, &mut upstream).await;

    // Surface a port-forward-level error (pod evicted, auth failure mid-stream)
    // that a plain EOF would otherwise hide — the diagnostic `send` also logs.
    if let Some(fut) = error_fut
        && let Some(msg) = fut.await
    {
        let _ = tx.send(Action::Error(format!(
            "{label} port-forward channel: {msg}"
        )));
    }
}