nodedb-lite 0.1.0

// SPDX-License-Identifier: Apache-2.0
//! Ingest handler for the timeseries physical visitor.

use std::collections::HashSet;
use std::sync::Mutex;

use nodedb_types::columnar::{ColumnDef, ColumnType, ColumnarProfile};
use nodedb_types::datetime::NdbDateTime;
use nodedb_types::result::QueryResult;
use nodedb_types::timeseries::MetricSample;
use nodedb_types::value::Value;

use crate::error::LiteError;
use crate::query::engine::LiteQueryEngine;
use crate::storage::engine::StorageEngine;

/// Per-process, per-collection deduplication set for WAL-LSN replay.
/// Keyed by (collection, lsn). Cleared on process restart — that is
/// acceptable because the WAL LSN dedup is a best-effort guard against
/// double-replay on crash recovery; after a clean restart the WAL is
/// replayed from scratch anyway.
static SEEN_LSNS: std::sync::LazyLock<Mutex<HashSet<(String, u64)>>> =
    std::sync::LazyLock::new(|| Mutex::new(HashSet::new()));

/// Ingest samples into a timeseries collection.
///
/// Decodes `payload` per `format` ("ilp", "msgpack", "samples"), performs
/// WAL-LSN deduplication when `wal_lsn` is `Some`, and delegates to
/// `ingest_metric` for each decoded sample.
///
/// Returns the `QueryResult` together with the decoded samples. The caller is
/// responsible for converting the samples to column-ordered rows (using the
/// collection's columnar schema) and calling `engine.columnar.enqueue_outbound`
/// from an async context to durably queue them for replication to Origin.
pub fn ingest<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    collection: &str,
    payload: &[u8],
    format: &str,
    wal_lsn: Option<u64>,
    surrogates: &[nodedb_types::Surrogate],
) -> Result<(QueryResult, Vec<ParsedSample>), LiteError> {
    // WAL-LSN deduplication: skip the entire batch if already applied.
    if let Some(lsn) = wal_lsn {
        let mut seen = SEEN_LSNS.lock().map_err(|_| LiteError::LockPoisoned)?;
        let key = (collection.to_string(), lsn);
        if seen.contains(&key) {
            return Ok((
                QueryResult {
                    columns: Vec::new(),
                    rows: Vec::new(),
                    rows_affected: 0,
                    command: Some("INSERT".into()),
                },
                Vec::new(),
            ));
        }
        seen.insert(key);
    }

    let time_key = declared_time_key(engine, collection);
    let samples = decode_payload(payload, format, time_key.as_deref())?;
    let count = samples.len() as u64;

    let _ = surrogates; // Lite TS engine assigns internal series IDs; surrogate
    // allocation for cross-engine identity is managed by the
    // Origin CP before dispatch. For opaque payloads (ILP,
    // raw msgpack) the CP cannot pre-assign surrogates per
    // the TimeseriesOp definition; Lite uses its own series
    // catalog for identity within the embedded context.

    {
        let mut ts_engine = engine
            .timeseries
            .lock()
            .map_err(|_| LiteError::LockPoisoned)?;

        for (metric_name, tags, sample) in &samples {
            ts_engine.ingest_metric(collection, metric_name, tags.clone(), *sample);
        }
    }

    Ok((
        QueryResult {
            columns: Vec::new(),
            rows: Vec::new(),
            rows_affected: count,
            command: Some("INSERT".into()),
        },
        samples,
    ))
}

/// The `TIME_KEY` a timeseries-profile collection declares. `None` for a
/// collection without a columnar schema or with a non-timeseries profile.
pub fn declared_time_key<S: StorageEngine>(
    engine: &LiteQueryEngine<S>,
    collection: &str,
) -> Option<String> {
    match engine.columnar.profile(collection)? {
        ColumnarProfile::Timeseries { time_key, .. } => Some(time_key),
        ColumnarProfile::Plain | ColumnarProfile::Spatial { .. } => None,
    }
}

/// Is `column` the time column of the row being ingested? Matches only the
/// declared `TIME_KEY` when DDL exists; falls back to conventional names
/// (`ts`/`timestamp`/`time`/`timestamp_ms`) for a collection with no DDL
/// behind it.
fn is_time_column(column: &str, declared: Option<&str>) -> bool {
    match declared {
        Some(time_key) => column.eq_ignore_ascii_case(time_key),
        None => {
            column.eq_ignore_ascii_case("ts")
                || column.eq_ignore_ascii_case("timestamp")
                || column.eq_ignore_ascii_case("time")
                || column.eq_ignore_ascii_case("timestamp_ms")
        }
    }
}

/// The millisecond timestamp a time-column cell denotes: `None` when the cell
/// is NULL (the ingest clock stamps the row), an error naming the column when
/// the cell holds a value no timestamp can be read from.
///
/// Accepts the shapes Origin's Data Plane accepts: a typed instant of either
/// kind, an integer or float millisecond count, and ISO-8601 / RFC 3339 text.
fn time_cell_millis(cell: &Value, column: &str) -> Result<Option<i64>, LiteError> {
    let refused = |literal: String| LiteError::BadRequest {
        detail: format!("timeseries ingest: {column}={literal} is not a timestamp"),
    };
    match cell {
        Value::Null => Ok(None),
        Value::Integer(ms) => Ok(Some(*ms)),
        Value::Float(f) => {
            let whole = f.trunc();
            if !whole.is_finite() || whole < i64::MIN as f64 || whole >= i64::MAX as f64 {
                return Err(refused(f.to_string()));
            }
            Ok(Some(whole as i64))
        }
        Value::NaiveDateTime(at) | Value::DateTime(at) => Ok(Some(at.unix_millis())),
        Value::String(s) => NdbDateTime::parse(s)
            .map(|at| Some(at.unix_millis()))
            .ok_or_else(|| refused(format!("\"{s}\""))),
        Value::Bool(b) => Err(refused(b.to_string())),
        Value::Decimal(d) => Err(refused(d.to_string())),
        Value::Uuid(id) | Value::Ulid(id) => Err(refused(id.clone())),
        Value::Bytes(_) => Err(refused("a byte string".into())),
        Value::Array(_) | Value::Set(_) => Err(refused("an array".into())),
        Value::Object(_) => Err(refused("an object".into())),
        Value::Geometry(_) => Err(refused("a geometry".into())),
        Value::Duration(_) => Err(refused("a duration".into())),
        Value::Regex(_) => Err(refused("a regex".into())),
        Value::Range { .. } => Err(refused("a range".into())),
        Value::Record { .. } => Err(refused("a record reference".into())),
        Value::ArrayCell(_) => Err(refused("an array cell".into())),
        Value::Vector(_) => Err(refused("a vector".into())),
        other => Err(refused(format!("{other:?}"))),
    }
}

/// Convert decoded `ParsedSample`s to schema-ordered rows.
///
/// The declared time column (`time_key`, else a conventional name) receives
/// the sample timestamp typed by its column kind: a `TIMESTAMP` column holds
/// `Value::NaiveDateTime`, a `TIMESTAMPTZ` column `Value::DateTime`, any
/// other kind the millisecond count. Columns named `value`/`val`/`v` receive
/// the numeric value. Every other column is `Value::Null`.
pub fn samples_to_rows(
    samples: &[ParsedSample],
    columns: &[ColumnDef],
    time_key: Option<&str>,
) -> Result<Vec<Vec<Value>>, LiteError> {
    samples
        .iter()
        .map(|(_, _, sample)| {
            columns
                .iter()
                .map(|col| {
                    if is_time_column(&col.name, time_key) {
                        return typed_time_value(sample.timestamp_ms, &col.column_type);
                    }
                    Ok(match col.name.to_ascii_lowercase().as_str() {
                        "value" | "val" | "v" => Value::Float(sample.value),
                        _ => Value::Null,
                    })
                })
                .collect()
        })
        .collect()
}

/// The stored form of a millisecond timestamp under a column of `col_type`.
fn typed_time_value(timestamp_ms: i64, col_type: &ColumnType) -> Result<Value, LiteError> {
    let instant = || {
        NdbDateTime::from_millis(timestamp_ms).map_err(|e| LiteError::BadRequest {
            detail: format!("timeseries ingest: {e}"),
        })
    };
    match col_type {
        ColumnType::Timestamp => instant().map(Value::NaiveDateTime),
        ColumnType::Timestamptz => instant().map(Value::DateTime),
        ColumnType::Int64
        | ColumnType::Float64
        | ColumnType::String
        | ColumnType::Bool
        | ColumnType::Bytes
        | ColumnType::SystemTimestamp
        | ColumnType::Decimal { .. }
        | ColumnType::Geometry
        | ColumnType::Vector(_)
        | ColumnType::SparseVector
        | ColumnType::Uuid
        | ColumnType::Json
        | ColumnType::Ulid
        | ColumnType::Duration
        | ColumnType::Array => Ok(Value::Integer(timestamp_ms)),
        other => Err(LiteError::Unsupported {
            detail: format!("column type {other} has no timeseries time-key form in Lite"),
        }),
    }
}

// ── Payload decoders ──────────────────────────────────────────────────────────

/// Decoded sample ready for `ingest_metric`.
pub type ParsedSample = (String, Vec<(String, String)>, MetricSample);

fn decode_payload(
    payload: &[u8],
    format: &str,
    time_key: Option<&str>,
) -> Result<Vec<ParsedSample>, LiteError> {
    match format {
        "ilp" => parse_ilp(payload),
        "msgpack" => parse_msgpack(payload, time_key),
        "samples" | "structured" => parse_structured(payload, time_key),
        other => Err(LiteError::BadRequest {
            detail: format!(
                "unknown timeseries ingest format '{other}'; expected ilp/msgpack/samples"
            ),
        }),
    }
}

/// Test-only re-export of the ILP parser so integration tests can verify
/// round-trip correctness without going through the full ingest stack.
/// Expose the ILP parser for integration tests.
pub fn parse_ilp_for_test(payload: &[u8]) -> Result<Vec<ParsedSample>, LiteError> {
    parse_ilp(payload)
}

// ── ILP parser ────────────────────────────────────────────────────────────────

/// Minimal InfluxDB Line Protocol parser for timeseries ingest.
///
/// Grammar: `measurement[,tag=val]* field=val[,field=val]* [timestamp_ns]`
///
/// - Each line produces one sample.
/// - The first numeric field found is used as `value`.
/// - The trailing integer (if present) is the timestamp in nanoseconds;
///   we convert to milliseconds by dividing by 1_000_000.
/// - Tags become the `tags` vector.
/// - `measurement` is the metric name.
fn parse_ilp(payload: &[u8]) -> Result<Vec<ParsedSample>, LiteError> {
    let text = std::str::from_utf8(payload).map_err(|e| LiteError::Serialization {
        detail: format!("ILP payload is not valid UTF-8: {e}"),
    })?;

    let mut out: Vec<ParsedSample> = Vec::new();

    for raw_line in text.lines() {
        let line = raw_line.trim();
        if line.is_empty() || line.starts_with('#') {
            continue;
        }

        // Split into key_part and rest on first unescaped space.
        let (key_part, rest) = split_ilp_space(line).ok_or_else(|| LiteError::Serialization {
            detail: format!("ILP line missing field set: {line}"),
        })?;

        // Optionally split timestamp off the trailing rest.
        let (fields_part, timestamp_str) = split_ilp_space(rest)
            .map(|(f, t)| (f, Some(t)))
            .unwrap_or((rest, None));

        // Derive measurement and tags.
        let mut key_iter = key_part.splitn(2, ',');
        let measurement = key_iter.next().unwrap_or("unknown").to_string();

        let mut tags: Vec<(String, String)> = Vec::new();
        if let Some(tag_str) = key_iter.next() {
            for kv in tag_str.split(',') {
                if let Some((k, v)) = kv.split_once('=') {
                    tags.push((k.to_string(), v.to_string()));
                }
            }
        }

        // Parse fields — use the first numeric field as the sample value.
        let mut value_opt: Option<f64> = None;
        for kv in fields_part.split(',') {
            if let Some((_k, v)) = kv.split_once('=')
                && let Some(f) = parse_numeric(v)
            {
                value_opt = Some(f);
                break;
            }
        }

        let value = value_opt.unwrap_or(0.0);

        // Parse timestamp: nanoseconds → milliseconds.
        let timestamp_ms = timestamp_str
            .and_then(|s| s.parse::<i64>().ok())
            .map(|ns| ns / 1_000_000)
            .unwrap_or_else(crate::runtime::now_millis_i64);

        out.push((
            measurement,
            tags,
            MetricSample {
                timestamp_ms,
                value,
            },
        ));
    }

    Ok(out)
}

fn split_ilp_space(s: &str) -> Option<(&str, &str)> {
    let bytes = s.as_bytes();
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'\\' {
            i += 2;
            continue;
        }
        if bytes[i] == b' ' {
            return Some((&s[..i], s[i + 1..].trim_start()));
        }
        i += 1;
    }
    None
}

fn parse_numeric(v: &str) -> Option<f64> {
    if let Some(stripped) = v.strip_suffix('i') {
        return stripped.parse::<i64>().ok().map(|n| n as f64);
    }
    v.parse::<f64>().ok()
}

// ── MessagePack decoder ───────────────────────────────────────────────────────

/// Decode a msgpack-encoded array of sample objects.
///
/// Expected shape: `[{metric, value, <time column>?, tags?}, ...]`
/// or a single object `{metric, value, <time column>?, tags?}`. The time
/// column is the declared `time_key`, else a conventional name; its cell
/// takes any shape `time_cell_millis` reads.
fn parse_msgpack(payload: &[u8], time_key: Option<&str>) -> Result<Vec<ParsedSample>, LiteError> {
    let top: Value = zerompk::from_msgpack(payload).map_err(|e| LiteError::Serialization {
        detail: format!("msgpack timeseries payload: {e}"),
    })?;

    match top {
        Value::Array(items) => items
            .into_iter()
            .map(|item| decode_msgpack_sample(item, time_key))
            .collect(),
        obj @ Value::Object(_) => decode_msgpack_sample(obj, time_key).map(|s| vec![s]),
        _ => Err(LiteError::Serialization {
            detail: "msgpack timeseries payload must be an object or array of objects".into(),
        }),
    }
}

fn decode_msgpack_sample(v: Value, time_key: Option<&str>) -> Result<ParsedSample, LiteError> {
    let Value::Object(map) = v else {
        return Err(LiteError::Serialization {
            detail: "each msgpack timeseries sample must be an object".into(),
        });
    };

    let metric = match map.get("metric").or_else(|| map.get("name")) {
        Some(Value::String(s)) => s.clone(),
        _ => "value".to_string(),
    };

    let value = match map.get("value") {
        Some(Value::Float(f)) => *f,
        Some(Value::Integer(i)) => *i as f64,
        _ => 0.0,
    };

    let mut timestamp_ms: Option<i64> = None;
    for (key, cell) in &map {
        if is_time_column(key, time_key) {
            timestamp_ms = time_cell_millis(cell, key)?;
            break;
        }
    }
    let timestamp_ms = timestamp_ms.unwrap_or_else(crate::runtime::now_millis_i64);

    let tags = match map.get("tags") {
        Some(Value::Object(t)) => t
            .iter()
            .map(|(k, v)| {
                let val = match v {
                    Value::String(s) => s.clone(),
                    other => format!("{other:?}"),
                };
                (k.clone(), val)
            })
            .collect(),
        _ => Vec::new(),
    };

    Ok((
        metric,
        tags,
        MetricSample {
            timestamp_ms,
            value,
        },
    ))
}

// ── Structured / samples decoder ──────────────────────────────────────────────

/// Decode a structured payload: msgpack-encoded `Vec<MetricSample>` with
/// a flat metric name stored at key `"_metric"` in the outer object, or
/// a msgpack array of `{metric, value, timestamp_ms}` objects.
///
/// This is the format produced by the CP when it can enumerate rows at
/// planning time (SQL VALUES path). It falls back to msgpack object decode.
fn parse_structured(
    payload: &[u8],
    time_key: Option<&str>,
) -> Result<Vec<ParsedSample>, LiteError> {
    // Try as flat JSON array of MetricSample objects (CP-produced path with
    // surrogate pre-assignment). MetricSample is serde-only (not zerompk),
    // so try JSON decode first before falling back to msgpack object decode.
    if let Ok(samples) = sonic_rs::from_slice::<Vec<MetricSample>>(payload) {
        return Ok(samples
            .into_iter()
            .map(|s| ("value".to_string(), Vec::new(), s))
            .collect());
    }

    // Fall back to object/array decode.
    parse_msgpack(payload, time_key)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn ilp_round_trip_basic() {
        let ilp = b"cpu,host=server01 usage=0.75 1700000000000000000\n";
        let samples = parse_ilp(ilp).expect("parse ILP");
        assert_eq!(samples.len(), 1);
        let (metric, tags, sample) = &samples[0];
        assert_eq!(metric, "cpu");
        assert_eq!(tags[0], ("host".to_string(), "server01".to_string()));
        assert!((sample.value - 0.75_f64).abs() < 1e-9);
        // 1700000000000000000 ns / 1_000_000 = 1_700_000_000_000 ms
        assert_eq!(sample.timestamp_ms, 1_700_000_000_000);
    }

    #[test]
    fn ilp_multiple_lines() {
        let ilp = b"mem,host=a free=1024i 1000000000\nmem,host=b free=2048i 2000000000\n";
        let samples = parse_ilp(ilp).expect("parse ILP multi-line");
        assert_eq!(samples.len(), 2);
    }

    #[test]
    fn ilp_skips_comments() {
        let ilp = b"# comment\ncpu usage=1.0 1000000000\n";
        let samples = parse_ilp(ilp).expect("ILP with comment");
        assert_eq!(samples.len(), 1);
    }

    #[test]
    fn msgpack_round_trip() {
        use std::collections::HashMap;
        let mut obj: HashMap<String, Value> = HashMap::new();
        obj.insert("metric".into(), Value::String("temperature".into()));
        obj.insert("value".into(), Value::Float(22.5));
        obj.insert("timestamp_ms".into(), Value::Integer(1_700_000_000_000));
        let bytes = zerompk::to_msgpack_vec(&Value::Object(obj)).expect("encode");
        let samples = parse_msgpack(&bytes, None).expect("parse msgpack");
        assert_eq!(samples.len(), 1);
        let (metric, _, sample) = &samples[0];
        assert_eq!(metric, "temperature");
        assert!((sample.value - 22.5).abs() < 1e-9);
        assert_eq!(sample.timestamp_ms, 1_700_000_000_000);
    }

    #[test]
    fn structured_flat_metric_samples() {
        let samples = vec![
            MetricSample {
                timestamp_ms: 1000,
                value: 1.0,
            },
            MetricSample {
                timestamp_ms: 2000,
                value: 2.0,
            },
        ];
        // structured format falls back to JSON decode of Vec<MetricSample>.
        let bytes = sonic_rs::to_vec(&samples).expect("encode");
        let parsed = parse_structured(&bytes, None).expect("parse structured");
        assert_eq!(parsed.len(), 2);
        assert!((parsed[0].2.value - 1.0).abs() < 1e-9);
        assert!((parsed[1].2.value - 2.0).abs() < 1e-9);
    }

    /// `2024-01-01T00:00:00Z` in epoch milliseconds.
    const NEW_YEAR_MS: i64 = 1_704_067_200_000;

    fn sample_row(time_col: &str, cell: Value) -> Vec<u8> {
        use std::collections::HashMap;
        let mut obj: HashMap<String, Value> = HashMap::new();
        obj.insert(time_col.into(), cell);
        obj.insert("value".into(), Value::Float(1.0));
        zerompk::to_msgpack_vec(&Value::Object(obj)).expect("encode")
    }

    /// A typed instant, an integer, or ISO-8601 text under the declared time
    /// key all give the same millisecond timestamp.
    #[test]
    fn declared_time_key_reads_every_instant_shape() {
        let at = NdbDateTime::from_millis(NEW_YEAR_MS).expect("instant");
        for cell in [
            Value::NaiveDateTime(at),
            Value::DateTime(at),
            Value::Integer(NEW_YEAR_MS),
            Value::Float(NEW_YEAR_MS as f64 + 0.5),
            Value::String("2024-01-01 00:00:00".into()),
            Value::String("2024-01-01T00:00:00Z".into()),
        ] {
            let samples = parse_msgpack(&sample_row("time", cell.clone()), Some("time"))
                .expect("parse msgpack");
            assert_eq!(samples[0].2.timestamp_ms, NEW_YEAR_MS, "cell {cell:?}");
        }
    }

    /// With DDL, only the declared time key is the time column: a `ts` cell
    /// under a `captured_at` key is a plain field and the clock stamps the row.
    #[test]
    fn declared_time_key_overrides_conventional_names() {
        let before = crate::runtime::now_millis_i64();
        let samples = parse_msgpack(
            &sample_row("ts", Value::Integer(NEW_YEAR_MS)),
            Some("captured_at"),
        )
        .expect("parse msgpack");
        assert!(samples[0].2.timestamp_ms >= before);
    }

    /// Without DDL the conventional names resolve, case-insensitively.
    #[test]
    fn conventional_time_names_resolve_without_ddl() {
        for name in ["time", "Timestamp", "TS", "timestamp_ms"] {
            let samples = parse_msgpack(&sample_row(name, Value::Integer(NEW_YEAR_MS)), None)
                .expect("parse msgpack");
            assert_eq!(samples[0].2.timestamp_ms, NEW_YEAR_MS, "name {name}");
        }
    }

    /// A time cell no timestamp can be read from is an error naming the
    /// column, never a row stamped with the clock.
    #[test]
    fn unreadable_time_cell_is_refused() {
        for cell in [Value::String("not a date".into()), Value::Bool(true)] {
            let err = parse_msgpack(&sample_row("time", cell), Some("time")).expect_err("refused");
            assert!(
                matches!(&err, LiteError::BadRequest { detail } if detail.contains("time=")),
                "{err:?}"
            );
        }
    }

    /// Rows for outbound sync carry the timestamp typed by the time column's
    /// declared kind.
    #[test]
    fn samples_to_rows_types_the_time_column() {
        let samples = vec![(
            "value".to_string(),
            Vec::new(),
            MetricSample {
                timestamp_ms: NEW_YEAR_MS,
                value: 2.5,
            },
        )];
        let at = NdbDateTime::from_millis(NEW_YEAR_MS).expect("instant");
        let columns = vec![
            ColumnDef::required("time", ColumnType::Timestamp),
            ColumnDef::nullable("host", ColumnType::String),
            ColumnDef::nullable("value", ColumnType::Float64),
        ];
        let rows = samples_to_rows(&samples, &columns, Some("time")).expect("rows");
        assert_eq!(
            rows,
            vec![vec![
                Value::NaiveDateTime(at),
                Value::Null,
                Value::Float(2.5)
            ]]
        );

        let tz_columns = vec![ColumnDef::required("at", ColumnType::Timestamptz)];
        let rows = samples_to_rows(&samples, &tz_columns, Some("at")).expect("rows");
        assert_eq!(rows, vec![vec![Value::DateTime(at)]]);

        let int_columns = vec![ColumnDef::required("ts", ColumnType::Int64)];
        let rows = samples_to_rows(&samples, &int_columns, None).expect("rows");
        assert_eq!(rows, vec![vec![Value::Integer(NEW_YEAR_MS)]]);
    }

    #[test]
    fn lsn_dedup_skips_duplicate() {
        // The SEEN_LSNS map is process-wide; use a unique collection name per test.
        let key = ("__test_dedup_coll__".to_string(), 9999u64);
        {
            let mut seen = SEEN_LSNS.lock().unwrap();
            seen.insert(key.clone());
        }
        // Verify the key is present.
        let seen = SEEN_LSNS.lock().unwrap();
        assert!(seen.contains(&key));
    }
}