haematite-cli 0.21.8

use std::num::{NonZeroU64, NonZeroUsize};
use std::path::PathBuf;
use std::time::{Duration, Instant};

use serde::{Deserialize, Deserializer, de::Error};

pub const NAMES: [&str; 6] = [
    "query_time_ms",
    "query_memory_bytes",
    "query_outputs_per_record",
    "query_scan_records",
    "query_max_running",
    "query_worker",
];

/// Omission disables a group; an explicit null is an invalid setting.
#[derive(Clone, Copy, Debug, Default)]
pub enum Setting<T> {
    #[default]
    Absent,
    Present(T),
}

impl<'de, T: Deserialize<'de>> Deserialize<'de> for Setting<T> {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        T::deserialize(deserializer).map(Self::Present)
    }
}

impl<T> Setting<T> {
    pub const fn is_set(&self) -> bool {
        matches!(self, Self::Present(_))
    }

    fn into_value(self) -> Option<T> {
        match self {
            Self::Absent => None,
            Self::Present(value) => Some(value),
        }
    }
}

fn named<'de, T: Deserialize<'de>, D: Deserializer<'de>>(
    deserializer: D,
    name: &str,
) -> Result<Setting<T>, D::Error> {
    Setting::deserialize(deserializer)
        .map_err(|error| D::Error::custom(format!("query setting {name}: {error}")))
}

/// Deserialize the query time allowance.
///
/// # Errors
/// Returns a deserialization error naming `query_time_ms` for null or an invalid value.
pub fn time_ms<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
    named(deserializer, NAMES[0])
}
/// Deserialize the query memory allowance.
///
/// # Errors
/// Returns a deserialization error naming `query_memory_bytes` for null or an invalid value.
pub fn memory_bytes<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
    named(deserializer, NAMES[1])
}
/// Deserialize the per-record output allowance.
///
/// # Errors
/// Returns a deserialization error naming `query_outputs_per_record` for null or an invalid value.
pub fn outputs_per_record<'de, D: Deserializer<'de>>(
    deserializer: D,
) -> Result<Setting<u64>, D::Error> {
    named(deserializer, NAMES[2])
}
/// Deserialize the record scan allowance.
///
/// # Errors
/// Returns a deserialization error naming `query_scan_records` for null or an invalid value.
pub fn scan_records<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
    named(deserializer, NAMES[3])
}
/// Deserialize the concurrent-query limit.
///
/// # Errors
/// Returns a deserialization error naming `query_max_running` for null or an invalid value.
pub fn max_running<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<u64>, D::Error> {
    named(deserializer, NAMES[4])
}
/// Deserialize the query worker path.
///
/// # Errors
/// Returns a deserialization error naming `query_worker` for null or an invalid value.
pub fn worker<'de, D: Deserializer<'de>>(deserializer: D) -> Result<Setting<PathBuf>, D::Error> {
    named(deserializer, NAMES[5])
}

#[derive(Clone, Debug, Default, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Options {
    #[serde(default, deserialize_with = "time_ms")]
    pub query_time_ms: Setting<u64>,
    #[serde(default, deserialize_with = "memory_bytes")]
    pub query_memory_bytes: Setting<u64>,
    #[serde(default, deserialize_with = "outputs_per_record")]
    pub query_outputs_per_record: Setting<u64>,
    #[serde(default, deserialize_with = "scan_records")]
    pub query_scan_records: Setting<u64>,
    #[serde(default, deserialize_with = "max_running")]
    pub query_max_running: Setting<u64>,
    #[serde(default, deserialize_with = "worker")]
    pub query_worker: Setting<PathBuf>,
}

#[derive(Debug)]
pub struct Settings {
    pub time_ms: NonZeroU64,
    pub memory_bytes: NonZeroU64,
    pub outputs_per_record: NonZeroU64,
    pub scan_records: NonZeroU64,
    pub max_running: NonZeroUsize,
    pub worker: PathBuf,
}

/// The fewest records a page may read: a later page reads its pending record
/// again, and must read one more to move on.
pub const MIN_SCAN_RECORDS: u64 = 2;

fn required<T>(setting: Setting<T>, name: &str) -> Result<T, String> {
    setting
        .into_value()
        .ok_or_else(|| format!("missing query setting {name}"))
}

fn positive(setting: Setting<u64>, name: &str) -> Result<NonZeroU64, String> {
    NonZeroU64::new(required(setting, name)?)
        .ok_or_else(|| format!("query setting {name} must be positive"))
}

impl Options {
    /// Resolve the optional query settings as a complete validated group.
    ///
    /// # Errors
    /// Refuses missing or zero settings, an unrepresentable time or process count, a scan limit below two, or a worker path that is not absolute.
    pub fn resolve(self) -> Result<Option<Settings>, String> {
        let present = [
            self.query_time_ms.is_set(),
            self.query_memory_bytes.is_set(),
            self.query_outputs_per_record.is_set(),
            self.query_scan_records.is_set(),
            self.query_max_running.is_set(),
            self.query_worker.is_set(),
        ];
        if present.iter().all(|set| !set) {
            return Ok(None);
        }
        let missing: Vec<_> = NAMES
            .into_iter()
            .zip(present)
            .filter_map(|(name, set)| (!set).then_some(name))
            .collect();
        if !missing.is_empty() {
            return Err(format!("missing query settings: {}", missing.join(", ")));
        }
        let time_ms = positive(self.query_time_ms, NAMES[0])?;
        if Instant::now()
            .checked_add(Duration::from_millis(time_ms.get()))
            .is_none()
        {
            return Err(format!(
                "query setting {} exceeds the monotonic clock range",
                NAMES[0]
            ));
        }
        let memory_bytes = positive(self.query_memory_bytes, NAMES[1])?;
        let outputs_per_record = positive(self.query_outputs_per_record, NAMES[2])?;
        let scan_records = positive(self.query_scan_records, NAMES[3])?;
        if scan_records.get() < MIN_SCAN_RECORDS {
            return Err(format!(
                "query setting {} must be at least {MIN_SCAN_RECORDS}, so that every page \
                 moves on past the record it reads again",
                NAMES[3]
            ));
        }
        let count = positive(self.query_max_running, NAMES[4])?;
        let max_running = NonZeroUsize::new(
            usize::try_from(count.get())
                .map_err(|_| format!("query setting {} does not fit this platform", NAMES[4]))?,
        )
        .ok_or_else(|| format!("query setting {} must be positive", NAMES[4]))?;
        let worker = required(self.query_worker, NAMES[5])?;
        if !worker.is_absolute() {
            return Err(format!(
                "query setting {} must be an absolute path",
                NAMES[5]
            ));
        }
        Ok(Some(Settings {
            time_ms,
            memory_bytes,
            outputs_per_record,
            scan_records,
            max_running,
            worker,
        }))
    }
}