frame-cli 0.4.0

//! `frame dev` — one command to a running node (F-7b): boot through the
//! shared spine, then stay attached — filesystem events rebuild changed
//! component inputs and successful candidates hot-swap against the landed
//! lifecycle, while build or swap failure is typed and loud and the node
//! serves the last known-good bytecode.
//!
//! The startup output NAMES the resolved application root as its first
//! line (ruling C1's condition): resolution walks parent directories
//! exactly as every landed verb does, and a long-lived loop that pushes
//! bytecode must say where it resolved to before the first swap, not
//! after.

use std::path::PathBuf;
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use frame_host::dev::{
    CandidateBytes, DevManagementConversation, DevReply, DevRequest, DevResumeStore,
    parse_management_line,
};
use notify::Watcher;

use crate::dev::build::{BuildError, build_candidate, snapshot_component};
use crate::dev::debounce::BuildGeneration;
use crate::dev::session::{LoopInput, Session, SessionEffects, SessionExit};
use crate::dev::watch::{WatchedSet, assert_native_backend};
use crate::doctor;
use crate::error::CliError;
use crate::preflight::ensure_ports_available;
use crate::project::App;

use super::boot::{BootOutcome, spawn_host, stop_child, wait_until_ready};
use super::build::{build_application, host_binary};

/// Deadline for one S1 stage-and-activate request: the node's whole
/// barrier (drain, stop, purge, load, start, both witnesses) must answer
/// inside it. Elapse is the typed `DeadlineElapsed` OUTCOME (R4), printed
/// and survived — never an invented failure.
const PUSH_DEADLINE: Duration = Duration::from_secs(60);

/// The dev session's staging area, inside the application's own
/// `.frame/` (never a system temp dir): per-generation snapshot copies.
const STAGING_DIR: &str = ".frame/dev";

/// Runs the dev loop. `quiet_millis` is the semantic edit-quiet deadline
/// (the named CLI parameter; 100 ms is the declared, visible default).
///
/// # Errors
///
/// Refuses on missing prerequisites, a non-native watch backend, or a
/// failed initial build/boot/attach; fails typed when the node dies (the
/// loop never auto-restarts it).
pub fn dev(quiet_millis: u64) -> Result<(), CliError> {
    let app = App::find_from_current_dir()?;
    // Ruling C1's condition: the resolved root is the FIRST line.
    println!("frame dev: application root {}", app.root.display());
    let backend = assert_native_backend().map_err(|error| CliError::Dev {
        detail: error.to_string(),
    })?;
    println!("frame dev: watch backend {backend:?} (native events)");
    doctor::require(doctor::BUILD_PREREQUISITES, "frame dev")?;
    build_application(&app)?;
    let config = app.load_config()?;
    ensure_ports_available(&config)?;
    let binary = host_binary(&app);
    let gleam_name = component_gleam_name(&app)?;

    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .map_err(|source| CliError::Runtime { source })?;
    let (input_tx, input_rx) = mpsc::channel::<LoopInput>();

    // Boot the node (--dev) through the shared spine; learn the URL and
    // the management handoff from its own stdout.
    let started = runtime.block_on(start_dev_node(&app, &binary, input_tx.clone()))?;
    println!(
        "frame dev: serving at http://{}  (Ctrl-C to stop)",
        started.page_addr
    );

    // Join the management conversation the node advertised.
    let attachment = frame_conv::BusAttachment {
        endpoint: started.management_endpoint.clone(),
        session_capacity: 1,
        operation_window_millis: 10,
        inbound_buffer: 16,
        answer_window: Duration::from_secs(30),
    };
    let conversation = DevManagementConversation::join(
        &attachment,
        started.conversation,
        DevResumeStore::default(),
    )
    .map_err(|error| CliError::Dev {
        detail: format!("joining the management conversation: {error}"),
    })?;
    println!("frame dev: RUNNING CURRENT generation 0 (the embedded boot bytes)");

    // Signals feed the loop as inputs so teardown is always the ordered
    // path.
    spawn_signal_bridge(&runtime, input_tx.clone());

    // The watcher registers on component/ recursively; classification
    // keeps the set honest (build outputs never dirty).
    let set = WatchedSet::of_project(&app.root);
    let watch_tx = input_tx.clone();
    let mut watcher = notify::recommended_watcher(move |event| {
        let _ = watch_tx.send(LoopInput::Fs(event));
    })
    .map_err(|error| CliError::Dev {
        detail: format!("establishing the watch: {error}"),
    })?;
    watcher
        .watch(&set.watch_root, notify::RecursiveMode::Recursive)
        .map_err(|error| CliError::Dev {
            detail: format!("watching {}: {error}", set.watch_root.display()),
        })?;

    let effects = RealEffects {
        component_dir: set.watch_root.clone(),
        staging_root: app.root.join(STAGING_DIR),
        gleam_name,
        input_tx: input_tx.clone(),
        candidates: Arc::new(Mutex::new(std::collections::HashMap::new())),
        conversation,
        watcher: Some(watcher),
        watch_root: set.watch_root.clone(),
    };
    let quiet = Duration::from_millis(quiet_millis);
    let mut session = Session::new(set, quiet, effects);
    let exit = session.run(&input_rx);

    match exit {
        SessionExit::Interrupted => {
            // The operator's stop: ordered teardown through the spine.
            runtime.block_on(started.stop(&binary))
        }
        SessionExit::NodeFailed { detail } => {
            eprintln!("frame dev: NODE FAILED — {detail}");
            eprintln!("frame dev: watching stopped; fix the cause and rerun `frame dev`");
            // Reap whatever the node left (it may already be gone).
            let _ = runtime.block_on(started.stop(&binary));
            Err(CliError::Dev {
                detail: format!("NODE FAILED: {detail}"),
            })
        }
        SessionExit::WatchRootGone { path } => {
            eprintln!(
                "frame dev: WATCHING HALTED — {} disappeared; restore it, then restart `frame dev`",
                path.display()
            );
            // The node keeps serving last good until the operator stops.
            passive_serve(&input_rx);
            runtime.block_on(started.stop(&binary))
        }
        SessionExit::ResubscribeFailed { detail } => {
            eprintln!(
                "frame dev: WATCHING HALTED — re-subscribing after watcher loss failed: {detail}; \
                 the node serves last good; restart `frame dev` to resume watching"
            );
            passive_serve(&input_rx);
            runtime.block_on(started.stop(&binary))
        }
        SessionExit::InputsClosed => runtime.block_on(started.stop(&binary)),
    }
}

/// After watching halts, the node keeps serving: wait (blocking — no
/// timer) for the operator's signal or the node's own exit.
fn passive_serve(inputs: &mpsc::Receiver<LoopInput>) {
    while let Ok(input) = inputs.recv() {
        match input {
            LoopInput::Interrupted => return,
            LoopInput::NodeFailed { detail } => {
                eprintln!("frame dev: NODE FAILED — {detail}");
                return;
            }
            // Watching is over: filesystem noise and stale build
            // completions have no consumer.
            LoopInput::Fs(_) | LoopInput::BuildCompleted { .. } => {}
        }
    }
}

/// The booted dev node: what startup learned from its stdout, plus the
/// stop request channel into its monitor task.
struct StartedNode {
    page_addr: std::net::SocketAddr,
    management_endpoint: String,
    conversation: frame_conv::id::ConversationId,
    stop_tx: tokio::sync::oneshot::Sender<tokio::sync::oneshot::Sender<Result<(), CliError>>>,
}

impl StartedNode {
    /// Ordered stop through the monitor task (one SIGTERM, deadline,
    /// verdict) — the same single teardown implementation `frame run`
    /// uses.
    async fn stop(self, binary: &std::path::Path) -> Result<(), CliError> {
        let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
        if self.stop_tx.send(reply_tx).is_err() {
            // The monitor already exited: the child is gone (its exit was
            // reported as NodeFailed).
            return Ok(());
        }
        reply_rx.await.unwrap_or_else(|_| {
            Err(CliError::Dev {
                detail: format!(
                    "the node monitor dropped the stop verdict for {}",
                    binary.display()
                ),
            })
        })
    }
}

/// Boots the node with `--dev`, pumps its stdout (forwarding every line),
/// and resolves once BOTH the page URL and the management handoff line
/// have been seen and the page address accepts connections.
async fn start_dev_node(
    app: &App,
    binary: &PathBuf,
    input_tx: mpsc::Sender<LoopInput>,
) -> Result<StartedNode, CliError> {
    use tokio::io::AsyncBufReadExt;
    use tokio::signal::unix::{SignalKind, signal};

    let mut interrupt =
        signal(SignalKind::interrupt()).map_err(|source| CliError::Runtime { source })?;
    let mut terminate =
        signal(SignalKind::terminate()).map_err(|source| CliError::Runtime { source })?;

    let mut child = spawn_host(&app.root, binary, &["--dev"])?;
    let stdout = child.stdout.take().ok_or_else(|| CliError::CommandSpawn {
        command: binary.display().to_string(),
        source: std::io::Error::other("child host stdout was not captured"),
    })?;

    // One pump for the child's whole lifetime: forwards every line,
    // detects the URL line and the management handoff line.
    let (url_tx, url_rx) = tokio::sync::oneshot::channel();
    let (management_tx, management_rx) = tokio::sync::oneshot::channel();
    tokio::spawn(async move {
        let mut lines = tokio::io::BufReader::new(stdout).lines();
        let mut url_tx = Some(url_tx);
        let mut management_tx = Some(management_tx);
        loop {
            match lines.next_line().await {
                Ok(Some(line)) => {
                    println!("{line}");
                    if let Some(addr) = super::boot::parse_serving_url(&line)
                        && let Some(sender) = url_tx.take()
                    {
                        let _ = sender.send(addr);
                    }
                    if let Some(handoff) = parse_management_line(&line)
                        && let Some(sender) = management_tx.take()
                    {
                        let _ = sender.send(handoff);
                    }
                }
                Ok(None) => return,
                Err(error) => {
                    eprintln!("frame dev: reading host output failed: {error}");
                    return;
                }
            }
        }
    });

    let page_addr =
        match wait_until_ready(&mut child, url_rx, &mut interrupt, &mut terminate).await? {
            BootOutcome::Ready(addr) => addr,
            BootOutcome::Signaled => {
                return stop_child(binary, child).await.and(Err(CliError::Dev {
                    detail: "interrupted during boot".to_owned(),
                }));
            }
        };
    let (management_endpoint, conversation) =
        management_rx.await.map_err(|_| CliError::CommandFailed {
            command: "the application host".to_owned(),
            status: "exited before announcing its dev management conversation".to_owned(),
        })?;

    // The monitor owns the child from here: it reports an unexpected exit
    // as NODE FAILED, or executes the one ordered stop on request.
    let (stop_tx, stop_rx) =
        tokio::sync::oneshot::channel::<tokio::sync::oneshot::Sender<Result<(), CliError>>>();
    let monitor_binary = binary.clone();
    tokio::spawn(async move {
        tokio::select! {
            status = child.wait() => {
                let detail = match status {
                    Ok(status) => format!("the node exited: {status}"),
                    Err(error) => format!("waiting on the node failed: {error}"),
                };
                let _ = input_tx.send(LoopInput::NodeFailed { detail });
            }
            request = stop_rx => {
                if let Ok(reply) = request {
                    let verdict = stop_child(&monitor_binary, child).await;
                    let _ = reply.send(verdict);
                }
            }
        }
    });

    Ok(StartedNode {
        page_addr,
        management_endpoint,
        conversation,
        stop_tx,
    })
}

/// Bridges Ctrl-C/SIGTERM into the session loop as ordered inputs.
fn spawn_signal_bridge(runtime: &tokio::runtime::Runtime, input_tx: mpsc::Sender<LoopInput>) {
    runtime.spawn(async move {
        use tokio::signal::unix::{SignalKind, signal};
        let (Ok(mut interrupt), Ok(mut terminate)) = (
            signal(SignalKind::interrupt()),
            signal(SignalKind::terminate()),
        ) else {
            return;
        };
        tokio::select! {
            _ = interrupt.recv() => {}
            _ = terminate.recv() => {}
        }
        let _ = input_tx.send(LoopInput::Interrupted);
    });
}

/// Reads the component's gleam package name from its manifest — the name
/// the landed ebin shape buries in its path.
fn component_gleam_name(app: &App) -> Result<String, CliError> {
    let manifest = app.root.join("component/gleam.toml");
    let content = std::fs::read_to_string(&manifest).map_err(|source| CliError::Dev {
        detail: format!("reading {}: {source}", manifest.display()),
    })?;
    content
        .lines()
        .find_map(|line| {
            let (key, value) = line.split_once('=')?;
            if key.trim() != "name" {
                return None;
            }
            Some(value.trim().trim_matches('"').to_owned())
        })
        .ok_or_else(|| CliError::Dev {
            detail: format!("{} declares no package name", manifest.display()),
        })
}

/// The real session effects: builds run on worker threads against staged
/// snapshots; completed candidates push through S1.
struct RealEffects {
    component_dir: PathBuf,
    staging_root: PathBuf,
    gleam_name: String,
    input_tx: mpsc::Sender<LoopInput>,
    /// Built candidates awaiting push, by generation.
    candidates: Arc<Mutex<std::collections::HashMap<u64, CandidateBytes>>>,
    conversation: DevManagementConversation<DevResumeStore>,
    watcher: Option<notify::RecommendedWatcher>,
    watch_root: PathBuf,
}

impl SessionEffects for RealEffects {
    fn start_build(&mut self, generation: BuildGeneration) {
        println!("frame dev: BUILDING generation {}", generation.0);
        let component_dir = self.component_dir.clone();
        let staging_root = self.staging_root.clone();
        let gleam_name = self.gleam_name.clone();
        let input_tx = self.input_tx.clone();
        let candidates = Arc::clone(&self.candidates);
        std::thread::spawn(move || {
            let result = build_one(
                &component_dir,
                &staging_root,
                &gleam_name,
                generation,
                &candidates,
            );
            let _ = input_tx.send(LoopInput::BuildCompleted { generation, result });
        });
    }

    fn discard_stale(&mut self, generation: BuildGeneration) {
        println!(
            "frame dev: generation {} superseded before activation; discarded",
            generation.0
        );
        if let Ok(mut candidates) = self.candidates.lock() {
            candidates.remove(&generation.0);
        }
    }

    fn push_candidate(&mut self, generation: BuildGeneration) -> Result<(), String> {
        let Some(candidate) = self
            .candidates
            .lock()
            .ok()
            .and_then(|mut candidates| candidates.remove(&generation.0))
        else {
            return Err(format!(
                "generation {} completed without stored candidate bytes",
                generation.0
            ));
        };
        println!("frame dev: RELOADING generation {}", generation.0);
        let outcome = self
            .conversation
            .handle_mut()
            .request::<DevRequest, DevReply>(
                &DevRequest::StageAndActivate(candidate),
                PUSH_DEADLINE,
            )
            .map_err(|error| format!("management request: {error}"))?;
        match outcome {
            frame_conv::outcome::RequestOutcome::Replied { reply, .. } => {
                report_reply(generation, &reply);
                match reply {
                    DevReply::NodeFailed { detail, .. } => Err(detail),
                    DevReply::Activated(_) | DevReply::Refused(_) | DevReply::RolledBack { .. } => {
                        Ok(())
                    }
                }
            }
            frame_conv::outcome::RequestOutcome::DeadlineElapsed { deadline } => {
                // The typed deadline outcome (R4): reported, survived.
                eprintln!(
                    "frame dev: generation {} push deadline elapsed after {deadline:?}; \
                     the node's status remains authoritative",
                    generation.0
                );
                Ok(())
            }
            frame_conv::outcome::RequestOutcome::ResponderFailed { failure, .. } => {
                // S4: the node-side participant died mid-request — the
                // exactly-once death delivery, surfaced as the node
                // failure it is.
                Err(format!(
                    "the node's management participant died mid-request: {failure:?}"
                ))
            }
        }
    }

    fn report_build_failure(&mut self, generation: BuildGeneration, error: &BuildError) {
        eprintln!(
            "frame dev: RELOAD FAILED — RUNNING LAST GOOD (generation {} build failed)\n{error}",
            generation.0
        );
    }

    fn resubscribe(&mut self) -> Result<(), String> {
        // Drop the desynchronized watch and establish a fresh one — an
        // event-driven recovery, never a rescan loop.
        self.watcher = None;
        let watch_tx = self.input_tx.clone();
        let mut watcher = notify::recommended_watcher(move |event| {
            let _ = watch_tx.send(LoopInput::Fs(event));
        })
        .map_err(|error| format!("re-creating the watcher: {error}"))?;
        watcher
            .watch(&self.watch_root, notify::RecursiveMode::Recursive)
            .map_err(|error| format!("re-watching {}: {error}", self.watch_root.display()))?;
        self.watcher = Some(watcher);
        println!("frame dev: watcher re-established after reported loss");
        Ok(())
    }
}

/// One candidate build: snapshot to the generation's staging, build, read
/// the BEAM files, store the candidate for the push.
fn build_one(
    component_dir: &std::path::Path,
    staging_root: &std::path::Path,
    gleam_name: &str,
    generation: BuildGeneration,
    candidates: &Mutex<std::collections::HashMap<u64, CandidateBytes>>,
) -> Result<(), BuildError> {
    let snapshot = snapshot_component(component_dir, staging_root, generation)?;
    let (component_beam, ffi_beam) = build_candidate(&snapshot, gleam_name)?;
    if let Ok(mut slot) = candidates.lock() {
        slot.insert(
            generation.0,
            CandidateBytes {
                build_generation: generation.0,
                content_digest: snapshot.digest,
                component_beam,
                ffi_beam,
            },
        );
    }
    Ok(())
}

/// Prints the node's exactly-one typed answer in constraint 7's language:
/// generation, stage, and state, with diagnostics attached — "still
/// serving" without "stale" is the forbidden report.
fn report_reply(generation: BuildGeneration, reply: &DevReply) {
    match reply {
        DevReply::Activated(report) => {
            println!(
                "frame dev: RUNNING CURRENT generation {} (incarnation {}, module generation {})",
                report.build_generation, report.component_incarnation, report.module_generation
            );
        }
        DevReply::Refused(refusal) => {
            eprintln!(
                "frame dev: generation {} refused: {refusal:?}; the node is unchanged",
                generation.0
            );
        }
        DevReply::RolledBack {
            failed,
            detail,
            serving,
        } => {
            eprintln!(
                "frame dev: RELOAD FAILED — RUNNING LAST GOOD generation {} \
                 (generation {} failed at {failed:?}: {detail})",
                serving.build_generation, generation.0
            );
        }
        DevReply::NodeFailed { failed, detail } => {
            eprintln!(
                "frame dev: NODE FAILED at {failed:?}: {detail} (generation {})",
                generation.0
            );
        }
    }
}