chronix 0.7.0

//! Accessor methods (getters / setters) for [`Chronix`].

use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::error::{DbError, Result};

use metrics::gauge;

use chronix_core::{ChronixConfig, MeasurementSchema, SchemaRegistry};
#[cfg(feature = "streaming")]
use chronix_streaming::cdc::{EventBus, FilteredSubscription, SubscriptionFilter};

use super::{Chronix, DatabaseStatistics};

/// A read-only snapshot of one active segment.
///
/// Returned by [`Chronix::segments`] and [`Chronix::segments_of`]. It is an
/// owned copy taken under the catalog lock and handed back after it is
/// released, so holding one cannot block a flush, a compaction or a
/// retention pass — and cannot deadlock against them.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct SegmentInfo {
    /// The measurement whose rows this segment holds.
    pub measurement: String,
    /// Absolute path to the `.csx` file.
    pub path: std::path::PathBuf,
    /// Rows in the segment, before deduplication and tombstones.
    ///
    /// A query may return fewer: a later segment can overwrite a point
    /// (last-write-wins) and a tombstone can mask one. This is what the
    /// file holds, not what a `SELECT count(*)` answers.
    pub rows: u64,
    /// Distinct series in the segment.
    pub series: u32,
    /// Oldest timestamp held, inclusive (nanoseconds since the epoch).
    pub min_time: i64,
    /// Newest timestamp held, inclusive (nanoseconds since the epoch).
    pub max_time: i64,
    /// Size of the file on disk, in bytes.
    pub bytes: u64,
}

impl SegmentInfo {
    fn from_entry(e: &chronix_engine::index::SegmentCatalogEntry, segments_dir: &Path) -> Self {
        Self {
            measurement: e.measurement.clone(),
            path: e.file.resolve(segments_dir),
            rows: e.row_count,
            series: e.series_count,
            min_time: e.min_timestamp,
            max_time: e.max_timestamp,
            bytes: e.byte_size,
        }
    }
}

impl Chronix {
    #[cfg(feature = "sql")]
    /// Register a custom scalar UDF that will be available in SQL queries.
    pub fn register_udf(&self, udf: Arc<datafusion::logical_expr::ScalarUDF>) {
        *self.sql_ctx.write() = None;
        self.custom_udfs.write().push(udf);
    }

    #[cfg(feature = "sql")]
    /// Register a custom aggregate UDF that will be available in SQL queries.
    pub fn register_udaf(&self, udaf: Arc<datafusion::logical_expr::AggregateUDF>) {
        *self.sql_ctx.write() = None;
        self.custom_udafs.write().push(udaf);
    }

    #[cfg(feature = "sql")]
    /// Run a SQL query and collect the result.
    ///
    /// The whole SQL surface — DataFusion with Chronix's measurements as
    /// tables, `time_bucket`, `rate`, `forecast` and the rest of the
    /// analytics functions — from one call:
    ///
    /// ```no_run
    /// # use chronix::Chronix;
    /// # let db = Chronix::open_small("/tmp/db").unwrap();
    /// let batches = db.sql(
    ///     "SELECT time_bucket('5m', _time) AS t, avg(usage) FROM cpu GROUP BY t ORDER BY t",
    /// )?;
    /// # Ok::<(), chronix::DbError>(())
    /// ```
    ///
    /// Queries are read-only, verified before they are planned: `SET`,
    /// DDL and `COPY` are rejected, exactly as on the server. Writes go
    /// through [`insert_batch`](Self::insert_batch).
    ///
    /// This call **blocks**: on a private runtime from synchronous code, or
    /// by stepping the current worker aside on a multi-thread tokio runtime.
    /// On a current-thread runtime it is an error rather than a deadlock —
    /// use [`sql_async`](Self::sql_async) there.
    ///
    /// # Errors
    ///
    /// Returns [`DbError::Sql`] for a query that does not parse, plan or
    /// execute, and [`DbError::Internal`] when called from inside a
    /// current-thread async runtime.
    pub fn sql(&self, query: &str) -> Result<Vec<arrow::record_batch::RecordBatch>> {
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            // On a multi-thread runtime the worker can step aside for the
            // duration; on a current-thread runtime blocking would deadlock
            // the only worker, so that is an error rather than a hang.
            if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
                return tokio::task::block_in_place(|| handle.block_on(self.sql_async(query)));
            }
            return Err(DbError::Internal(
                "Chronix::sql() blocks and was called inside a current-thread async runtime; \
                 use sql_async()"
                    .into(),
            ));
        }
        let runtime = self.sql_runtime.get_or_init(|| {
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("a current-thread tokio runtime can always be built")
        });
        runtime.block_on(self.sql_async(query))
    }

    #[cfg(feature = "sql")]
    /// [`sql`](Self::sql) for async callers.
    ///
    /// # Errors
    ///
    /// As [`sql`](Self::sql).
    pub async fn sql_async(&self, query: &str) -> Result<Vec<arrow::record_batch::RecordBatch>> {
        let ctx = self.session_context();
        let df = crate::sql::sql_read_only(&ctx, query).await?;
        Ok(df.collect().await?)
    }

    #[cfg(feature = "sql")]
    /// The DataFusion [`SessionContext`](datafusion::execution::context::SessionContext)
    /// behind [`sql`](Self::sql): every measurement as a table, all
    /// analytics functions registered. For callers that want DataFusion's
    /// own API — `DataFrame`, `EXPLAIN`, streaming execution.
    ///
    /// Built on first use and shared; rebuilt after
    /// [`register_udf`](Self::register_udf) / [`register_udaf`](Self::register_udaf).
    /// Multi-tenant servers scope their own contexts with
    /// [`create_namespaced_session_context`](crate::sql::create_namespaced_session_context).
    pub fn session_context(&self) -> datafusion::execution::context::SessionContext {
        if let Some(ctx) = self.sql_ctx.read().as_ref() {
            return ctx.clone();
        }
        let mut slot = self.sql_ctx.write();
        if let Some(ctx) = slot.as_ref() {
            return ctx.clone();
        }
        let ctx = crate::sql::create_session_context(Arc::new(self.clone()));
        *slot = Some(ctx.clone());
        ctx
    }

    /// Evaluate a PromQL instant query at `at_ns`.
    ///
    /// Measurements are metrics and tags are labels, exactly as
    /// `chronixd` serves them to Grafana; the evaluator tracks Prometheus
    /// 3.x semantics.
    ///
    /// ```no_run
    /// # use chronix::Chronix;
    /// # let db = Chronix::open_small("/tmp/db").unwrap();
    /// let now = 1_700_000_000_000_000_000;
    /// let value = db.promql(r#"rate(requests{host="a"}[5m])"#, now)?;
    /// # Ok::<(), chronix::DbError>(())
    /// ```
    ///
    /// # Errors
    ///
    /// Returns [`DbError::PromQl`] for a query that does not parse or
    /// evaluate.
    pub fn promql(&self, query: &str, at_ns: i64) -> Result<crate::promql::ast::PromQLValue> {
        let expr = crate::promql::parse(query).map_err(|e| DbError::PromQl(e.to_string()))?;
        let params = crate::promql::eval::QueryParams {
            time: at_ns,
            ..Default::default()
        };
        crate::promql::PromQLEvaluator::new(Arc::new(self.clone()))
            .instant_query(&expr, &params)
            .map_err(|e| DbError::PromQl(e.0))
    }

    /// Evaluate a PromQL range query over `[start_ns, end_ns]` at `step_ns`.
    ///
    /// The result is one series per label set, one sample per step; the
    /// window is read once, not once per step.
    ///
    /// # Errors
    ///
    /// As [`promql`](Self::promql); a non-positive step is an error.
    pub fn promql_range(
        &self,
        query: &str,
        start_ns: i64,
        end_ns: i64,
        step_ns: i64,
    ) -> Result<crate::promql::ast::PromQLValue> {
        let expr = crate::promql::parse(query).map_err(|e| DbError::PromQl(e.to_string()))?;
        let params = crate::promql::eval::QueryParams {
            time: end_ns,
            start: Some(start_ns),
            end: Some(end_ns),
            step: Some(step_ns),
            ..Default::default()
        };
        crate::promql::PromQLEvaluator::new(Arc::new(self.clone()))
            .range_query(&expr, &params)
            .map_err(|e| DbError::PromQl(e.0))
    }

    #[cfg(feature = "sql")]
    /// Return a snapshot of all runtime-registered scalar UDFs.
    pub fn custom_udfs(&self) -> Vec<Arc<datafusion::logical_expr::ScalarUDF>> {
        self.custom_udfs.read().clone()
    }

    #[cfg(feature = "sql")]
    /// Return a snapshot of all runtime-registered aggregate UDFs.
    pub fn custom_udafs(&self) -> Vec<Arc<datafusion::logical_expr::AggregateUDF>> {
        self.custom_udafs.read().clone()
    }

    /// Return the schema for a measurement, if it exists.
    ///
    /// `None` for a measurement pending a soft-delete, exactly as for one
    /// that never existed: a "drop" that leaves the data fully readable for
    /// its whole grace period is not a drop. The segments and the schema
    /// registry entry are untouched, so a restore before the deadline is
    /// instant and lossless.
    #[must_use]
    pub fn schema(&self, measurement: &str) -> Option<Arc<MeasurementSchema>> {
        if self.catalog.read().is_measurement_pending_drop(measurement) {
            return None;
        }
        self.schema.lookup(measurement)
    }

    /// Return a snapshot of the schema registry.
    #[must_use]
    pub fn schema_registry(&self) -> &SchemaRegistry {
        &self.schema
    }

    /// Return the database configuration.
    #[must_use]
    pub fn config(&self) -> &ChronixConfig {
        &self.config
    }

    /// Return a snapshot of current database statistics.
    ///
    /// Provides on-demand database metrics including series
    /// cardinality, segment counts, memory usage, and WAL position.
    /// Also emits the values as gauge metrics via the `metrics` crate.
    ///
    /// `series_count` is the **exact** number of live series — the same
    /// number the write-path cardinality limit is checked against, so the
    /// gauge and the admission decision cannot disagree.
    #[must_use]
    pub fn statistics(&self) -> DatabaseStatistics {
        let series_count = self.known_series.len();
        let segment_count = {
            let catalog = self.catalog.read();
            catalog.segment_count()
        };
        let shard_count = self.catalog.read().shard_count();
        let memtable_memory_bytes = self.shards.total_memory();
        let interner_memory_bytes = self.shards.total_interner_memory();
        let wal_buffer_bytes = self.wal.memory_bytes();
        let catalog_memory_bytes = self.catalog.read().memory_bytes();
        let measurement_count = self.schema.measurement_count();
        let wal_sequence = self.wal.current_sequence();
        let tombstone_count = {
            let ts = self.tombstones.read();
            ts.len()
        };
        let leased_segments = self.segment_leases.len();

        // Emit gauge metrics so Prometheus/OTLP scrapers pick them up.
        gauge!("chronix_series_count").set(series_count as f64);
        gauge!("chronix_segment_count").set(segment_count as f64);
        gauge!("chronix_shard_count").set(shard_count as f64);
        gauge!("chronix_memtable_memory_bytes").set(memtable_memory_bytes as f64);
        gauge!("chronix_interner_memory_bytes").set(interner_memory_bytes as f64);
        gauge!("chronix_wal_buffer_bytes").set(wal_buffer_bytes as f64);
        gauge!("chronix_catalog_memory_bytes").set(catalog_memory_bytes as f64);
        gauge!("chronix_measurement_count").set(measurement_count as f64);
        gauge!("chronix_wal_sequence").set(wal_sequence as f64);
        gauge!("chronix_tombstone_count").set(tombstone_count as f64);
        gauge!("chronix_leased_segments").set(leased_segments as f64);
        gauge!("chronix_storage_disk_usage_bytes").set(self.disk_usage_bytes() as f64);

        DatabaseStatistics {
            series_count,
            segment_count,
            shard_count,
            memtable_memory_bytes,
            interner_memory_bytes,
            wal_buffer_bytes,
            catalog_memory_bytes,
            measurement_count,
            wal_sequence,
            tombstone_count,
            leased_segments,
        }
    }

    /// Bytes the database occupies on disk, cached for a minute.
    ///
    /// Disk usage is the storage number an operator actually alerts on, and
    /// nothing exported it. It needs a directory walk, so the result is
    /// cached: `statistics()` runs on every metrics scrape, and a walk per
    /// scrape would make the exporter the most expensive thing in the
    /// process once a deployment holds many segments.
    #[must_use]
    pub fn disk_usage_bytes(&self) -> u64 {
        const TTL: Duration = Duration::from_secs(60);

        {
            let cached = self.disk_usage.read();
            if let Some((measured_at, bytes)) = *cached
                && measured_at.elapsed() < TTL
            {
                return bytes;
            }
        }

        let bytes = directory_size(self.data_dir());
        *self.disk_usage.write() = Some((Instant::now(), bytes));
        bytes
    }

    /// Measurement names this namespace holds series for.
    ///
    /// `None` means unscoped — every measurement, which is what a
    /// single-tenant deployment wants and what the embedded API always
    /// wants.
    ///
    /// The SQL catalog answers from this rather than from the process-wide
    /// schema registry, which listed **every** tenant's measurement names:
    /// `SELECT * FROM another_tenants_measurement` returned zero rows where a
    /// name that does not exist errors, so a tenant could enumerate the
    /// others by probing.
    /// A measurement pending a whole-measurement drop is filtered out of
    /// **both** branches. It used to be filtered out of the unscoped one
    /// only, so a drop with a grace period configured was invisible on a
    /// single-tenant server and fully listed on a tenanted one — one
    /// question, two answers, in one function.
    #[must_use]
    pub fn measurement_names_in(&self, namespace: Option<&str>) -> Vec<String> {
        let pending = self.catalog.read().pending_measurement_drops().clone();
        let Some(namespace) = namespace else {
            return self
                .schema
                .measurement_names()
                .into_iter()
                .filter(|m| !pending.contains_key(m))
                .collect();
        };
        let mut names: Vec<String> = self
            .namespace_measurements
            .get(namespace)
            .map(|set| {
                set.iter()
                    .filter(|m| !pending.contains_key(m.as_str()))
                    .map(|m| m.clone())
                    .collect()
            })
            .unwrap_or_default();
        names.sort_unstable();
        names
    }

    /// Whether `namespace` holds any series of `measurement`.
    ///
    /// `None` means unscoped, and then only the measurement's existence
    /// matters. A measurement pending a drop exists for nobody, on either
    /// branch — the scoped one used to say yes, so under tenancy a dropped
    /// table still planned and answered zero rows where a name that does not
    /// exist is a planning error.
    #[must_use]
    pub fn has_measurement_in(&self, namespace: Option<&str>, measurement: &str) -> bool {
        if self.is_measurement_pending_drop(measurement) {
            return false;
        }
        match namespace {
            None => self.schema(measurement).is_some(),
            Some(namespace) => self
                .namespace_measurements
                .get(namespace)
                .is_some_and(|set| set.contains(measurement)),
        }
    }

    /// Return the path to the data directory.
    #[must_use]
    pub fn data_dir(&self) -> &Path {
        &self.config.data_dir
    }

    // ── Segment inspection ──────────────────────────────────────────
    //
    // These replace a `catalog()` accessor that handed callers the
    // `RwLock` the engine's own lock hierarchy is built on. Three things
    // were wrong with that and only the third is obvious. The lock's type
    // lives in a `pub(crate)` module, so a caller could call the method and
    // could not name what it returned. The hierarchy is documented as "an
    // invariant of this crate's own implementation, not a contract with
    // callers" — and was then enforced on callers by a debug-build panic
    // naming levels they had no way to read. And in release the assertions
    // compile away, so the same two reads deadlock against the maintenance
    // thread instead.
    //
    // Every caller in this repository wanted one of three facts. They get
    // exactly those, as owned values, with the lock taken and released
    // inside.

    /// How many active segments the database holds.
    #[must_use]
    pub fn segment_count(&self) -> usize {
        self.catalog.read().segment_count()
    }

    /// A snapshot of every active segment.
    ///
    /// Ordered by measurement, then by start time, so two calls on an
    /// unchanged database compare equal. The paths are absolute and
    /// resolved against this database's directory.
    #[must_use]
    pub fn segments(&self) -> Vec<SegmentInfo> {
        let dir = self.config.data_dir.join("segments");
        let mut out: Vec<SegmentInfo> = self
            .catalog
            .read()
            .all_segments()
            .iter()
            .map(|e| SegmentInfo::from_entry(e, &dir))
            .collect();
        out.sort_by(|a, b| {
            a.measurement
                .cmp(&b.measurement)
                .then(a.min_time.cmp(&b.min_time))
                .then(a.path.cmp(&b.path))
        });
        out
    }

    /// A snapshot of the active segments holding data for `measurement`.
    #[must_use]
    pub fn segments_of(&self, measurement: &str) -> Vec<SegmentInfo> {
        let dir = self.config.data_dir.join("segments");
        let mut out: Vec<SegmentInfo> = self
            .catalog
            .read()
            .active_segments_for_measurement(measurement)
            .iter()
            .map(|e| SegmentInfo::from_entry(e, &dir))
            .collect();
        out.sort_by(|a, b| a.min_time.cmp(&b.min_time).then(a.path.cmp(&b.path)));
        out
    }

    /// Refuse to write an encrypted column's plaintext outside the segment
    /// format.
    ///
    /// Three paths take a decrypted `RecordBatch` and write it elsewhere —
    /// the Parquet export, the cold archive, and a rollup's materialised
    /// aggregate — where the segment's protection does not reach. Each calls
    /// this first.
    ///
    /// # Errors
    ///
    /// [`DbError::InvalidRequest`] naming the first encrypted column found.
    #[cfg(feature = "field-encryption")]
    pub(crate) fn refuse_encrypted_columns<'a>(
        &self,
        columns: impl IntoIterator<Item = &'a str>,
        operation: &str,
    ) -> Result<()> {
        if self.field_encryption.is_none() {
            return Ok(());
        }
        for column in columns {
            if self.config.field_encryption.covers(column) {
                return Err(DbError::InvalidRequest(format!(
                    "{operation} would write column '{column}' in plaintext, and it is \
                     declared encrypted. An encrypted column does not leave the segment \
                     format: project it away, or drop the declaration if this data does not \
                     need it"
                )));
            }
        }
        Ok(())
    }

    /// The columns of `measurement`, or an empty list if it is unknown.
    #[cfg(feature = "field-encryption")]
    pub(crate) fn column_names_of(&self, measurement: &str) -> Vec<String> {
        self.schema(measurement)
            .map(|s| s.columns().iter().map(|c| c.name.clone()).collect())
            .unwrap_or_default()
    }

    /// Open a segment for reading, with this database's decryption keys.
    ///
    /// **The only way a read path opens a segment.** A `SegmentReader` with
    /// no key provider fails loudly on an encrypted column rather than
    /// returning ciphertext — which is the right default and a bad thing to
    /// reach by forgetting, because whether it fires depends on which of
    /// seven call sites a query happened to take. One function, so a reader
    /// added tomorrow cannot forget.
    pub(crate) fn open_segment(
        &self,
        path: impl AsRef<std::path::Path>,
    ) -> std::result::Result<
        chronix_engine::segment::SegmentReader,
        chronix_engine::segment::SegmentError,
    > {
        #[allow(unused_mut)]
        let mut reader = chronix_engine::segment::SegmentReader::open(path)?;
        #[cfg(feature = "field-encryption")]
        if let Some(enc) = &self.field_encryption {
            reader.set_key_provider(std::sync::Arc::clone(&enc.reader));
        }
        Ok(reader)
    }

    /// Where this database's segment files live.
    ///
    /// Every [`SegmentFile`](chronix_core::SegmentFile) in the catalog is
    /// relative to this, so a reader that wants to open one resolves it here
    /// — which is the whole reason the catalog no longer stores an absolute
    /// path.
    #[must_use]
    pub(crate) fn segments_dir(&self) -> std::path::PathBuf {
        super::segments_dir_of(&self.config.data_dir)
    }

    /// Number of WAL records `open()` replayed into the memtable.
    ///
    /// Zero after a graceful `close()`: everything acknowledged was already
    /// in a segment, and the catalog records the WAL floor so replay has
    /// nothing to do.
    #[must_use]
    pub fn wal_replayed_records(&self) -> usize {
        self.replayed_records
    }

    /// Return the current WAL sequence number.
    #[must_use]
    pub fn wal_sequence(&self) -> u64 {
        self.wal.current_sequence()
    }

    /// Every distinct tag key this database holds.
    ///
    /// **Both halves.** The inverted index is built at flush, so it knows
    /// only the segments on disk; a series written a second ago is in no
    /// segment and therefore in no index. Answering from the index alone —
    /// which this did — meant a freshly started gateway reported *no labels
    /// at all* until its first flush, and a tag that only ever appears in
    /// recent data was permanently invisible. The server's `/labels` never
    /// had the bug because it scans; this is the embedded answer to the same
    /// question, and the two must agree.
    ///
    /// Sorted, and deduplicated across the two sources.
    #[must_use]
    pub fn tag_keys(&self) -> Vec<String> {
        let mut keys: std::collections::BTreeSet<String> =
            self.tag_index.all_keys().into_iter().collect();
        for (key, _) in self.shards.live_tag_pairs() {
            keys.insert(key);
        }
        keys.into_iter().collect()
    }

    /// Every distinct value of `key` this database holds.
    ///
    /// Segments and memtables both, for the reason in
    /// [`tag_keys`](Self::tag_keys). Sorted.
    #[must_use]
    pub fn tag_values(&self, key: &str) -> Vec<String> {
        let mut values: std::collections::BTreeSet<String> =
            self.tag_index.values_for_key(key).into_iter().collect();
        for (k, value) in self.shards.live_tag_pairs() {
            if k == key {
                values.insert(value);
            }
        }
        values.into_iter().collect()
    }

    /// Access the underlying CDC event bus.
    ///
    /// Use this to create raw [`chronix_streaming::cdc::Subscription`]s or inspect
    /// bus statistics (published/lagged counts).
    #[cfg(feature = "streaming")]
    pub fn event_bus(&self) -> &EventBus {
        &self.cdc_bus
    }

    /// Create a filtered CDC subscription.
    ///
    /// Returns a [`FilteredSubscription`] that only delivers events matching
    /// the provided filter criteria (measurement, tags, event type).
    ///
    /// # Example
    ///
    /// ```no_run
    /// use chronix_streaming::cdc::SubscriptionFilter;
    /// use chronix::Chronix;
    /// # let db: Chronix = todo!();
    ///
    /// let filter = SubscriptionFilter::all()
    ///     .measurement("cpu")
    ///     .event_type("point_written");
    /// let sub = db.subscribe(filter);
    /// ```
    #[cfg(feature = "streaming")]
    pub fn subscribe(&self, filter: SubscriptionFilter) -> FilteredSubscription {
        FilteredSubscription::new(&self.cdc_bus, filter)
    }
}

/// Total size of every regular file under `root`, following no symlinks.
///
/// Errors are skipped rather than propagated: a file that vanished
/// mid-walk (a compaction finishing) is normal, and a disk-usage figure
/// that is a little stale is more useful than none.
fn directory_size(root: &Path) -> u64 {
    let mut total = 0u64;
    let mut stack = vec![root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let Ok(entries) = std::fs::read_dir(&dir) else {
            continue;
        };
        for entry in entries.flatten() {
            let Ok(meta) = entry.metadata() else { continue };
            if meta.is_dir() {
                stack.push(entry.path());
            } else if meta.is_file() {
                total += meta.len();
            }
        }
    }
    total
}

/// Build the namespace → measurements index from a set of canonical series
/// keys.
///
/// A canonical key is `measurement\0k1\0v1\0k2\0v2…` with the tags sorted,
/// so both halves are already here — the index only makes the lookup cheap
/// enough for the SQL planner to ask on every table reference.
pub(super) fn index_by_namespace(
    known_series: &dashmap::DashSet<String>,
) -> dashmap::DashMap<String, dashmap::DashSet<String>> {
    let index = dashmap::DashMap::new();
    for key in known_series.iter() {
        record_series(&index, key.as_str());
    }
    index
}

/// Record one canonical series key in the index.
pub(super) fn record_series(
    index: &dashmap::DashMap<String, dashmap::DashSet<String>>,
    canonical: &str,
) {
    let Some((measurement, namespace)) = split_canonical(canonical) else {
        return;
    };
    index
        .entry(namespace.to_string())
        .or_default()
        .insert(measurement.to_string());
}

/// The measurement and namespace of a canonical series key.
///
/// A canonical key is `measurement` then, for each tag in sorted key order,
/// [`TAG_SEPARATOR`], the key, [`KV_SEPARATOR`], the value. Both separators
/// are control characters that [`SeriesKey::validate_name`] rejects in user
/// data, so splitting on them is unambiguous.
///
/// The namespace is `""` when the series carries no namespace tag, which is
/// every series on a single-tenant deployment.
///
/// [`TAG_SEPARATOR`]: chronix_core::TAG_SEPARATOR
/// [`KV_SEPARATOR`]: chronix_core::KV_SEPARATOR
/// [`SeriesKey::validate_name`]: chronix_core::SeriesKey::validate_name
pub(super) fn split_canonical(canonical: &str) -> Option<(&str, &str)> {
    let mut parts = canonical.split(chronix_core::TAG_SEPARATOR);
    let measurement = parts.next()?;
    let namespace = parts
        .find_map(|tag| {
            let (key, value) = tag.split_once(chronix_core::KV_SEPARATOR)?;
            (key == chronix_core::NAMESPACE_TAG).then_some(value)
        })
        .unwrap_or("");
    Some((measurement, namespace))
}