haematite_cli/serve/

cli.rs

//! Explicit operator startup. Reading configuration never installs a service.

use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use haematite::branch::KeptLimit;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde::Deserialize;

use super::{
    changes, embed,
    live::LysLive,
    pass::Gate,
    placement,
    query::{Options, Setting, Settings, probe},
    releaser,
    service::Service,
    storage::{KEPT_VERSIONS_DEFAULT, KEPT_VERSIONS_MOST, Storage, UndatedGeneration},
    transport::{Limits, Listener},
};

/// Checks ownership of a resolved default name endpoint without changing it.
///
/// The expected uid is supplied by the caller's validated OS runtime context.
/// A successful probe lock is released before returning stopped; pids and
/// socket connection results do not participate in this observation.
///
/// # Errors
/// Returns `name_owner_probe_failed` with the anchor's cause when its location,
/// owner, mode, identity or OS lock cannot be checked or released.
pub fn name_is_live(default_endpoint: &std::path::Path, expected_uid: u32) -> Result<bool, String> {
    super::socket_owner::name_is_live(default_endpoint, expected_uid).map_err(|error| {
        format!(
            "name_owner_probe_failed: {}: {error}",
            default_endpoint.display()
        )
    })
}

const USAGE: &str = "usage: haematite serve --config <file.json> [--create] [--long-wait-interval <duration>]\n\
--create requires a new database directory; omit it when restarting.\n\
An explicit socket's parent directory must already exist with mode 0700.\n\
An omitted socket uses the private OS runtime namespace. No auto-launch is installed.";

const DEFAULT_LONG_WAIT_INTERVAL: Duration = Duration::from_secs(30);

struct ServeOptions {
    path: PathBuf,
    create: bool,
    long_wait_interval: Duration,
}

fn long_wait_help() -> String {
    format!(
        "--long-wait-interval <duration> is the interval between long-wait warnings, not a deadline. \
         default: {}s; use an integer followed by ns, us, ms, s, m or h.",
        DEFAULT_LONG_WAIT_INTERVAL.as_secs()
    )
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Config {
    pub mcp_http: Option<crate::mcp::http::Config>,
    pub data_dir: PathBuf,
    pub socket: Option<PathBuf>,
    pub name: Option<String>,
    pub branch_cache_bytes_per_shard: NonZeroUsize,
    pub max_frame: NonZeroU32,
    pub max_clients: NonZeroUsize,
    pub max_page: NonZeroU32,
    #[serde(default = "append_prune_keep_ceiling")]
    pub append_prune_keep_ceiling: u32,
    #[serde(default = "append_prune_removal_cap")]
    pub append_prune_removal_cap: u32,
    pub timeout_ms: NonZeroU64,
    pub create: Option<Creation>,
    /// `changes.list`'s limits, each defaulted when absent.
    #[serde(default = "changes_scan_records")]
    pub changes_scan_records: NonZeroU64,
    #[serde(default = "changes_node_budget")]
    pub changes_node_budget: NonZeroUsize,
    #[serde(default = "changes_max_page")]
    pub changes_max_page: NonZeroU32,
    #[serde(default, deserialize_with = "super::query::settings::time_ms")]
    pub query_time_ms: Setting<u64>,
    #[serde(default, deserialize_with = "super::query::settings::memory_bytes")]
    pub query_memory_bytes: Setting<u64>,
    #[serde(
        default,
        deserialize_with = "super::query::settings::outputs_per_record"
    )]
    pub query_outputs_per_record: Setting<u64>,
    #[serde(default, deserialize_with = "super::query::settings::scan_records")]
    pub query_scan_records: Setting<u64>,
    #[serde(default, deserialize_with = "super::query::settings::max_running")]
    pub query_max_running: Setting<u64>,
    #[serde(default, deserialize_with = "super::query::settings::worker")]
    pub query_worker: Setting<PathBuf>,
    #[serde(default)]
    pub embed_worker: Setting<PathBuf>,
    #[serde(default)]
    pub embed_model_dir: Setting<PathBuf>,
    #[serde(default)]
    pub embed_model_sha256: Setting<String>,
    #[serde(default)]
    pub embed_memory_bytes: Setting<u64>,
    #[serde(default)]
    pub embed_threads: Setting<u64>,
    #[serde(default)]
    pub embed_batch_items: Setting<u64>,
    #[serde(default)]
    pub embed_input_bytes: Setting<u64>,
    #[serde(default)]
    pub embed_tokens: Setting<u64>,
    #[serde(default)]
    pub embed_frame_bytes: Setting<u32>,
    #[serde(default)]
    pub embed_ready_ms: Setting<u64>,
    #[serde(default)]
    pub embed_call_ms: Setting<u64>,
    #[serde(default)]
    pub embed_waiting: Setting<u64>,
    #[serde(default)]
    pub embed_request_items: Setting<u64>,
    #[serde(default)]
    pub embed_page_records: Setting<u64>,
    #[serde(default)]
    pub embed_writer_wait_ms: Setting<u64>,
    /// EMB-E2: embed a text longer than the model reads whole, as windows.
    #[serde(default)]
    pub embed_windows: Setting<bool>,
    /// EMB-E2: the most windows one text is embedded as; 32 when absent.
    #[serde(default)]
    pub embed_max_windows: Setting<u64>,
    /// EMB-E2: the tokens consecutive windows share; 64 when absent.
    #[serde(default)]
    pub embed_window_overlap_tokens: Setting<u64>,
    /// EWR1: the wait before the second start of an embedding worker that
    /// died, doubled for each later one; one second when absent.
    #[serde(default)]
    pub embed_restart_ms: Setting<u64>,
    /// EWR1: the longest wait between two starts of the embedding worker;
    /// five minutes when absent.
    #[serde(default)]
    pub embed_restart_most_ms: Setting<u64>,
    /// SL1: whether text search asks its questions of a worker of its own,
    /// started on the first search, so a search never waits behind the
    /// embedding walk. On when absent. It holds its own copy of the model:
    /// with it the two workers may use twice `embed_memory_bytes`.
    #[serde(default)]
    pub embed_question_worker: Setting<bool>,
    /// Number of previous versions each branch record keeps,
    /// a whole number from 1 to [`KEPT_VERSIONS_MOST`].
    #[serde(default = "kept_versions", deserialize_with = "kept_versions_within")]
    pub kept_versions: usize,
    /// STOR-003 HR4: when the release thread runs and what it keeps.
    #[serde(default)]
    pub retention: releaser::Config,
    /// ACCESS-001 R2: decide every request through this Lys. Absent, the
    /// service is its owner's alone, reached only by its 0700 socket, and
    /// decides nothing.
    #[serde(default)]
    pub lys: Option<LysConfig>,
}

/// The Lys a served database decides every request through. Nothing here
/// grants anything: every right comes from a pass or a live answer.
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LysConfig {
    /// The issuer origin whose passes are trusted and whose live check
    /// deliberate routes ask.
    pub issuer: String,
    /// This database's id as Lys registers it, a `haematite.database`.
    pub database: String,
    /// The absolute path of a file holding haematite's Lys app credential,
    /// `lys-app.haematite.<secret>`. Given, every collection and branch is
    /// placed in Lys under this database, at start and after the write that
    /// makes each one commits, so a grant on the database reaches it
    /// (serve/placement.rs). Absent, nothing is placed, and a new collection
    /// or branch is reached only by a grant on itself until someone places it.
    #[serde(default)]
    pub placement_credential_file: Option<PathBuf>,
}

/// The Lys gate and placer a configuration asks for, both made before a
/// store opens or a socket binds. Neither waits on Lys: the gate reads its
/// keys on a thread of its own and refuses every decided act as
/// `lys_keys_pending` until they are read, so the service listens at once
/// and never serves undecided. A wrong credential file stops it by name.
fn lys_parts(lys: Option<&LysConfig>) -> Result<(Option<Gate>, Option<placement::Placer>), String> {
    let gate = lys.map(lys_gate).transpose()?;
    let placer = lys
        .and_then(|lys| {
            lys.placement_credential_file.as_deref().map(|file| {
                placement::Lys::new(&lys.issuer, &lys.database, file)
                    .and_then(placement::Placer::start)
            })
        })
        .transpose()?;
    Ok((gate, placer))
}

/// `service` deciding through `gate` and placing through `placer`, each
/// when given.
fn served_through_lys(
    mut service: Service,
    gate: Option<Gate>,
    placer: Option<placement::Placer>,
) -> Result<Service, String> {
    if let Some(gate) = gate {
        service = service.with_lys(gate);
    }
    if let Some(placer) = placer {
        service = service
            .with_placer(placer)
            .map_err(|error| error.to_string())?;
    }
    Ok(service)
}

/// The gate for `config`, its keys read from Lys in the background.
///
/// # Errors
///
/// Refuses an empty or placed database id, or an issuer that is not an HTTP
/// origin, by name.
fn lys_gate(config: &LysConfig) -> Result<Gate, String> {
    if config.database.is_empty()
        || !config
            .database
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-')
    {
        return Err(format!(
            "lys database id {:?} must be letters, digits, _ or - only",
            config.database
        ));
    }
    let mut live = LysLive::new(&config.issuer)
        .map_err(|error| format!("{}: the lys issuer {}", error.name(), config.issuer))?;
    if let Some(file) = config.placement_credential_file.as_deref() {
        live = live.asking_as(super::lys_schema::read_credential(file)?);
    }
    let live = Arc::new(live);
    let asking = Arc::clone(&live);
    Gate::fetching(
        move || asking.keys(),
        &config.issuer,
        &config.database,
        live,
    )
    .map_err(|error| format!("the thread reading Lys's keys could not start: {error}"))
}

const fn kept_versions() -> usize {
    KEPT_VERSIONS_DEFAULT
}

/// `kept_versions` as read, refused by name outside 1 to the most.
fn kept_versions_within<'de, D>(deserializer: D) -> Result<usize, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let value = u64::deserialize(deserializer)?;
    usize::try_from(value)
        .ok()
        .filter(|kept| (1..=KEPT_VERSIONS_MOST).contains(kept))
        .ok_or_else(|| {
            serde::de::Error::custom(format!(
                "kept_versions must be a whole number from 1 to {KEPT_VERSIONS_MOST}, not {value}"
            ))
        })
}

const fn append_prune_keep_ceiling() -> u32 {
    4_096
}

const fn append_prune_removal_cap() -> u32 {
    64
}

fn changes_scan_records() -> NonZeroU64 {
    changes::Settings::default().scan_records
}

fn changes_node_budget() -> NonZeroUsize {
    changes::Settings::default().node_budget
}

fn changes_max_page() -> NonZeroU32 {
    changes::Settings::default().max_page
}

impl Config {
    /// Resolves the endpoint when a caller needs it; deserialization remains pure.
    ///
    /// # Errors
    /// Returns the runtime resolver's named input, ownership and OS-query refusals.
    pub fn endpoint(
        &self,
        create: bool,
    ) -> Result<super::runtime::Endpoint, super::runtime::Refusal> {
        #[cfg(test)]
        if let Some((root, uid)) = endpoint_tests::input() {
            return super::runtime::resolve_input(
                &self.data_dir,
                self.name.as_deref(),
                self.socket.as_deref(),
                create,
                &root,
                uid,
            );
        }
        super::runtime::resolve(
            &self.data_dir,
            self.name.as_deref(),
            self.socket.as_deref(),
            create,
        )
    }

    /// The kept window every branch record keeps.
    ///
    /// # Errors
    /// Only for a `kept_versions` of zero, which loading refuses first.
    pub fn kept_limit(&self) -> Result<KeptLimit, String> {
        KeptLimit::new(self.kept_versions).map_err(|error| error.to_string())
    }

    /// # Errors
    /// Refuses either zero retention setting before opening a store.
    pub fn append_prune_limits(&self) -> Result<super::storage::AppendPruneLimits, String> {
        super::storage::AppendPruneLimits::new(
            self.append_prune_keep_ceiling,
            self.append_prune_removal_cap,
        )
        .map_err(|failure| failure.to_string())
    }

    /// `changes.list`'s limits; a page must be able to move past a change.
    ///
    /// # Errors
    /// Refuses `changes_scan_records` below two, which cannot advance a page.
    pub fn changes_settings(&self) -> Result<changes::Settings, String> {
        if self.changes_scan_records.get() < 2 {
            return Err("changes_scan_records must be at least 2".to_owned());
        }
        Ok(changes::Settings {
            scan_records: self.changes_scan_records,
            node_budget: self.changes_node_budget,
            max_page: self.changes_max_page,
        })
    }

    /// Resolve the optional query settings as one complete group.
    ///
    /// # Errors
    /// Refuses missing, zero, out-of-range or incompatible settings and a worker path that is not absolute.
    pub fn query_settings(&self) -> Result<Option<Settings>, String> {
        Options {
            query_time_ms: self.query_time_ms,
            query_memory_bytes: self.query_memory_bytes,
            query_outputs_per_record: self.query_outputs_per_record,
            query_scan_records: self.query_scan_records,
            query_max_running: self.query_max_running,
            query_worker: self.query_worker.clone(),
        }
        .resolve()
    }

    /// The required embedding settings, plus the optional writer wait.
    ///
    /// # Errors
    ///
    /// As [`embed::Options::resolve`].
    pub fn embed_settings(&self) -> Result<Option<embed::Settings>, String> {
        embed::Options {
            worker: self.embed_worker.clone(),
            model_dir: self.embed_model_dir.clone(),
            model_sha256: self.embed_model_sha256.clone(),
            memory_bytes: self.embed_memory_bytes,
            threads: self.embed_threads,
            batch_items: self.embed_batch_items,
            input_bytes: self.embed_input_bytes,
            tokens: self.embed_tokens,
            frame_bytes: self.embed_frame_bytes,
            ready_ms: self.embed_ready_ms,
            call_ms: self.embed_call_ms,
            waiting: self.embed_waiting,
            request_items: self.embed_request_items,
            page_records: self.embed_page_records,
            writer_wait_ms: self.embed_writer_wait_ms,
            windows: self.embed_windows,
            max_windows: self.embed_max_windows,
            overlap_tokens: self.embed_window_overlap_tokens,
        }
        .resolve()
    }
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Creation {
    pub shards: NonZeroUsize,
    pub cache_bytes_total: NonZeroUsize,
    pub executor_threads: NonZeroUsize,
}

/// The engine configuration of a store the service creates: the one place it
/// creates one. The store reclaims what nothing keeps, under the default
/// marking bound (STOR-003 H1a).
const fn creation_config(data_dir: PathBuf, creation: &Creation) -> DatabaseConfig {
    DatabaseConfig {
        data_dir,
        shard_count: creation.shards.get(),
        distributed: None,
        executor_threads: Some(creation.executor_threads.get()),
        node_cache_budget: Some(NodeCacheBudget::Bytes(creation.cache_bytes_total)),
        retention: haematite::Retention::reclaim(0),
    }
}

fn claim_name(
    config: &Config,
    endpoint: &super::runtime::Endpoint,
    create: bool,
) -> Result<std::fs::File, String> {
    let name_endpoint = if config.socket.is_some() {
        #[cfg(test)]
        let resolved = match endpoint_tests::input() {
            Some((root, uid)) => super::runtime::resolve_input(
                &config.data_dir,
                Some(&endpoint.name),
                None,
                create,
                &root,
                uid,
            ),
            None => super::runtime::resolve(&config.data_dir, Some(&endpoint.name), None, create),
        };
        #[cfg(not(test))]
        let resolved =
            super::runtime::resolve(&config.data_dir, Some(&endpoint.name), None, create);
        Some(resolved.map_err(|error| format!("bind service socket: {error}"))?)
    } else {
        None
    };
    let name_socket = name_endpoint
        .as_ref()
        .map_or(&endpoint.socket, |named| &named.socket);
    let namespace = name_socket
        .parent()
        .ok_or("bind service socket: name endpoint needs a parent")?;
    let name_uid = {
        use std::os::unix::fs::MetadataExt;
        std::fs::symlink_metadata(namespace)
            .map_err(|error| format!("bind service socket: {error}"))?
            .uid()
    };
    super::socket_owner::acquire_name(name_socket, name_uid)
        .map_err(|error| format!("bind service socket: {error}"))
}

/// The data directory and socket are absolute, the timeout fits the
/// monotonic clock, and the socket's parent is one this user may bind in;
/// the timeout is answered.
fn check_paths_and_timeout(config: &Config) -> Result<Duration, String> {
    if !config.data_dir.is_absolute()
        || config
            .socket
            .as_ref()
            .is_some_and(|socket| !socket.is_absolute())
    {
        return Err("data_dir and socket must be absolute paths".to_owned());
    }
    let timeout = Duration::from_millis(config.timeout_ms.get());
    if std::time::Instant::now().checked_add(timeout).is_none() {
        return Err("timeout cannot be represented by the monotonic clock".to_owned());
    }
    if let Some(socket) = config.socket.as_ref() {
        super::socket_owner::check_parent(socket)
            .map_err(|error| format!("bind service socket: {error}"))?;
    }
    Ok(timeout)
}

/// # Errors
///
/// Propagates any error from `parse`.
pub fn run(args: impl Iterator<Item = String>) -> Result<(), String> {
    let Some(options) = parse(args)? else {
        println!("{USAGE}\n{}", long_wait_help());
        return Ok(());
    };
    let config = read_config(&options.path)?;
    // Resolve the complete group before opening a store or binding a socket.
    let retention_limits = config.append_prune_limits()?;
    let query_settings = config.query_settings()?;
    let changes_settings = config.changes_settings()?;
    let release = config.retention.settings()?;
    // Read now, so a wrong key stops the service.
    let embedding = config.embed_settings()?;
    let (gate, placer) = lys_parts(config.lys.as_ref())?;
    let restart = embed::settings::Restart::of(
        config.embed_restart_ms,
        config.embed_restart_most_ms,
        embedding
            .as_ref()
            .map_or(Duration::ZERO, |settings| settings.ready),
    )?;
    let timeout = check_paths_and_timeout(&config)?;
    let endpoint = config
        .endpoint(options.create)
        .map_err(|error| error.to_string())?;
    let name_ownership = claim_name(&config, &endpoint, options.create)?;
    // A worker that cannot start does not stop the service: every other verb
    // is served, and the refusal is kept for `query.run` to answer by name.
    let query = query_settings.map(|settings| probe::prepare(settings, config.max_frame));
    if let Some(Err(refusal)) = query.as_ref().map(probe::Prepared::ready) {
        eprintln!("haem/1 query unavailable: {refusal}");
    }
    // Only the leaf-log mark is refused here; a missing or unreadable config
    // is left to the storage checks below, which answer as they always have.
    if let Err(error @ haematite::DatabaseError::LeafLogStore { .. }) =
        haematite::Database::check_ordinary_store(&config.data_dir)
    {
        return Err(error.to_string());
    }
    let branch_cache = NodeCacheBudget::Bytes(config.branch_cache_bytes_per_shard);
    let kept = config.kept_limit()?;
    let storage = if options.create {
        let creation = config
            .create
            .ok_or("--create needs a create section in the configuration")?;
        Storage::create_kept(
            creation_config(config.data_dir, &creation),
            branch_cache,
            kept,
            options.long_wait_interval,
        )
    } else {
        Storage::open_kept(
            &config.data_dir,
            branch_cache,
            kept,
            options.long_wait_interval,
        )
    }
    .map_err(|error| error.to_string())?;
    if options.create {
        endpoint
            .check_created()
            .map_err(|error| error.to_string())?;
    }
    let mut service =
        Service::with_query(storage, config.max_page, query).map_err(|error| error.to_string())?;
    service.changes = changes_settings;
    service.append_prune_limits = retention_limits;
    service = served_through_lys(service, gate, placer)?;
    service
        .start_release(release)
        .map_err(|error| error.to_string())?;
    // The worker starts on the embedding thread, off the serving path: until
    // it answers, text search is refused as `embed_worker_starting`.
    if let Some(settings) = embedding {
        let bounds = super::embedding::bounds(&settings);
        let writer_wait = settings.writer_wait;
        let apart = !matches!(config.embed_question_worker, Setting::Present(false));
        let questions = apart.then(|| super::embedding::embedder(settings.clone()));
        service
            .start_embedding_apart(
                super::embedding::embedder(settings),
                questions,
                bounds,
                writer_wait,
                restart,
            )
            .map_err(|error| error.to_string())?;
    }
    let service = Arc::new(service);
    service
        .activate_reclaim()
        .map_err(|error| format!("reclaim activation: {error}"))?;
    let served = serve_listener(
        &endpoint,
        &service,
        Limits {
            max_frame: config.max_frame,
            max_clients: config.max_clients,
            frame_timeout: timeout,
        },
    );
    drop(name_ownership);
    served
}

fn serve_listener(
    endpoint: &super::runtime::Endpoint,
    service: &Arc<Service>,
    limits: Limits,
) -> Result<(), String> {
    let mut listener = Listener::bind(&endpoint.socket)
        .map_err(|error| format!("bind service socket: {error}"))?;
    listener
        .publish(endpoint)
        .map_err(|error| error.to_string())?;
    eprintln!("haem/1 ready on {}", endpoint.socket.display());
    let served = listener
        .run(service, limits)
        .map_err(|error| format!("serve: {error}"));
    let cleanup = listener.finish().map_err(|error| error.to_string());
    // The name remains owned while registry and socket cleanup still run.
    drop(listener);
    match (served, cleanup) {
        (Ok(()), Ok(())) => Ok(()),
        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
        (Err(served), Err(cleanup)) => Err(format!("{served}; {cleanup}")),
    }
}

const RELEASE_USAGE: &str = "usage: haematite release date --config <file.json> --branch <name> \
--generation <created> --at <nanoseconds> [--long-wait-interval <duration>]\n\
Dates a generation discarded before its marker was written, as discarded at --at, so release \
takes its rows by keep.days from then. --generation is the record's creation time its rows are \
keyed by. Offline: the service must be stopped, as the store has one writer.";

struct DateOptions {
    branch: String,
    generation: u64,
    at: u64,
    long_wait_interval: Duration,
}

#[path = "cli/branch_rename.rs"]
mod branch_rename;
pub use branch_rename::{BRANCH_RENAME_USAGE, branch_rename};

/// DR-12: `haematite release date`, which dates a generation discarded before DR-1.
///
/// The operator's way out ([`Storage::date_undated_generation`], DR-11).
/// Offline: it opens the store as its one writer, so a running service holds
/// it off. Prints the marker's last version, `generation.sequence`.
///
/// # Errors
/// A usage error, the configuration's, the store's open refusal, or the
/// request's refusal by name (`branch_live`, `generation_marked`,
/// `generation_empty`).
pub fn release(mut args: impl Iterator<Item = String>) -> Result<(), String> {
    if args.next().as_deref() != Some("date") {
        return Err(RELEASE_USAGE.to_owned());
    }
    let mut args = args.peekable();
    if args.peek().is_some_and(|arg| arg == "--help") {
        if args.count() != 1 {
            return Err(RELEASE_USAGE.to_owned());
        }
        println!("{RELEASE_USAGE}\n{}", long_wait_help());
        return Ok(());
    }
    let (path, options) = parse_date(args)?;
    let config = read_config(&path)?;
    let branch_cache = NodeCacheBudget::Bytes(config.branch_cache_bytes_per_shard);
    let mut storage = Storage::open_kept(
        &config.data_dir,
        branch_cache,
        config.kept_limit()?,
        options.long_wait_interval,
    )
    .map_err(|error| error.to_string())?;
    let request = UndatedGeneration {
        branch: options.branch,
        generation: options.generation,
        at: options.at,
    };
    let last = storage
        .date_undated_generation(&request)
        .map_err(|failure| failure.to_string())?;
    println!("{}.{}", last.generation, last.sequence);
    Ok(())
}

/// The release request and its optional warning interval, each argument once;
/// the two record times are decimal nanoseconds.
fn parse_date(mut args: impl Iterator<Item = String>) -> Result<(PathBuf, DateOptions), String> {
    let (mut path, mut branch, mut generation, mut at) = (None, None, None, None);
    let mut long_wait_interval = None;
    while let Some(arg) = args.next() {
        let value = args.next().ok_or(RELEASE_USAGE)?;
        match arg.as_str() {
            "--config" if path.is_none() => path = Some(PathBuf::from(value)),
            "--branch" if branch.is_none() => branch = Some(value),
            "--generation" if generation.is_none() => generation = Some(decimal(&arg, &value)?),
            "--at" if at.is_none() => at = Some(decimal(&arg, &value)?),
            "--long-wait-interval" if long_wait_interval.is_none() => {
                long_wait_interval = Some(parse_long_wait_interval(&value)?);
            }
            _ => {
                return Err(format!(
                    "unknown or repeated argument {arg:?}\n{RELEASE_USAGE}"
                ));
            }
        }
    }
    let request = DateOptions {
        long_wait_interval: long_wait_interval.unwrap_or(DEFAULT_LONG_WAIT_INTERVAL),
        branch: branch.ok_or(RELEASE_USAGE)?,
        generation: generation.ok_or(RELEASE_USAGE)?,
        at: at.ok_or(RELEASE_USAGE)?,
    };
    Ok((path.ok_or(RELEASE_USAGE)?, request))
}

fn decimal(arg: &str, value: &str) -> Result<u64, String> {
    value
        .parse()
        .map_err(|error| format!("{arg} {value:?}: {error}\n{RELEASE_USAGE}"))
}

fn read_config(path: &std::path::Path) -> Result<Config, String> {
    let bytes = std::fs::read(path).map_err(|error| format!("read service config: {error}"))?;
    serde_json::from_slice(&bytes).map_err(|error| format!("service config: {error}"))
}

fn parse_long_wait_interval(value: &str) -> Result<Duration, String> {
    let invalid = || {
        format!(
            "invalid --long-wait-interval {value:?}: {}",
            long_wait_help()
        )
    };
    let end = value
        .find(|character: char| !character.is_ascii_digit())
        .ok_or_else(invalid)?;
    let (number, unit) = value.split_at(end);
    let amount: u64 = number.parse().map_err(|_| invalid())?;
    let interval = match unit {
        "ns" => Duration::from_nanos(amount),
        "us" => Duration::from_micros(amount),
        "ms" => Duration::from_millis(amount),
        "s" => Duration::from_secs(amount),
        "m" => Duration::from_secs(amount.checked_mul(60).ok_or_else(invalid)?),
        "h" => Duration::from_secs(amount.checked_mul(3_600).ok_or_else(invalid)?),
        _ => return Err(invalid()),
    };
    super::storage::runtime_options(interval).map_err(|error| error.to_string())?;
    Ok(interval)
}

fn parse(args: impl Iterator<Item = String>) -> Result<Option<ServeOptions>, String> {
    let mut args = args.peekable();
    if args.peek().is_some_and(|arg| arg == "--help") {
        return if args.count() == 1 {
            Ok(None)
        } else {
            Err(USAGE.to_owned())
        };
    }
    let mut path = None;
    let mut create = false;
    let mut long_wait_interval = None;
    while let Some(arg) = args.next() {
        match arg.as_str() {
            "--config" if path.is_none() => path = Some(PathBuf::from(args.next().ok_or(USAGE)?)),
            "--create" if !create => create = true,
            "--long-wait-interval" if long_wait_interval.is_none() => {
                long_wait_interval = Some(parse_long_wait_interval(&args.next().ok_or(USAGE)?)?);
            }
            _ => return Err(format!("unknown or repeated argument {arg:?}\n{USAGE}")),
        }
    }
    Ok(Some(ServeOptions {
        path: path.ok_or(USAGE)?,
        create,
        long_wait_interval: long_wait_interval.unwrap_or(DEFAULT_LONG_WAIT_INTERVAL),
    }))
}

#[cfg(test)]
#[path = "cli_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "cli_lys_tests.rs"]
mod lys_tests;

#[cfg(test)]
mod long_wait_tests {
    use std::error::Error;
    use std::time::Duration;

    use super::{DEFAULT_LONG_WAIT_INTERVAL, long_wait_help, parse, parse_date};

    #[test]
    fn serve_default_is_the_named_thirty_second_interval() -> Result<(), Box<dyn Error>> {
        let options = parse(["--config", "/config.json"].map(str::to_owned).into_iter())?
            .ok_or("serve options")?;
        assert_eq!(DEFAULT_LONG_WAIT_INTERVAL, Duration::from_secs(30));
        assert_eq!(options.long_wait_interval, DEFAULT_LONG_WAIT_INTERVAL);
        assert!(long_wait_help().contains("default: 30s"));
        Ok(())
    }

    #[test]
    fn serve_explicit_interval_is_carried_into_the_engine_options() -> Result<(), Box<dyn Error>> {
        let options = parse(
            ["--config", "/config.json", "--long-wait-interval", "125ms"]
                .map(str::to_owned)
                .into_iter(),
        )?
        .ok_or("serve options")?;
        let runtime = super::super::storage::runtime_options(options.long_wait_interval)?;
        assert_eq!(runtime.long_wait_interval, Duration::from_millis(125));
        Ok(())
    }

    #[test]
    fn release_date_keeps_request_fields_and_carries_the_interval() -> Result<(), Box<dyn Error>> {
        let arguments = [
            "--config",
            "/config.json",
            "--branch",
            "work",
            "--generation",
            "7",
            "--at",
            "9",
        ];
        let (path, options) = parse_date(arguments.map(str::to_owned).into_iter())?;
        assert_eq!(path, std::path::PathBuf::from("/config.json"));
        assert_eq!(
            (options.branch.as_str(), options.generation, options.at),
            ("work", 7, 9)
        );
        assert_eq!(options.long_wait_interval, DEFAULT_LONG_WAIT_INTERVAL);
        let explicit = arguments
            .into_iter()
            .chain(["--long-wait-interval", "2m"])
            .map(str::to_owned);
        let (_, options) = parse_date(explicit)?;
        let runtime = super::super::storage::runtime_options(options.long_wait_interval)?;
        assert_eq!(runtime.long_wait_interval, Duration::from_secs(120));
        Ok(())
    }

    #[test]
    fn invalid_intervals_are_refused_at_the_cli_boundary() {
        for value in [
            "0s",
            "-1s",
            "1",
            "1.5s",
            "1fortnight",
            "18446744073709551615h",
        ] {
            let parsed = parse(
                ["--config", "/config.json", "--long-wait-interval", value]
                    .map(str::to_owned)
                    .into_iter(),
            );
            assert!(parsed.is_err(), "interval {value}");
        }
        let refused = parse(
            ["--config", "/config.json", "--long-wait-interval", "0s"]
                .map(str::to_owned)
                .into_iter(),
        );
        assert!(
            matches!(refused, Err(message) if message.contains("long-wait interval 0ns is refused"))
        );
        let repeated = parse(
            [
                "--config",
                "/config.json",
                "--long-wait-interval",
                "1s",
                "--long-wait-interval",
                "2s",
            ]
            .map(str::to_owned)
            .into_iter(),
        );
        assert!(repeated.is_err());
    }

    #[test]
    fn help_does_not_require_a_configuration_or_open_a_store() -> Result<(), String> {
        super::run(["--help".to_owned()].into_iter())?;
        super::release(["date", "--help"].map(str::to_owned).into_iter())?;
        Ok(())
    }
}

#[cfg(test)]
#[path = "cli_endpoint_tests.rs"]
pub(crate) mod endpoint_tests;