haematite-cli 0.21.7

use super::super::{
    changes, embed,
    live::LysLive,
    pass::Gate,
    placement,
    query::{Options, Setting, Settings},
    releaser,
    storage::{KEPT_VERSIONS_DEFAULT, KEPT_VERSIONS_MOST},
};
#[cfg(test)]
use super::endpoint_tests;
use crate::serve::service::Service;
use haematite::branch::KeptLimit;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde::Deserialize;
use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
use std::path::PathBuf;
use std::sync::Arc;
#[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::super::query::settings::time_ms")]
    pub query_time_ms: Setting<u64>,
    #[serde(
        default,
        deserialize_with = "super::super::query::settings::memory_bytes"
    )]
    pub query_memory_bytes: Setting<u64>,
    #[serde(
        default,
        deserialize_with = "super::super::query::settings::outputs_per_record"
    )]
    pub query_outputs_per_record: Setting<u64>,
    #[serde(
        default,
        deserialize_with = "super::super::query::settings::scan_records"
    )]
    pub query_scan_records: Setting<u64>,
    #[serde(
        default,
        deserialize_with = "super::super::query::settings::max_running"
    )]
    pub query_max_running: Setting<u64>,
    #[serde(default, deserialize_with = "super::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.
pub(super) 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.
pub(super) 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.
pub(super) 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::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::super::runtime::Endpoint, super::super::runtime::Refusal> {
        #[cfg(test)]
        if let Some((root, uid)) = endpoint_tests::input() {
            return super::super::runtime::resolve_input(
                &self.data_dir,
                self.name.as_deref(),
                self.socket.as_deref(),
                create,
                &root,
                uid,
            );
        }
        super::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::super::storage::AppendPruneLimits, String> {
        super::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).
pub(super) 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),
    }
}