liminal-server 0.21.1

use std::io;
use std::io::IsTerminal;
use std::path::PathBuf;

use clap::Parser;

/// Level directives applied when the operator sets no `RUST_LOG`.
///
/// The server's own crates run at `info` — that is where the lifecycle events an
/// operator watches for live (startup, shutdown requested, drain outcomes) — while
/// everything else is floored at `warn` so a dependency that later starts emitting
/// `tracing` events cannot flood the log by default, yet still cannot go silent
/// about a warning or an error. `liminal` is the library crate's lib name (its
/// package is published as `liminal-rs`), which is what `tracing` uses as the
/// event target.
const DEFAULT_FILTER: &str = "warn,liminal_server=info,liminal=info";

#[derive(Debug, Parser)]
struct Cli {
    #[arg(long)]
    config: PathBuf,
    #[command(subcommand)]
    command: Option<Command>,
}

#[derive(Debug, clap::Subcommand)]
enum Command {
    /// Rewrite every participant conversation log into the current row
    /// grammar, offline, then exit. Run it with the server stopped.
    MigrateParticipantLogs,
    /// Count what one conversation's checkpoint holds (DUR-024 R5): history
    /// nodes by kind with their bytes, the identities captured, the rows
    /// ahead of and behind the latest locator, and the batches a
    /// publication would commit at the store's bound. Writes nothing. Run
    /// it with the server stopped, or against a copy you name with --store.
    CheckpointCensus {
        /// The conversation to count.
        #[arg(long)]
        conversation: u64,
        /// A store directory to read instead of the configuration's
        /// persistence path (a copy the operator made; never one liminal
        /// makes).
        #[arg(long)]
        store: Option<PathBuf>,
    },
}

/// Installs the process-wide `tracing` subscriber that renders events to stderr.
///
/// Without this the server's 120 `tracing` events — connection failures, drain
/// timeouts, durable-flush failures among them — are dropped by the no-op global
/// dispatcher and the process runs from boot to shutdown printing nothing.
///
/// Output goes to **stderr** (the `fmt` builder's own default is stdout), leaving
/// stdout free for anything the operator pipes.
///
/// `RUST_LOG` is honoured when set; when it is unset — or set to something
/// `EnvFilter` cannot parse — [`DEFAULT_FILTER`] applies instead.
///
/// Installing a global subscriber can only ever succeed once per process, so this
/// uses `try_init` and treats an already-installed dispatcher as success: a
/// consumer that embeds this crate and installs its own subscriber first keeps it,
/// and neither case is worth aborting startup over.
fn init_tracing() {
    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(DEFAULT_FILTER));

    // Deliberately discarded: the sole error case is "a global subscriber is
    // already installed", which is the outcome this call wanted anyway.
    drop(
        tracing_subscriber::fmt()
            .with_env_filter(filter)
            // Colour only when a human is watching: non-tty stderr must stay
            // byte-clean because it is teed into committed gate/evidence logs.
            .with_ansi(io::stderr().is_terminal())
            .with_writer(io::stderr)
            .try_init(),
    );
}

fn main() -> anyhow::Result<()> {
    let cli = Cli::parse();

    init_tracing();

    match cli.command {
        None => liminal_server::run(cli.config.as_path())?,
        Some(Command::MigrateParticipantLogs) => {
            let migration = liminal_server::run_participant_log_migration(cli.config.as_path())?;
            for conversation in &migration.conversations {
                println!(
                    "{} {:?}",
                    conversation.conversation_id, conversation.outcome
                );
            }
        }
        Some(Command::CheckpointCensus {
            conversation,
            store,
        }) => {
            let census = liminal_server::run_checkpoint_census(
                cli.config.as_path(),
                conversation,
                store.as_deref(),
            )?;
            for line in census {
                println!("{line}");
            }
        }
    }

    Ok(())
}