faucet-core 1.11.0

//! Incremental replication support.

use crate::error::FaucetError;
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::cmp::Ordering;

/// Determines how records are replicated from the source.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "type")]
pub enum ReplicationMethod {
    /// All records are fetched on every run (default).
    #[default]
    FullTable,
    /// Only records where the `replication_key` field is strictly greater than
    /// the stored bookmark (`start_replication_value`) are kept.
    Incremental,
}

/// Filter `records` to only those where `record[key] > start`.
///
/// Records missing the key are excluded. Strings compare lexicographically
/// (ISO-8601 dates compare correctly this way); integers compare exactly
/// (no `f64` precision loss); floats compare as `f64`.
///
/// If a record's key value is a *different JSON type* than `start` (e.g. a
/// numeric key against a string bookmark), the comparison is not meaningful;
/// rather than silently dropping the record — which is data loss (#78/#27) —
/// it is **kept** and a warning is logged.
pub fn filter_incremental(records: Vec<Value>, key: &str, start: &Value) -> Vec<Value> {
    records
        .into_iter()
        .filter(|r| match r.get(key) {
            None => false,
            Some(v) if type_rank(v) != type_rank(start) => {
                tracing::warn!(
                    key,
                    "incremental replication: record key type does not match the bookmark \
                     type; keeping the record to avoid silently dropping data"
                );
                true
            }
            Some(v) => json_gt(v, start),
        })
        .collect()
}

/// Return the maximum value of `record[key]` across all records, if any.
pub fn max_replication_value<'a>(records: &'a [Value], key: &str) -> Option<&'a Value> {
    records
        .iter()
        .filter_map(|r| r.get(key))
        .max_by(|a, b| json_compare(a, b))
}

/// Return the larger of two replication values using the same ordering as
/// [`max_replication_value`] (string lexicographic, numeric for numbers,
/// falling back to `a` on type mismatch).
pub fn max_value(a: Value, b: Value) -> Value {
    match json_compare(&a, &b) {
        Ordering::Less => b,
        _ => a,
    }
}

/// Type-rank for a total ordering across JSON value kinds, so comparisons of
/// differing types are deterministic instead of collapsing to `Equal`.
fn type_rank(v: &Value) -> u8 {
    match v {
        Value::Null => 0,
        Value::Bool(_) => 1,
        Value::Number(_) => 2,
        Value::String(_) => 3,
        Value::Array(_) => 4,
        Value::Object(_) => 5,
    }
}

/// Exact integer view of a JSON number (`i64` or `u64`), widened to `i128` so
/// both halves of the range compare without `f64` precision loss. `None` for
/// non-integral (floating) numbers.
fn number_as_i128(n: &serde_json::Number) -> Option<i128> {
    n.as_i64()
        .map(i128::from)
        .or_else(|| n.as_u64().map(i128::from))
}

/// Total ordering over JSON values used for replication bookmarks.
///
/// - Numbers: compared exactly as `i128` when both are integral (so cursors
///   above 2^53 don't lose precision); otherwise as `f64`, with NaN ordered
///   last.
/// - Same-type scalars/containers: natural ordering (strings lexicographic,
///   bools `false < true`, arrays element-wise, objects by serialized form).
/// - Different types: ordered by [`type_rank`] so the result is always total.
pub(crate) fn json_compare(a: &Value, b: &Value) -> Ordering {
    match (a, b) {
        (Value::Number(an), Value::Number(bn)) => {
            match (number_as_i128(an), number_as_i128(bn)) {
                (Some(ai), Some(bi)) => ai.cmp(&bi),
                _ => {
                    let af = an.as_f64().unwrap_or(f64::NAN);
                    let bf = bn.as_f64().unwrap_or(f64::NAN);
                    af.partial_cmp(&bf).unwrap_or_else(|| {
                        // At least one NaN — order NaN last, deterministically.
                        match (af.is_nan(), bf.is_nan()) {
                            (false, true) => Ordering::Less,
                            (true, false) => Ordering::Greater,
                            _ => Ordering::Equal,
                        }
                    })
                }
            }
        }
        (Value::String(x), Value::String(y)) => x.cmp(y),
        (Value::Bool(x), Value::Bool(y)) => x.cmp(y),
        (Value::Null, Value::Null) => Ordering::Equal,
        (Value::Array(x), Value::Array(y)) => {
            for (xi, yi) in x.iter().zip(y.iter()) {
                let c = json_compare(xi, yi);
                if c != Ordering::Equal {
                    return c;
                }
            }
            x.len().cmp(&y.len())
        }
        // Objects have no natural order; use the serialized form for a stable
        // total order (objects as replication keys are pathological).
        (Value::Object(_), Value::Object(_)) => a.to_string().cmp(&b.to_string()),
        // Different JSON types — order by type rank so comparison is total.
        _ => type_rank(a).cmp(&type_rank(b)),
    }
}

/// Total-order "greater than" over JSON values, using the same comparison
/// [`filter_incremental`] applies to replication keys (numbers numerically,
/// strings lexicographically — so RFC3339 timestamps order correctly). Public
/// so callers bounding a replay window (e.g. `faucet backfill --to-bookmark`)
/// compare exactly like the incremental filter does.
pub fn json_gt(a: &Value, b: &Value) -> bool {
    json_compare(a, b) == Ordering::Greater
}

// ── Server-side incremental push-down (#513) ─────────────────────────────────

/// The placeholder replaced by the formatted bookmark inside a
/// [`ReplicationBind::template`].
pub const BIND_PLACEHOLDER: &str = "${bookmark}";

fn default_bind_template() -> String {
    BIND_PLACEHOLDER.to_owned()
}

/// Where a rendered bookmark is injected into the outgoing request.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum BindTarget {
    /// A query-string parameter (default) — e.g. `?updated_after=…`.
    #[default]
    Query,
    /// A request header — e.g. `If-Modified-Since: …`.
    Header,
    /// A top-level field of the JSON request body (POST-search APIs).
    Body,
    /// A `{name}` placeholder in the request path.
    Path,
}

/// How the bookmark value is formatted before it is substituted into the
/// [`ReplicationBind::template`].
///
/// For every non-[`Raw`](BindFormat::Raw) format the bookmark is first parsed
/// into an instant: a string is read as RFC 3339, a bare `YYYY-MM-DD` date
/// (midnight UTC), or a naive `YYYY-MM-DDTHH:MM:SS` (assumed UTC); a JSON
/// number is read as **epoch seconds**. It is then re-emitted in the target
/// representation, so `epoch_ms` ← ISO string and `iso8601` ← epoch number both
/// work.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum BindFormat {
    /// Emit the scalar verbatim (string as-is, number as its decimal form).
    /// The default; no timestamp parsing.
    #[default]
    Raw,
    /// RFC 3339 / ISO-8601 UTC timestamp, e.g. `2024-06-01T00:00:00Z`.
    Iso8601,
    /// Unix epoch **seconds** (integer).
    EpochS,
    /// Unix epoch **milliseconds** (integer).
    EpochMs,
    /// Calendar date `YYYY-MM-DD` (UTC).
    Date,
}

/// Declarative binding of the stored bookmark into the **outgoing request** —
/// "server-side incremental push-down" (#513).
///
/// Today faucet tracks bookmarks and filters incrementally *client-side* (after
/// download). A bind lets a source instead push the bookmark into the request
/// (query param / header / body field / path) so the server returns only the
/// new rows. The existing client-side [`filter_incremental`] stays active as a
/// safety net for servers that don't honour the filter exactly.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct ReplicationBind {
    /// Where to place the rendered value.
    #[serde(default)]
    pub into: BindTarget,
    /// The parameter / header / body-field / path-placeholder name.
    pub name: String,
    /// Template rendered with [`BIND_PLACEHOLDER`] (`${bookmark}`) replaced by
    /// the formatted bookmark. Defaults to the bare `${bookmark}`; set e.g.
    /// `"gte|${bookmark}"` (Greenhouse) or `"[${bookmark} TO *]"` (Lucene).
    #[serde(default = "default_bind_template")]
    pub template: String,
    /// How to format the bookmark before substitution.
    #[serde(default)]
    pub format: BindFormat,
    /// Optional JSONPath into the response body to advance the bookmark from,
    /// instead of `max(record[replication_key])`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub advance_from: Option<String>,
}

impl ReplicationBind {
    /// Validate the binding at config-load time.
    pub fn validate(&self) -> Result<(), FaucetError> {
        if self.name.trim().is_empty() {
            return Err(FaucetError::Config(
                "replication bind: `name` must not be empty".to_owned(),
            ));
        }
        if !self.template.contains(BIND_PLACEHOLDER) {
            return Err(FaucetError::Config(format!(
                "replication bind: `template` must contain the `{BIND_PLACEHOLDER}` placeholder"
            )));
        }
        Ok(())
    }

    /// Render the binding for a concrete bookmark: format the value, then
    /// substitute it into the template.
    pub fn render(&self, bookmark: &Value) -> Result<String, FaucetError> {
        let formatted = format_bookmark(bookmark, self.format)?;
        Ok(self.template.replace(BIND_PLACEHOLDER, &formatted))
    }
}

/// Parse a bookmark value into a UTC instant (see [`BindFormat`] for the rules).
fn bookmark_instant(value: &Value) -> Result<DateTime<Utc>, FaucetError> {
    match value {
        Value::String(s) => {
            let s = s.trim();
            if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
                return Ok(dt.with_timezone(&Utc));
            }
            if let Ok(d) = NaiveDate::parse_from_str(s, "%Y-%m-%d")
                && let Some(ndt) = d.and_hms_opt(0, 0, 0)
            {
                return Ok(DateTime::<Utc>::from_naive_utc_and_offset(ndt, Utc));
            }
            if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S") {
                return Ok(DateTime::<Utc>::from_naive_utc_and_offset(ndt, Utc));
            }
            Err(FaucetError::Config(format!(
                "replication bind: cannot parse bookmark '{s}' as a timestamp \
                 (expected RFC 3339, YYYY-MM-DD, or YYYY-MM-DDTHH:MM:SS)"
            )))
        }
        Value::Number(n) => {
            let secs = n.as_i64().or_else(|| n.as_f64().map(|f| f as i64));
            secs.and_then(|s| DateTime::<Utc>::from_timestamp(s, 0))
                .ok_or_else(|| {
                    FaucetError::Config(format!(
                        "replication bind: numeric bookmark {n} is out of range for epoch seconds"
                    ))
                })
        }
        other => Err(FaucetError::Config(format!(
            "replication bind: bookmark must be a string or number, got {other}"
        ))),
    }
}

/// Parse a bookmark value into a UTC instant (public wrapper over the internal
/// parser; see [`BindFormat`] for the accepted forms). Used by datetime window
/// slicing (#527) to resolve the sweep's start bound from the stored bookmark.
pub fn parse_instant(value: &Value) -> Result<DateTime<Utc>, FaucetError> {
    bookmark_instant(value)
}

/// Format an already-resolved UTC instant per [`BindFormat`]. Unlike
/// [`format_bookmark`] (which takes an arbitrary scalar and, for [`BindFormat::Raw`],
/// echoes it verbatim), this always has a real instant, so `Raw` and `Iso8601`
/// both emit an RFC 3339 UTC timestamp. Used to render window boundaries (#527).
pub fn format_instant(dt: DateTime<Utc>, format: BindFormat) -> String {
    match format {
        BindFormat::Raw | BindFormat::Iso8601 => {
            dt.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
        }
        BindFormat::Date => dt.format("%Y-%m-%d").to_string(),
        BindFormat::EpochS => dt.timestamp().to_string(),
        BindFormat::EpochMs => dt.timestamp_millis().to_string(),
    }
}

/// Format a bookmark value per [`BindFormat`].
pub fn format_bookmark(value: &Value, format: BindFormat) -> Result<String, FaucetError> {
    match format {
        BindFormat::Raw => match value {
            Value::String(s) => Ok(s.clone()),
            Value::Number(n) => Ok(n.to_string()),
            Value::Bool(b) => Ok(b.to_string()),
            other => Err(FaucetError::Config(format!(
                "replication bind: cannot render {other} as a raw scalar"
            ))),
        },
        BindFormat::Iso8601 => {
            Ok(bookmark_instant(value)?.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
        }
        BindFormat::Date => Ok(bookmark_instant(value)?.format("%Y-%m-%d").to_string()),
        BindFormat::EpochS => Ok(bookmark_instant(value)?.timestamp().to_string()),
        BindFormat::EpochMs => Ok(bookmark_instant(value)?.timestamp_millis().to_string()),
    }
}

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

    #[test]
    fn test_filter_incremental_strings() {
        let records = vec![
            json!({"id": 1, "updated_at": "2024-01-01"}),
            json!({"id": 2, "updated_at": "2024-06-01"}),
            json!({"id": 3, "updated_at": "2024-12-01"}),
        ];
        let start = json!("2024-06-01");
        let filtered = filter_incremental(records, "updated_at", &start);
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0]["id"], 3);
    }

    #[test]
    fn test_filter_incremental_numbers() {
        let records = vec![
            json!({"id": 1, "seq": 100}),
            json!({"id": 2, "seq": 200}),
            json!({"id": 3, "seq": 300}),
        ];
        let start = json!(150);
        let filtered = filter_incremental(records, "seq", &start);
        assert_eq!(filtered.len(), 2);
        assert_eq!(filtered[0]["id"], 2);
        assert_eq!(filtered[1]["id"], 3);
    }

    #[test]
    fn test_filter_incremental_missing_key_excluded() {
        let records = vec![
            json!({"id": 1}),
            json!({"id": 2, "updated_at": "2024-12-01"}),
        ];
        let start = json!("2024-01-01");
        let filtered = filter_incremental(records, "updated_at", &start);
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0]["id"], 2);
    }

    #[test]
    fn test_filter_incremental_equal_excluded() {
        let records = vec![
            json!({"id": 1, "updated_at": "2024-06-01"}),
            json!({"id": 2, "updated_at": "2024-06-02"}),
        ];
        let start = json!("2024-06-01");
        let filtered = filter_incremental(records, "updated_at", &start);
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0]["id"], 2);
    }

    #[test]
    fn test_max_replication_value_strings() {
        let records = vec![
            json!({"updated_at": "2024-01-01"}),
            json!({"updated_at": "2024-12-01"}),
            json!({"updated_at": "2024-06-01"}),
        ];
        let max = max_replication_value(&records, "updated_at").unwrap();
        assert_eq!(max, &json!("2024-12-01"));
    }

    #[test]
    fn test_max_replication_value_numbers() {
        let records = vec![json!({"seq": 5}), json!({"seq": 10}), json!({"seq": 3})];
        let max = max_replication_value(&records, "seq").unwrap();
        assert_eq!(max, &json!(10));
    }

    #[test]
    fn test_max_replication_value_empty() {
        let records: Vec<Value> = vec![];
        assert!(max_replication_value(&records, "updated_at").is_none());
    }

    #[test]
    fn test_max_value_picks_larger_string() {
        assert_eq!(
            max_value(json!("2024-01-01"), json!("2024-06-01")),
            json!("2024-06-01")
        );
    }

    #[test]
    fn test_max_value_picks_larger_number() {
        assert_eq!(max_value(json!(5), json!(10)), json!(10));
    }

    #[test]
    fn test_max_value_returns_a_on_type_mismatch() {
        // String outranks Number in the total type-rank ordering, so the
        // larger (a) is returned.
        assert_eq!(max_value(json!("string"), json!(5)), json!("string"));
    }

    #[test]
    fn filter_incremental_keeps_large_integer_beyond_f64_precision() {
        // Regression for #78/#27: integer cursors above 2^53 lose precision
        // when compared as f64, so a genuinely-greater value compared Equal
        // and was silently dropped.
        let two_pow_53 = 9_007_199_254_740_992_i64; // 2^53
        let records = vec![
            json!({"id": 1, "seq": two_pow_53 + 1}),
            json!({"id": 2, "seq": two_pow_53 + 2}),
        ];
        let start = json!(two_pow_53);
        let filtered = filter_incremental(records, "seq", &start);
        assert_eq!(
            filtered.len(),
            2,
            "both values are strictly greater than 2^53"
        );
    }

    #[test]
    fn json_compare_distinguishes_large_integers() {
        let a = json!(9_007_199_254_740_993_i64); // 2^53 + 1
        let b = json!(9_007_199_254_740_992_i64); // 2^53
        assert_eq!(json_compare(&a, &b), Ordering::Greater);
    }

    #[test]
    fn filter_incremental_keeps_records_on_type_mismatch() {
        // Regression for #78/#27: a bookmark/key type mismatch must not be
        // silently treated as "not greater" and the record dropped — that is
        // data loss. Keep the record instead.
        let records = vec![json!({"id": 1, "seq": 20_240_701})];
        let start = json!("2024-06-01"); // string bookmark vs numeric key
        let filtered = filter_incremental(records, "seq", &start);
        assert_eq!(filtered.len(), 1, "type mismatch must not silently drop");
    }

    // ── ReplicationBind (#513) ──────────────────────────────────────────────

    fn bind(into: BindTarget, template: &str, format: BindFormat) -> ReplicationBind {
        ReplicationBind {
            into,
            name: "updated_after".to_owned(),
            template: template.to_owned(),
            format,
            advance_from: None,
        }
    }

    #[test]
    fn bind_defaults_template_to_bare_placeholder() {
        let b: ReplicationBind =
            serde_json::from_value(json!({ "name": "since" })).expect("deserializes");
        assert_eq!(b.into, BindTarget::Query);
        assert_eq!(b.template, "${bookmark}");
        assert_eq!(b.format, BindFormat::Raw);
        assert!(b.advance_from.is_none());
    }

    #[test]
    fn bind_render_raw_string_and_number() {
        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Raw);
        assert_eq!(b.render(&json!("2024-06-01")).unwrap(), "2024-06-01");
        assert_eq!(b.render(&json!(150)).unwrap(), "150");
    }

    #[test]
    fn bind_render_applies_operator_template() {
        let b = bind(BindTarget::Query, "gte|${bookmark}", BindFormat::Raw);
        assert_eq!(
            b.render(&json!("2024-06-01T00:00:00Z")).unwrap(),
            "gte|2024-06-01T00:00:00Z"
        );
        // Lucene range form (Bullhorn).
        let l = bind(BindTarget::Query, "[${bookmark} TO *]", BindFormat::Raw);
        assert_eq!(l.render(&json!("20240601")).unwrap(), "[20240601 TO *]");
    }

    #[test]
    fn bind_format_iso8601_from_date_and_epoch() {
        let b = bind(BindTarget::Header, "${bookmark}", BindFormat::Iso8601);
        assert_eq!(
            b.render(&json!("2024-06-01")).unwrap(),
            "2024-06-01T00:00:00Z"
        );
        // Epoch seconds → ISO.
        assert_eq!(
            b.render(&json!(1_717_200_000)).unwrap(),
            "2024-06-01T00:00:00Z"
        );
    }

    #[test]
    fn bind_format_epoch_s_and_ms_from_iso() {
        let s = bind(BindTarget::Query, "${bookmark}", BindFormat::EpochS);
        assert_eq!(
            s.render(&json!("2024-06-01T00:00:00Z")).unwrap(),
            "1717200000"
        );
        let ms = bind(BindTarget::Query, "${bookmark}", BindFormat::EpochMs);
        assert_eq!(
            ms.render(&json!("2024-06-01T00:00:00Z")).unwrap(),
            "1717200000000"
        );
    }

    #[test]
    fn bind_format_date_truncates_datetime() {
        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Date);
        assert_eq!(
            b.render(&json!("2024-06-01T12:34:56Z")).unwrap(),
            "2024-06-01"
        );
    }

    #[test]
    fn bind_format_naive_datetime_assumed_utc() {
        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Iso8601);
        assert_eq!(
            b.render(&json!("2024-06-01T08:00:00")).unwrap(),
            "2024-06-01T08:00:00Z"
        );
    }

    #[test]
    fn bind_format_unparseable_string_errors() {
        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Iso8601);
        assert!(b.render(&json!("not-a-date")).is_err());
    }

    #[test]
    fn bind_format_raw_rejects_composite() {
        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Raw);
        assert!(b.render(&json!({"a": 1})).is_err());
        assert!(b.render(&json!(null)).is_err());
    }

    #[test]
    fn bind_validate_rejects_empty_name_and_missing_placeholder() {
        let mut b = bind(BindTarget::Query, "${bookmark}", BindFormat::Raw);
        b.name = "  ".to_owned();
        assert!(b.validate().is_err());

        let mut b2 = bind(BindTarget::Query, "no placeholder here", BindFormat::Raw);
        b2.name = "since".to_owned();
        assert!(b2.validate().is_err());

        let ok = bind(BindTarget::Query, "gte|${bookmark}", BindFormat::Raw);
        assert!(ok.validate().is_ok());
    }

    #[test]
    fn bind_format_bookmark_bool_raw() {
        assert_eq!(
            format_bookmark(&json!(true), BindFormat::Raw).unwrap(),
            "true"
        );
    }

    #[test]
    fn bind_format_non_scalar_bookmark_errors() {
        // A composite / null bookmark cannot be parsed into an instant.
        assert!(format_bookmark(&json!({"a": 1}), BindFormat::Iso8601).is_err());
        assert!(format_bookmark(&json!(null), BindFormat::EpochS).is_err());
    }

    #[test]
    fn bind_format_out_of_range_epoch_errors() {
        // i64::MAX seconds is far outside chrono's representable range.
        assert!(format_bookmark(&json!(i64::MAX), BindFormat::Iso8601).is_err());
    }
}