faucet_core/
replication.rs

1//! Incremental replication support.
2
3use crate::error::FaucetError;
4use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use std::cmp::Ordering;
9
10/// Determines how records are replicated from the source.
11#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
12#[serde(tag = "type")]
13pub enum ReplicationMethod {
14    /// All records are fetched on every run (default).
15    #[default]
16    FullTable,
17    /// Only records where the `replication_key` field is strictly greater than
18    /// the stored bookmark (`start_replication_value`) are kept.
19    Incremental,
20}
21
22/// What to do with a record whose replication key is missing or `null` (#747).
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
24#[serde(rename_all = "snake_case")]
25pub enum OnMissingKey {
26    /// Keep the record, count it, and warn (default): dropping it would be
27    /// silent data loss.
28    #[default]
29    Keep,
30    /// Drop the record (counted and warned, never silent).
31    Drop,
32    /// Fail the run.
33    Fail,
34}
35
36#[derive(Debug, Clone, PartialEq, Eq)]
37enum KeyForm {
38    TopLevel,
39    DotPath(Vec<String>),
40    Pointer,
41}
42
43/// A compiled replication key: a top-level field name, a dot path
44/// (`fields.updated`, numeric segments index arrays), or an RFC 6901 JSON
45/// Pointer (`/fields/updated`, for field names that contain a dot) (#747).
46///
47/// A dot path first tries the whole string as a literal top-level field, so a
48/// flat column literally named `Account.LastModifiedDate` (a CSV header) still
49/// resolves.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct ReplicationKey {
52    raw: String,
53    form: KeyForm,
54}
55
56impl ReplicationKey {
57    /// Parse a user-facing key (see the type docs for the accepted forms).
58    pub fn parse(raw: &str) -> Result<Self, FaucetError> {
59        if raw.trim().is_empty() {
60            return Err(FaucetError::Config(
61                "replication_key must not be empty".to_owned(),
62            ));
63        }
64        if raw.starts_with('/') {
65            return Ok(Self {
66                raw: raw.to_owned(),
67                form: KeyForm::Pointer,
68            });
69        }
70        if !raw.contains('.') {
71            return Ok(Self::top_level(raw));
72        }
73        let segments: Vec<String> = raw.split('.').map(str::to_owned).collect();
74        if segments.iter().any(String::is_empty) {
75            return Err(FaucetError::Config(format!(
76                "replication_key '{raw}': empty path segment (use the JSON Pointer form \
77                 `/a/b` for field names that contain dots)"
78            )));
79        }
80        Ok(Self {
81            raw: raw.to_owned(),
82            form: KeyForm::DotPath(segments),
83        })
84    }
85
86    /// A literal top-level field name, never interpreted as a path.
87    pub fn top_level(name: &str) -> Self {
88        Self {
89            raw: name.to_owned(),
90            form: KeyForm::TopLevel,
91        }
92    }
93
94    /// The key as configured.
95    pub fn as_str(&self) -> &str {
96        &self.raw
97    }
98
99    /// Whether the key is a JSON Pointer (`/a/b`).
100    pub fn is_pointer(&self) -> bool {
101        self.form == KeyForm::Pointer
102    }
103
104    /// Whether the key addresses a nested value (dot path or pointer).
105    pub fn is_nested(&self) -> bool {
106        self.form != KeyForm::TopLevel
107    }
108
109    /// Resolve the key against one record.
110    pub fn resolve<'a>(&self, record: &'a Value) -> Option<&'a Value> {
111        match &self.form {
112            KeyForm::TopLevel => record.get(&self.raw),
113            KeyForm::Pointer => record.pointer(&self.raw),
114            KeyForm::DotPath(segments) => {
115                if let Some(v) = record.get(&self.raw) {
116                    return Some(v);
117                }
118                segments.iter().try_fold(record, |cur, seg| match cur {
119                    Value::Object(m) => m.get(seg),
120                    Value::Array(a) => seg.parse::<usize>().ok().and_then(|i| a.get(i)),
121                    _ => None,
122                })
123            }
124        }
125    }
126
127    fn resolve_present<'a>(&self, record: &'a Value) -> Option<&'a Value> {
128        self.resolve(record).filter(|v| !v.is_null())
129    }
130}
131
132/// The result of [`filter_incremental_path`].
133#[derive(Debug, Clone, PartialEq, Default)]
134pub struct IncrementalFilter {
135    /// Records to write.
136    pub records: Vec<Value>,
137    /// Records whose key was missing or `null` (kept or dropped per policy).
138    pub missing: usize,
139}
140
141/// Filter `records` to those where `key > start`, using a compiled key.
142///
143/// A record whose key is missing or `null` is handled per `on_missing`
144/// (counted in [`IncrementalFilter::missing`] either way); a record whose key
145/// has a different JSON type than `start` is kept with a warning. Returns
146/// `Err` only for [`OnMissingKey::Fail`].
147pub fn filter_incremental_path(
148    records: Vec<Value>,
149    key: &ReplicationKey,
150    start: &Value,
151    on_missing: OnMissingKey,
152) -> Result<IncrementalFilter, FaucetError> {
153    let mut missing = 0usize;
154    let mut kept = Vec::with_capacity(records.len());
155    for r in records {
156        let keep = match key.resolve_present(&r) {
157            None => {
158                missing += 1;
159                match on_missing {
160                    OnMissingKey::Keep => true,
161                    OnMissingKey::Drop => false,
162                    OnMissingKey::Fail => {
163                        return Err(FaucetError::Source(format!(
164                            "incremental replication: a record lacks replication_key '{}' \
165                             (on_missing_key: fail)",
166                            key.as_str()
167                        )));
168                    }
169                }
170            }
171            Some(v) if !comparable(v, start) => {
172                tracing::warn!(
173                    key = key.as_str(),
174                    "incremental replication: record key type does not match the bookmark \
175                     type; keeping the record to avoid silently dropping data"
176                );
177                true
178            }
179            Some(v) => json_gt(v, start),
180        };
181        if keep {
182            kept.push(r);
183        }
184    }
185    Ok(IncrementalFilter {
186        records: kept,
187        missing,
188    })
189}
190
191/// Filter `records` to only those where `record[key] > start`.
192///
193/// `key` is a literal top-level field. Values compare by what they hold (see
194/// [`json_gt`]): decimal strings numerically, timestamp strings as instants,
195/// integers exactly (no `f64` precision loss), floats as `f64`, other strings
196/// lexicographically.
197///
198/// Records missing the key (or holding `null`) are **kept** and a warning is
199/// logged (#747): dropping them silently is data loss. Likewise a record whose
200/// key value is a *different JSON type* than `start` is kept (#78/#27).
201pub fn filter_incremental(records: Vec<Value>, key: &str, start: &Value) -> Vec<Value> {
202    let out = filter_incremental_path(
203        records,
204        &ReplicationKey::top_level(key),
205        start,
206        OnMissingKey::Keep,
207    )
208    .unwrap_or_default();
209    if out.missing > 0 {
210        tracing::warn!(
211            key,
212            missing = out.missing,
213            "incremental replication: {} record(s) lacked replication_key '{key}'; kept to \
214             avoid silent data loss",
215            out.missing
216        );
217    }
218    out.records
219}
220
221/// Return the maximum non-null value of `key` across all records, if any.
222pub fn max_replication_value_path<'a>(
223    records: &'a [Value],
224    key: &ReplicationKey,
225) -> Option<&'a Value> {
226    records
227        .iter()
228        .filter_map(|r| key.resolve_present(r))
229        .max_by(|a, b| json_compare(a, b))
230}
231
232/// Return the maximum value of `record[key]` across all records, if any.
233pub fn max_replication_value<'a>(records: &'a [Value], key: &str) -> Option<&'a Value> {
234    records
235        .iter()
236        .filter_map(|r| r.get(key))
237        .max_by(|a, b| json_compare(a, b))
238}
239
240/// Return the larger of two replication values using the same ordering as
241/// [`max_replication_value`] (see [`json_gt`]; falling back to `a` on a tie).
242pub fn max_value(a: Value, b: Value) -> Value {
243    match json_compare(&a, &b) {
244        Ordering::Less => b,
245        _ => a,
246    }
247}
248
249/// Type-rank for a total ordering across JSON value kinds, so comparisons of
250/// differing types are deterministic instead of collapsing to `Equal`.
251fn type_rank(v: &Value) -> u8 {
252    match v {
253        Value::Null => 0,
254        Value::Bool(_) => 1,
255        Value::Number(_) => 2,
256        Value::String(_) => 3,
257        Value::Array(_) => 4,
258        Value::Object(_) => 5,
259    }
260}
261
262/// Exact integer view of a JSON number (`i64` or `u64`), widened to `i128` so
263/// both halves of the range compare without `f64` precision loss. `None` for
264/// non-integral (floating) numbers.
265fn number_as_i128(n: &serde_json::Number) -> Option<i128> {
266    n.as_i64()
267        .map(i128::from)
268        .or_else(|| n.as_u64().map(i128::from))
269}
270
271/// Whether two replication values can be ordered against each other: the same
272/// JSON type, or a number and a decimal string (sources such as Oracle emit a
273/// `NUMBER` as a number when it fits `f64` exactly and as a string otherwise).
274fn comparable(a: &Value, b: &Value) -> bool {
275    type_rank(a) == type_rank(b) || numeric_text(a).is_some() && numeric_text(b).is_some()
276}
277
278/// The decimal text of a number or a decimal string.
279fn numeric_text(v: &Value) -> Option<std::borrow::Cow<'_, str>> {
280    match v {
281        Value::Number(n) => Some(std::borrow::Cow::Owned(n.to_string())),
282        Value::String(s) if Decimal::parse(s).is_some() => Some(std::borrow::Cow::Borrowed(s)),
283        _ => None,
284    }
285}
286
287/// An exact decimal: `0.d₁d₂… × 10^exp` with no leading or trailing zero
288/// digits (`digits` empty means zero). Arbitrary length, so a DECIMAL(38)
289/// cursor orders exactly.
290#[derive(Debug, PartialEq, Eq)]
291struct Decimal {
292    negative: bool,
293    digits: Vec<u8>,
294    exp: i64,
295}
296
297impl Decimal {
298    fn parse(text: &str) -> Option<Self> {
299        let s = text.trim();
300        let (negative, s) = match s.as_bytes().first()? {
301            b'-' => (true, &s[1..]),
302            b'+' => (false, &s[1..]),
303            _ => (false, s),
304        };
305        let (mantissa, exponent) = match s.find(['e', 'E']) {
306            Some(i) => (&s[..i], s[i + 1..].parse::<i64>().ok()?),
307            None => (s, 0),
308        };
309        let (int, frac) = match mantissa.split_once('.') {
310            Some((i, f)) => (i, f),
311            None => (mantissa, ""),
312        };
313        if int.is_empty() && frac.is_empty()
314            || !int.bytes().chain(frac.bytes()).all(|b| b.is_ascii_digit())
315        {
316            return None;
317        }
318        let mut digits: Vec<u8> = int.bytes().chain(frac.bytes()).map(|b| b - b'0').collect();
319        let mut point = i64::try_from(int.len()).ok()?;
320        let leading = digits.iter().take_while(|d| **d == 0).count();
321        digits.drain(..leading);
322        point -= i64::try_from(leading).ok()?;
323        while digits.last() == Some(&0) {
324            digits.pop();
325        }
326        let exp = if digits.is_empty() {
327            0
328        } else {
329            point.checked_add(exponent)?
330        };
331        Some(Self {
332            negative: negative && !digits.is_empty(),
333            digits,
334            exp,
335        })
336    }
337
338    fn signum(&self) -> i8 {
339        match (self.digits.is_empty(), self.negative) {
340            (true, _) => 0,
341            (false, true) => -1,
342            (false, false) => 1,
343        }
344    }
345
346    fn cmp_magnitude(&self, other: &Self) -> Ordering {
347        self.exp
348            .cmp(&other.exp)
349            .then_with(|| self.digits.cmp(&other.digits))
350    }
351}
352
353impl Ord for Decimal {
354    fn cmp(&self, other: &Self) -> Ordering {
355        match self.signum().cmp(&other.signum()) {
356            Ordering::Equal => match self.signum() {
357                1 => self.cmp_magnitude(other),
358                -1 => other.cmp_magnitude(self),
359                _ => Ordering::Equal,
360            },
361            unequal => unequal,
362        }
363    }
364}
365
366impl PartialOrd for Decimal {
367    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
368        Some(self.cmp(other))
369    }
370}
371
372/// A timestamp string as an instant: RFC 3339 with an offset (a space may
373/// replace the `T`), compared in UTC; or the same without an offset, compared
374/// as a local date-time against another offset-less value.
375#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
376enum Instant {
377    Zoned(DateTime<Utc>),
378    Naive(chrono::NaiveDateTime),
379}
380
381impl Instant {
382    fn parse(text: &str) -> Option<Self> {
383        let s = text.trim();
384        if s.len() < 19 || !s.as_bytes()[..4].iter().all(u8::is_ascii_digit) {
385            return None;
386        }
387        let normalized = if s.as_bytes()[10] == b' ' {
388            let mut t = s.to_string();
389            t.replace_range(10..11, "T");
390            std::borrow::Cow::Owned(t)
391        } else {
392            std::borrow::Cow::Borrowed(s)
393        };
394        if let Ok(dt) = DateTime::parse_from_rfc3339(&normalized) {
395            return Some(Self::Zoned(dt.with_timezone(&Utc)));
396        }
397        chrono::NaiveDateTime::parse_from_str(&normalized, "%Y-%m-%dT%H:%M:%S%.f")
398            .ok()
399            .map(Self::Naive)
400    }
401}
402
403/// Order two strings by what they hold: both decimals → numerically, both
404/// instants of the same kind → chronologically, otherwise lexicographically.
405fn compare_strings(x: &str, y: &str) -> Ordering {
406    if let (Some(a), Some(b)) = (Decimal::parse(x), Decimal::parse(y)) {
407        return a.cmp(&b);
408    }
409    match (Instant::parse(x), Instant::parse(y)) {
410        (Some(a @ Instant::Zoned(_)), Some(b @ Instant::Zoned(_)))
411        | (Some(a @ Instant::Naive(_)), Some(b @ Instant::Naive(_))) => a.cmp(&b),
412        _ => x.cmp(y),
413    }
414}
415
416/// Total ordering over JSON values used for replication bookmarks.
417///
418/// - Numbers: compared exactly as `i128` when both are integral (so cursors
419///   above 2^53 don't lose precision); otherwise as `f64`, with NaN ordered
420///   last.
421/// - Strings: decimal strings numerically and exactly (`"99" < "100"`),
422///   timestamps chronologically after normalizing offsets and fraction widths,
423///   anything else lexicographically.
424/// - A number against a decimal string: numerically.
425/// - Other same-type values: natural ordering (bools `false < true`, arrays
426///   element-wise, objects by serialized form).
427/// - Different types: ordered by [`type_rank`] so the result is always total.
428pub(crate) fn json_compare(a: &Value, b: &Value) -> Ordering {
429    match (a, b) {
430        (Value::Number(_), Value::String(_)) | (Value::String(_), Value::Number(_)) => {
431            match (numeric_text(a), numeric_text(b)) {
432                (Some(x), Some(y)) => match (Decimal::parse(&x), Decimal::parse(&y)) {
433                    (Some(x), Some(y)) => x.cmp(&y),
434                    _ => type_rank(a).cmp(&type_rank(b)),
435                },
436                _ => type_rank(a).cmp(&type_rank(b)),
437            }
438        }
439        (Value::Number(an), Value::Number(bn)) => {
440            match (number_as_i128(an), number_as_i128(bn)) {
441                (Some(ai), Some(bi)) => ai.cmp(&bi),
442                _ => {
443                    let af = an.as_f64().unwrap_or(f64::NAN);
444                    let bf = bn.as_f64().unwrap_or(f64::NAN);
445                    af.partial_cmp(&bf).unwrap_or_else(|| {
446                        // At least one NaN — order NaN last, deterministically.
447                        match (af.is_nan(), bf.is_nan()) {
448                            (false, true) => Ordering::Less,
449                            (true, false) => Ordering::Greater,
450                            _ => Ordering::Equal,
451                        }
452                    })
453                }
454            }
455        }
456        (Value::String(x), Value::String(y)) => compare_strings(x, y),
457        (Value::Bool(x), Value::Bool(y)) => x.cmp(y),
458        (Value::Null, Value::Null) => Ordering::Equal,
459        (Value::Array(x), Value::Array(y)) => {
460            for (xi, yi) in x.iter().zip(y.iter()) {
461                let c = json_compare(xi, yi);
462                if c != Ordering::Equal {
463                    return c;
464                }
465            }
466            x.len().cmp(&y.len())
467        }
468        // Objects have no natural order; use the serialized form for a stable
469        // total order (objects as replication keys are pathological).
470        (Value::Object(_), Value::Object(_)) => a.to_string().cmp(&b.to_string()),
471        // Different JSON types — order by type rank so comparison is total.
472        _ => type_rank(a).cmp(&type_rank(b)),
473    }
474}
475
476/// Total-order "greater than" over JSON values, using the same comparison
477/// [`filter_incremental`] applies to replication keys (numbers and decimal
478/// strings numerically, timestamp strings chronologically, other strings
479/// lexicographically). Public
480/// so callers bounding a replay window (e.g. `faucet backfill --to-bookmark`)
481/// compare exactly like the incremental filter does.
482pub fn json_gt(a: &Value, b: &Value) -> bool {
483    json_compare(a, b) == Ordering::Greater
484}
485
486// ── Server-side incremental push-down (#513) ─────────────────────────────────
487
488/// The placeholder replaced by the formatted bookmark inside a
489/// [`ReplicationBind::template`].
490pub const BIND_PLACEHOLDER: &str = "${bookmark}";
491
492fn default_bind_template() -> String {
493    BIND_PLACEHOLDER.to_owned()
494}
495
496/// Where a rendered bookmark is injected into the outgoing request.
497#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
498#[serde(rename_all = "snake_case")]
499pub enum BindTarget {
500    /// A query-string parameter (default) — e.g. `?updated_after=…`.
501    #[default]
502    Query,
503    /// A request header — e.g. `If-Modified-Since: …`.
504    Header,
505    /// A field of the JSON request body (POST-search APIs): a top-level field
506    /// named by `name`, or any existing location addressed by a JSON Pointer
507    /// `path` (#748).
508    Body,
509    /// A `{name}` placeholder in the request path.
510    Path,
511}
512
513/// How the bookmark value is formatted before it is substituted into the
514/// [`ReplicationBind::template`].
515///
516/// For every non-[`Raw`](BindFormat::Raw) format the bookmark is first parsed
517/// into an instant: a string is read as RFC 3339, a bare `YYYY-MM-DD` date
518/// (midnight UTC), or a naive `YYYY-MM-DDTHH:MM:SS` (assumed UTC); a JSON
519/// number is read as **epoch seconds**. It is then re-emitted in the target
520/// representation, so `epoch_ms` ← ISO string and `iso8601` ← epoch number both
521/// work. A JSON number under `epoch_s` / `epoch_ms` is taken to be in that
522/// unit already and emitted as its integer value.
523#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
524#[serde(rename_all = "snake_case")]
525pub enum BindFormat {
526    /// Emit the scalar verbatim (string as-is, number as its decimal form).
527    /// The default; no timestamp parsing.
528    #[default]
529    Raw,
530    /// RFC 3339 / ISO-8601 UTC timestamp, e.g. `2024-06-01T00:00:00Z`.
531    Iso8601,
532    /// Unix epoch **seconds** (integer).
533    EpochS,
534    /// Unix epoch **milliseconds** (integer).
535    EpochMs,
536    /// Calendar date `YYYY-MM-DD` (UTC).
537    Date,
538}
539
540/// The JSON type a body-target bind writes (#748).
541#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
542#[serde(rename_all = "snake_case")]
543pub enum BindValueType {
544    /// A JSON string (default).
545    #[default]
546    String,
547    /// A JSON number (e.g. an `epoch_ms` bookmark an API requires unquoted).
548    Number,
549}
550
551impl BindValueType {
552    /// Convert a rendered bind value into the JSON value to write.
553    pub fn to_value(self, rendered: &str) -> Result<Value, FaucetError> {
554        match self {
555            Self::String => Ok(Value::String(rendered.to_owned())),
556            Self::Number => {
557                let n: Option<serde_json::Number> = rendered
558                    .parse::<i64>()
559                    .map(serde_json::Number::from)
560                    .ok()
561                    .or_else(|| rendered.parse::<u64>().ok().map(serde_json::Number::from))
562                    .or_else(|| {
563                        rendered
564                            .parse::<f64>()
565                            .ok()
566                            .and_then(serde_json::Number::from_f64)
567                    });
568                n.map(Value::Number).ok_or_else(|| {
569                    FaucetError::Source(format!(
570                        "bind: rendered value '{rendered}' is not a number (value_type: number)"
571                    ))
572                })
573            }
574        }
575    }
576}
577
578fn unescape_pointer_token(token: &str) -> String {
579    token.replace("~1", "/").replace("~0", "~")
580}
581
582/// Write `value` into `body` at the RFC 6901 JSON Pointer `pointer` (#748).
583///
584/// The pointer must address an existing scalar (or `null`) — or a missing
585/// final key whose parent is an existing object. Intermediate objects and
586/// array elements are never created, and an object/array target is refused
587/// (a bind replaces a value; it does not merge).
588pub fn set_body_pointer(body: &mut Value, pointer: &str, value: Value) -> Result<(), FaucetError> {
589    if !pointer.starts_with('/') {
590        return Err(FaucetError::Config(format!(
591            "JSON Pointer '{pointer}' must start with '/'"
592        )));
593    }
594    if let Some(slot) = body.pointer_mut(pointer) {
595        if slot.is_object() || slot.is_array() {
596            return Err(FaucetError::Source(format!(
597                "request body location '{pointer}' holds an object or array; a bind replaces \
598                 a scalar value"
599            )));
600        }
601        *slot = value;
602        return Ok(());
603    }
604    let cut = pointer.rfind('/').unwrap_or(0);
605    let (parent, leaf) = (&pointer[..cut], &pointer[cut + 1..]);
606    let parent_value = if parent.is_empty() {
607        Some(body)
608    } else {
609        body.pointer_mut(parent)
610    };
611    match parent_value {
612        Some(Value::Object(map)) => {
613            map.insert(unescape_pointer_token(leaf), value);
614            Ok(())
615        }
616        _ => Err(FaucetError::Source(format!(
617            "request body has no location '{pointer}' (the pointer must resolve to an \
618             existing value, or to a new key of an existing object)"
619        ))),
620    }
621}
622
623/// Load-time check of a bind's placement: a body bind needs exactly one of
624/// `name` / `path`; every other target needs `name` and refuses `path`.
625pub(crate) fn validate_bind_placement(
626    what: &str,
627    into: BindTarget,
628    name: &str,
629    path: Option<&str>,
630) -> Result<(), FaucetError> {
631    let has_name = !name.trim().is_empty();
632    match (into, path) {
633        (BindTarget::Body, Some(p)) => {
634            if has_name {
635                return Err(FaucetError::Config(format!(
636                    "{what}: set either `name` (top-level body field) or `path` (JSON Pointer), \
637                     not both"
638                )));
639            }
640            if !p.starts_with('/') || p.len() < 2 {
641                return Err(FaucetError::Config(format!(
642                    "{what}: `path` must be a JSON Pointer such as `/filters/0/value`, got '{p}'"
643                )));
644            }
645            Ok(())
646        }
647        (_, Some(_)) => Err(FaucetError::Config(format!(
648            "{what}: `path` applies only to `into: body`"
649        ))),
650        (BindTarget::Body, None) if !has_name => Err(FaucetError::Config(format!(
651            "{what}: `into: body` needs `name` (top-level field) or `path` (JSON Pointer)"
652        ))),
653        (_, None) if !has_name => Err(FaucetError::Config(format!(
654            "{what}: `name` must not be empty"
655        ))),
656        _ => Ok(()),
657    }
658}
659
660/// Declarative binding of the stored bookmark into the **outgoing request** —
661/// "server-side incremental push-down" (#513).
662///
663/// Today faucet tracks bookmarks and filters incrementally *client-side* (after
664/// download). A bind lets a source instead push the bookmark into the request
665/// (query param / header / body field / path) so the server returns only the
666/// new rows. The existing client-side [`filter_incremental`] stays active as a
667/// safety net for servers that don't honour the filter exactly.
668#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
669#[serde(deny_unknown_fields)]
670pub struct ReplicationBind {
671    /// Where to place the rendered value.
672    #[serde(default)]
673    pub into: BindTarget,
674    /// The parameter / header / body-field / path-placeholder name. Optional
675    /// only for `into: body` with a `path`.
676    #[serde(default)]
677    pub name: String,
678    /// `into: body` only: an RFC 6901 JSON Pointer into the configured `body`
679    /// (`/filterGroups/0/filters/0/value`) instead of a top-level `name` (#748).
680    /// It must resolve to an existing scalar or to a new key of an existing
681    /// object; array elements are never created.
682    #[serde(default, skip_serializing_if = "Option::is_none")]
683    pub path: Option<String>,
684    /// JSON type written by a body bind: `string` (default) or `number`.
685    #[serde(default)]
686    pub value_type: BindValueType,
687    /// Template rendered with [`BIND_PLACEHOLDER`] (`${bookmark}`) replaced by
688    /// the formatted bookmark. Defaults to the bare `${bookmark}`; set e.g.
689    /// `"gte|${bookmark}"` (an operator-prefixed filter) or `"[${bookmark} TO *]"` (Lucene).
690    #[serde(default = "default_bind_template")]
691    pub template: String,
692    /// How to format the bookmark before substitution.
693    #[serde(default)]
694    pub format: BindFormat,
695    /// Optional JSONPath into the response body to advance the bookmark from,
696    /// instead of `max(record[replication_key])`.
697    #[serde(default, skip_serializing_if = "Option::is_none")]
698    pub advance_from: Option<String>,
699}
700
701impl ReplicationBind {
702    /// Validate the binding at config-load time.
703    pub fn validate(&self) -> Result<(), FaucetError> {
704        validate_bind_placement(
705            "replication bind",
706            self.into,
707            &self.name,
708            self.path.as_deref(),
709        )?;
710        if !self.template.contains(BIND_PLACEHOLDER) {
711            return Err(FaucetError::Config(format!(
712                "replication bind: `template` must contain the `{BIND_PLACEHOLDER}` placeholder"
713            )));
714        }
715        Ok(())
716    }
717
718    /// Render the binding for a concrete bookmark: format the value, then
719    /// substitute it into the template.
720    pub fn render(&self, bookmark: &Value) -> Result<String, FaucetError> {
721        let formatted = format_bookmark(bookmark, self.format)?;
722        Ok(self.template.replace(BIND_PLACEHOLDER, &formatted))
723    }
724}
725
726/// Parse a bookmark value into a UTC instant (see [`BindFormat`] for the rules).
727fn bookmark_instant(value: &Value) -> Result<DateTime<Utc>, FaucetError> {
728    match value {
729        Value::String(s) => {
730            let s = s.trim();
731            if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
732                return Ok(dt.with_timezone(&Utc));
733            }
734            if let Ok(d) = NaiveDate::parse_from_str(s, "%Y-%m-%d")
735                && let Some(ndt) = d.and_hms_opt(0, 0, 0)
736            {
737                return Ok(DateTime::<Utc>::from_naive_utc_and_offset(ndt, Utc));
738            }
739            if let Ok(ndt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S") {
740                return Ok(DateTime::<Utc>::from_naive_utc_and_offset(ndt, Utc));
741            }
742            Err(FaucetError::Config(format!(
743                "replication bind: cannot parse bookmark '{s}' as a timestamp \
744                 (expected RFC 3339, YYYY-MM-DD, or YYYY-MM-DDTHH:MM:SS)"
745            )))
746        }
747        Value::Number(n) => {
748            let secs = n.as_i64().or_else(|| n.as_f64().map(|f| f as i64));
749            secs.and_then(|s| DateTime::<Utc>::from_timestamp(s, 0))
750                .ok_or_else(|| {
751                    FaucetError::Config(format!(
752                        "replication bind: numeric bookmark {n} is out of range for epoch seconds"
753                    ))
754                })
755        }
756        other => Err(FaucetError::Config(format!(
757            "replication bind: bookmark must be a string or number, got {other}"
758        ))),
759    }
760}
761
762/// Parse a bookmark value into a UTC instant (public wrapper over the internal
763/// parser; see [`BindFormat`] for the accepted forms). Used by datetime window
764/// slicing (#527) to resolve the sweep's start bound from the stored bookmark.
765pub fn parse_instant(value: &Value) -> Result<DateTime<Utc>, FaucetError> {
766    bookmark_instant(value)
767}
768
769/// Format an already-resolved UTC instant per [`BindFormat`]. Unlike
770/// [`format_bookmark`] (which takes an arbitrary scalar and, for [`BindFormat::Raw`],
771/// echoes it verbatim), this always has a real instant, so `Raw` and `Iso8601`
772/// both emit an RFC 3339 UTC timestamp. Used to render window boundaries (#527).
773pub fn format_instant(dt: DateTime<Utc>, format: BindFormat) -> String {
774    match format {
775        BindFormat::Raw | BindFormat::Iso8601 => {
776            dt.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
777        }
778        BindFormat::Date => dt.format("%Y-%m-%d").to_string(),
779        BindFormat::EpochS => dt.timestamp().to_string(),
780        BindFormat::EpochMs => dt.timestamp_millis().to_string(),
781    }
782}
783
784/// Format a bookmark value per [`BindFormat`].
785pub fn format_bookmark(value: &Value, format: BindFormat) -> Result<String, FaucetError> {
786    match format {
787        BindFormat::Raw => match value {
788            Value::String(s) => Ok(s.clone()),
789            Value::Number(n) => Ok(n.to_string()),
790            Value::Bool(b) => Ok(b.to_string()),
791            other => Err(FaucetError::Config(format!(
792                "replication bind: cannot render {other} as a raw scalar"
793            ))),
794        },
795        BindFormat::Iso8601 => {
796            Ok(bookmark_instant(value)?.to_rfc3339_opts(chrono::SecondsFormat::Secs, true))
797        }
798        BindFormat::Date => Ok(bookmark_instant(value)?.format("%Y-%m-%d").to_string()),
799        BindFormat::EpochS | BindFormat::EpochMs if value.is_number() => Ok(epoch_number(value)),
800        BindFormat::EpochS => Ok(bookmark_instant(value)?.timestamp().to_string()),
801        BindFormat::EpochMs => Ok(bookmark_instant(value)?.timestamp_millis().to_string()),
802    }
803}
804
805/// A numeric bookmark under an epoch bind is already in the bind's unit (an
806/// `epoch_ms` cursor stores milliseconds), so it is emitted as the integer it
807/// holds rather than re-read as seconds and scaled.
808fn epoch_number(value: &Value) -> String {
809    match value.as_i64() {
810        Some(i) => i.to_string(),
811        None => match value.as_u64() {
812            Some(u) => u.to_string(),
813            None => (value.as_f64().unwrap_or(0.0).trunc() as i64).to_string(),
814        },
815    }
816}
817
818#[cfg(test)]
819mod tests {
820    use super::*;
821    use serde_json::json;
822
823    #[test]
824    fn test_filter_incremental_strings() {
825        let records = vec![
826            json!({"id": 1, "updated_at": "2024-01-01"}),
827            json!({"id": 2, "updated_at": "2024-06-01"}),
828            json!({"id": 3, "updated_at": "2024-12-01"}),
829        ];
830        let start = json!("2024-06-01");
831        let filtered = filter_incremental(records, "updated_at", &start);
832        assert_eq!(filtered.len(), 1);
833        assert_eq!(filtered[0]["id"], 3);
834    }
835
836    #[test]
837    fn test_filter_incremental_numbers() {
838        let records = vec![
839            json!({"id": 1, "seq": 100}),
840            json!({"id": 2, "seq": 200}),
841            json!({"id": 3, "seq": 300}),
842        ];
843        let start = json!(150);
844        let filtered = filter_incremental(records, "seq", &start);
845        assert_eq!(filtered.len(), 2);
846        assert_eq!(filtered[0]["id"], 2);
847        assert_eq!(filtered[1]["id"], 3);
848    }
849
850    #[test]
851    fn test_filter_incremental_missing_key_kept() {
852        // #747: a record without the key used to be dropped silently.
853        let records = vec![
854            json!({"id": 1}),
855            json!({"id": 2, "updated_at": "2024-12-01"}),
856            json!({"id": 3, "updated_at": null}),
857        ];
858        let start = json!("2024-01-01");
859        let filtered = filter_incremental(records, "updated_at", &start);
860        assert_eq!(filtered.len(), 3);
861    }
862
863    #[test]
864    fn replication_key_parse_forms() {
865        assert!(!ReplicationKey::parse("updated").unwrap().is_nested());
866        let dot = ReplicationKey::parse("fields.updated").unwrap();
867        assert!(dot.is_nested() && !dot.is_pointer());
868        assert_eq!(dot.as_str(), "fields.updated");
869        let ptr = ReplicationKey::parse("/a.b/c").unwrap();
870        assert!(ptr.is_pointer() && ptr.is_nested());
871        assert!(ReplicationKey::parse(" ").is_err());
872        assert!(ReplicationKey::parse("a..b").is_err());
873        assert!(ReplicationKey::parse(".a").is_err());
874    }
875
876    #[test]
877    fn replication_key_resolves_nested_array_and_pointer() {
878        let r = json!({
879            "fields": {"updated": "2024-06-01"},
880            "items": [{"date": 1}, {"date": 2}],
881            "a.b": {"c": 7},
882            "x": 5
883        });
884        let k = |s: &str| ReplicationKey::parse(s).unwrap();
885        assert_eq!(k("fields.updated").resolve(&r), Some(&json!("2024-06-01")));
886        assert_eq!(k("items.1.date").resolve(&r), Some(&json!(2)));
887        assert_eq!(k("items.x.date").resolve(&r), None);
888        assert_eq!(k("items.9.date").resolve(&r), None);
889        assert_eq!(k("x.y").resolve(&r), None);
890        assert_eq!(k("/a.b/c").resolve(&r), Some(&json!(7)));
891        assert_eq!(k("x").resolve(&r), Some(&json!(5)));
892        let flat = json!({"Account.LastModifiedDate": "2024"});
893        assert_eq!(
894            k("Account.LastModifiedDate").resolve(&flat),
895            Some(&json!("2024"))
896        );
897        assert_eq!(
898            ReplicationKey::top_level("a.b").resolve(&r),
899            Some(&json!({"c": 7}))
900        );
901    }
902
903    #[test]
904    fn filter_incremental_path_nested_and_policies() {
905        let records = || {
906            vec![
907                json!({"id": 1, "fields": {"updated": "2024-01-01"}}),
908                json!({"id": 2, "fields": {"updated": "2024-12-01"}}),
909                json!({"id": 3, "fields": {}}),
910                json!({"id": 4, "fields": {"updated": 5}}),
911            ]
912        };
913        let key = ReplicationKey::parse("fields.updated").unwrap();
914        let start = json!("2024-06-01");
915        let keep = filter_incremental_path(records(), &key, &start, OnMissingKey::Keep).unwrap();
916        let ids: Vec<i64> = keep
917            .records
918            .iter()
919            .map(|r| r["id"].as_i64().unwrap())
920            .collect();
921        assert_eq!(ids, vec![2, 3, 4]);
922        assert_eq!(keep.missing, 1);
923        let drop = filter_incremental_path(records(), &key, &start, OnMissingKey::Drop).unwrap();
924        assert_eq!(drop.records.len(), 2);
925        assert_eq!(drop.missing, 1);
926        let err = filter_incremental_path(records(), &key, &start, OnMissingKey::Fail).unwrap_err();
927        assert!(err.to_string().contains("fields.updated"), "{err}");
928    }
929
930    #[test]
931    fn max_replication_value_path_skips_missing_and_null() {
932        let key = ReplicationKey::parse("fields.updated").unwrap();
933        let records = vec![
934            json!({"fields": {"updated": "2024-01-01"}}),
935            json!({"fields": {"updated": null}}),
936            json!({"fields": {"updated": "2024-12-01"}}),
937            json!({}),
938        ];
939        assert_eq!(
940            max_replication_value_path(&records, &key),
941            Some(&json!("2024-12-01"))
942        );
943        assert!(max_replication_value_path(&records[1..2], &key).is_none());
944    }
945
946    #[test]
947    fn on_missing_key_serde() {
948        assert_eq!(OnMissingKey::default(), OnMissingKey::Keep);
949        let v: OnMissingKey = serde_json::from_value(json!("fail")).unwrap();
950        assert_eq!(v, OnMissingKey::Fail);
951    }
952
953    #[test]
954    fn test_filter_incremental_equal_excluded() {
955        let records = vec![
956            json!({"id": 1, "updated_at": "2024-06-01"}),
957            json!({"id": 2, "updated_at": "2024-06-02"}),
958        ];
959        let start = json!("2024-06-01");
960        let filtered = filter_incremental(records, "updated_at", &start);
961        assert_eq!(filtered.len(), 1);
962        assert_eq!(filtered[0]["id"], 2);
963    }
964
965    #[test]
966    fn test_max_replication_value_strings() {
967        let records = vec![
968            json!({"updated_at": "2024-01-01"}),
969            json!({"updated_at": "2024-12-01"}),
970            json!({"updated_at": "2024-06-01"}),
971        ];
972        let max = max_replication_value(&records, "updated_at").unwrap();
973        assert_eq!(max, &json!("2024-12-01"));
974    }
975
976    #[test]
977    fn test_max_replication_value_numbers() {
978        let records = vec![json!({"seq": 5}), json!({"seq": 10}), json!({"seq": 3})];
979        let max = max_replication_value(&records, "seq").unwrap();
980        assert_eq!(max, &json!(10));
981    }
982
983    #[test]
984    fn test_max_replication_value_empty() {
985        let records: Vec<Value> = vec![];
986        assert!(max_replication_value(&records, "updated_at").is_none());
987    }
988
989    #[test]
990    fn test_max_value_picks_larger_string() {
991        assert_eq!(
992            max_value(json!("2024-01-01"), json!("2024-06-01")),
993            json!("2024-06-01")
994        );
995    }
996
997    #[test]
998    fn test_max_value_picks_larger_number() {
999        assert_eq!(max_value(json!(5), json!(10)), json!(10));
1000    }
1001
1002    #[test]
1003    fn test_max_value_returns_a_on_type_mismatch() {
1004        // String outranks Number in the total type-rank ordering, so the
1005        // larger (a) is returned.
1006        assert_eq!(max_value(json!("string"), json!(5)), json!("string"));
1007    }
1008
1009    #[test]
1010    fn filter_incremental_keeps_large_integer_beyond_f64_precision() {
1011        // Regression for #78/#27: integer cursors above 2^53 lose precision
1012        // when compared as f64, so a genuinely-greater value compared Equal
1013        // and was silently dropped.
1014        let two_pow_53 = 9_007_199_254_740_992_i64; // 2^53
1015        let records = vec![
1016            json!({"id": 1, "seq": two_pow_53 + 1}),
1017            json!({"id": 2, "seq": two_pow_53 + 2}),
1018        ];
1019        let start = json!(two_pow_53);
1020        let filtered = filter_incremental(records, "seq", &start);
1021        assert_eq!(
1022            filtered.len(),
1023            2,
1024            "both values are strictly greater than 2^53"
1025        );
1026    }
1027
1028    #[test]
1029    fn json_compare_distinguishes_large_integers() {
1030        let a = json!(9_007_199_254_740_993_i64); // 2^53 + 1
1031        let b = json!(9_007_199_254_740_992_i64); // 2^53
1032        assert_eq!(json_compare(&a, &b), Ordering::Greater);
1033    }
1034
1035    #[test]
1036    fn filter_incremental_keeps_records_on_type_mismatch() {
1037        // Regression for #78/#27: a bookmark/key type mismatch must not be
1038        // silently treated as "not greater" and the record dropped — that is
1039        // data loss. Keep the record instead.
1040        let records = vec![json!({"id": 1, "seq": 20_240_701})];
1041        let start = json!("2024-06-01"); // string bookmark vs numeric key
1042        let filtered = filter_incremental(records, "seq", &start);
1043        assert_eq!(filtered.len(), 1, "type mismatch must not silently drop");
1044    }
1045
1046    fn cmp(a: Value, b: Value) -> Ordering {
1047        json_compare(&a, &b)
1048    }
1049
1050    #[test]
1051    fn decimal_strings_order_numerically_and_exactly() {
1052        use Ordering::*;
1053        let cases = [
1054            ("99", "100", Less),
1055            ("100", "99", Greater),
1056            ("989", "99", Greater),
1057            ("007", "7", Equal),
1058            ("1.50", "1.5", Equal),
1059            ("0.000", "-0", Equal),
1060            ("-1", "0", Less),
1061            ("-10", "-9", Less),
1062            ("-0.5", "-0.25", Less),
1063            ("1e3", "999", Greater),
1064            ("1E-2", "0.01", Equal),
1065            ("+5", "5", Equal),
1066            (".5", "0.4", Greater),
1067            (
1068                "12345678901234567890123456789012345678",
1069                "12345678901234567890123456789012345677",
1070                Greater,
1071            ),
1072            (
1073                "0.000000000000000000000000000000000002",
1074                "0.000000000000000000000000000000000001",
1075                Greater,
1076            ),
1077        ];
1078        for (a, b, want) in cases {
1079            assert_eq!(cmp(json!(a), json!(b)), want, "{a} vs {b}");
1080        }
1081    }
1082
1083    #[test]
1084    fn decimal_parse_rejects_what_is_not_a_number() {
1085        for text in [
1086            "",
1087            "-",
1088            ".",
1089            "1.2.3",
1090            "1e",
1091            "12a",
1092            "2024-06-01",
1093            "e5",
1094            "1e9999999999999999999",
1095        ] {
1096            assert!(Decimal::parse(text).is_none(), "{text:?}");
1097        }
1098    }
1099
1100    #[test]
1101    fn a_decimal_string_bookmark_keeps_larger_rows() {
1102        let records: Vec<Value> = ["100", "989", "50", "99"]
1103            .iter()
1104            .map(|v| json!({ "k": v }))
1105            .collect();
1106        let kept = filter_incremental(records, "k", &json!("99"));
1107        let ks: Vec<&str> = kept.iter().map(|r| r["k"].as_str().unwrap()).collect();
1108        assert_eq!(ks, ["100", "989"]);
1109    }
1110
1111    #[test]
1112    fn numbers_and_decimal_strings_compare_numerically() {
1113        assert_eq!(cmp(json!(100), json!("99")), Ordering::Greater);
1114        assert_eq!(
1115            cmp(json!("123456789012345678901234567890"), json!(5)),
1116            Ordering::Greater
1117        );
1118        assert_eq!(cmp(json!(2.5), json!("2.50")), Ordering::Equal);
1119        let records = vec![
1120            json!({"k": 7}),
1121            json!({"k": 123456789012345678901234567890_f64}),
1122        ];
1123        let kept = filter_incremental(records, "k", &json!("6.5"));
1124        assert_eq!(kept.len(), 2);
1125        assert_eq!(
1126            max_value(json!("12345678901234567890"), json!(12)),
1127            json!("12345678901234567890")
1128        );
1129        assert_eq!(max_value(json!("12"), json!(13)), json!(13));
1130    }
1131
1132    #[test]
1133    fn timestamps_order_as_instants_across_offsets_and_fraction_widths() {
1134        use Ordering::*;
1135        let cases = [
1136            ("2024-01-01T10:00:00+05:00", "2024-01-01T06:00:00Z", Less),
1137            ("2024-01-01T10:00:00.5Z", "2024-01-01T10:00:00.45Z", Greater),
1138            ("2024-01-01T10:00:00Z", "2024-01-01T10:00:00.000Z", Equal),
1139            ("2024-01-01 10:00:00+00:00", "2024-01-01T09:59:59Z", Greater),
1140            ("2024-01-01T10:00:00.5", "2024-01-01T10:00:00.45", Greater),
1141            ("2024-01-01 10:00:00", "2024-01-01T09:00:00", Greater),
1142        ];
1143        for (a, b, want) in cases {
1144            assert_eq!(cmp(json!(a), json!(b)), want, "{a} vs {b}");
1145        }
1146        let records = vec![
1147            json!({"t": "2024-01-01T08:00:00+01:00"}),
1148            json!({"t": "2024-01-01T08:00:00-01:00"}),
1149        ];
1150        let kept = filter_incremental(records, "t", &json!("2024-01-01T08:00:00Z"));
1151        assert_eq!(kept, vec![json!({"t": "2024-01-01T08:00:00-01:00"})]);
1152    }
1153
1154    #[test]
1155    fn mixed_or_unparseable_strings_stay_lexicographic() {
1156        assert_eq!(
1157            cmp(json!("2024-01-01T10:00:00Z"), json!("2024-01-01T10:00:00")),
1158            "2024-01-01T10:00:00Z".cmp("2024-01-01T10:00:00")
1159        );
1160        assert_eq!(cmp(json!("b"), json!("a")), Ordering::Greater);
1161        assert_eq!(
1162            cmp(json!("2024-13-99T99:99:99"), json!("2024")),
1163            Ordering::Greater
1164        );
1165        assert_eq!(cmp(json!("abc"), json!(5)), Ordering::Greater);
1166        assert_eq!(cmp(json!(true), json!("1")), Ordering::Less);
1167    }
1168
1169    // ── ReplicationBind (#513) ──────────────────────────────────────────────
1170
1171    fn bind(into: BindTarget, template: &str, format: BindFormat) -> ReplicationBind {
1172        ReplicationBind {
1173            into,
1174            name: "updated_after".to_owned(),
1175            template: template.to_owned(),
1176            format,
1177            advance_from: None,
1178            path: None,
1179            value_type: BindValueType::String,
1180        }
1181    }
1182
1183    #[test]
1184    fn bind_defaults_template_to_bare_placeholder() {
1185        let b: ReplicationBind =
1186            serde_json::from_value(json!({ "name": "since" })).expect("deserializes");
1187        assert_eq!(b.into, BindTarget::Query);
1188        assert_eq!(b.template, "${bookmark}");
1189        assert_eq!(b.format, BindFormat::Raw);
1190        assert!(b.advance_from.is_none());
1191    }
1192
1193    #[test]
1194    fn bind_render_raw_string_and_number() {
1195        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Raw);
1196        assert_eq!(b.render(&json!("2024-06-01")).unwrap(), "2024-06-01");
1197        assert_eq!(b.render(&json!(150)).unwrap(), "150");
1198    }
1199
1200    #[test]
1201    fn bind_render_applies_operator_template() {
1202        let b = bind(BindTarget::Query, "gte|${bookmark}", BindFormat::Raw);
1203        assert_eq!(
1204            b.render(&json!("2024-06-01T00:00:00Z")).unwrap(),
1205            "gte|2024-06-01T00:00:00Z"
1206        );
1207        // Lucene range form.
1208        let l = bind(BindTarget::Query, "[${bookmark} TO *]", BindFormat::Raw);
1209        assert_eq!(l.render(&json!("20240601")).unwrap(), "[20240601 TO *]");
1210    }
1211
1212    #[test]
1213    fn bind_format_iso8601_from_date_and_epoch() {
1214        let b = bind(BindTarget::Header, "${bookmark}", BindFormat::Iso8601);
1215        assert_eq!(
1216            b.render(&json!("2024-06-01")).unwrap(),
1217            "2024-06-01T00:00:00Z"
1218        );
1219        // Epoch seconds → ISO.
1220        assert_eq!(
1221            b.render(&json!(1_717_200_000)).unwrap(),
1222            "2024-06-01T00:00:00Z"
1223        );
1224    }
1225
1226    #[test]
1227    fn bind_format_epoch_s_and_ms_from_iso() {
1228        let s = bind(BindTarget::Query, "${bookmark}", BindFormat::EpochS);
1229        assert_eq!(
1230            s.render(&json!("2024-06-01T00:00:00Z")).unwrap(),
1231            "1717200000"
1232        );
1233        let ms = bind(BindTarget::Query, "${bookmark}", BindFormat::EpochMs);
1234        assert_eq!(
1235            ms.render(&json!("2024-06-01T00:00:00Z")).unwrap(),
1236            "1717200000000"
1237        );
1238    }
1239
1240    #[test]
1241    fn bind_format_date_truncates_datetime() {
1242        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Date);
1243        assert_eq!(
1244            b.render(&json!("2024-06-01T12:34:56Z")).unwrap(),
1245            "2024-06-01"
1246        );
1247    }
1248
1249    #[test]
1250    fn bind_format_naive_datetime_assumed_utc() {
1251        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Iso8601);
1252        assert_eq!(
1253            b.render(&json!("2024-06-01T08:00:00")).unwrap(),
1254            "2024-06-01T08:00:00Z"
1255        );
1256    }
1257
1258    #[test]
1259    fn bind_format_unparseable_string_errors() {
1260        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Iso8601);
1261        assert!(b.render(&json!("not-a-date")).is_err());
1262    }
1263
1264    #[test]
1265    fn bind_format_raw_rejects_composite() {
1266        let b = bind(BindTarget::Query, "${bookmark}", BindFormat::Raw);
1267        assert!(b.render(&json!({"a": 1})).is_err());
1268        assert!(b.render(&json!(null)).is_err());
1269    }
1270
1271    #[test]
1272    fn bind_validate_rejects_empty_name_and_missing_placeholder() {
1273        let mut b = bind(BindTarget::Query, "${bookmark}", BindFormat::Raw);
1274        b.name = "  ".to_owned();
1275        assert!(b.validate().is_err());
1276
1277        let mut b2 = bind(BindTarget::Query, "no placeholder here", BindFormat::Raw);
1278        b2.name = "since".to_owned();
1279        assert!(b2.validate().is_err());
1280
1281        let ok = bind(BindTarget::Query, "gte|${bookmark}", BindFormat::Raw);
1282        assert!(ok.validate().is_ok());
1283    }
1284
1285    #[test]
1286    fn bind_placement_validation() {
1287        let mut b = bind(BindTarget::Body, "${bookmark}", BindFormat::Raw);
1288        assert!(b.validate().is_ok());
1289        b.path = Some("/a/0/b".into());
1290        assert!(b.validate().unwrap_err().to_string().contains("not both"));
1291        b.name.clear();
1292        assert!(b.validate().is_ok());
1293        b.path = Some("a".into());
1294        assert!(b.validate().is_err());
1295        b.path = Some("/".into());
1296        assert!(b.validate().is_err());
1297        b.path = None;
1298        assert!(
1299            b.validate()
1300                .unwrap_err()
1301                .to_string()
1302                .contains("needs `name`")
1303        );
1304        let mut q = bind(BindTarget::Query, "${bookmark}", BindFormat::Raw);
1305        q.path = Some("/a".into());
1306        assert!(
1307            q.validate()
1308                .unwrap_err()
1309                .to_string()
1310                .contains("only to `into: body`")
1311        );
1312        let parsed: ReplicationBind = serde_json::from_value(json!({
1313            "into": "body", "path": "/f/0/v", "value_type": "number"
1314        }))
1315        .unwrap();
1316        assert!(parsed.validate().is_ok());
1317        assert_eq!(parsed.value_type, BindValueType::Number);
1318    }
1319
1320    #[test]
1321    fn value_type_conversion() {
1322        assert_eq!(BindValueType::String.to_value("5").unwrap(), json!("5"));
1323        assert_eq!(BindValueType::Number.to_value("5").unwrap(), json!(5));
1324        assert_eq!(
1325            BindValueType::Number
1326                .to_value("18446744073709551615")
1327                .unwrap(),
1328            json!(18_446_744_073_709_551_615_u64)
1329        );
1330        assert_eq!(BindValueType::Number.to_value("1.5").unwrap(), json!(1.5));
1331        assert!(BindValueType::Number.to_value("x").is_err());
1332    }
1333
1334    #[test]
1335    fn set_body_pointer_rules() {
1336        let mut body = json!({"filterGroups": [{"filters": [{"value": null}]}], "v": {}, "a/b": 1});
1337        set_body_pointer(&mut body, "/filterGroups/0/filters/0/value", json!("x")).unwrap();
1338        assert_eq!(body["filterGroups"][0]["filters"][0]["value"], json!("x"));
1339        set_body_pointer(&mut body, "/v/after", json!("c")).unwrap();
1340        assert_eq!(body["v"]["after"], json!("c"));
1341        set_body_pointer(&mut body, "/top", json!(1)).unwrap();
1342        assert_eq!(body["top"], json!(1));
1343        set_body_pointer(&mut body, "/a~1b", json!(2)).unwrap();
1344        assert_eq!(body["a/b"], json!(2));
1345        set_body_pointer(&mut body, "/v/x~1y~0z", json!(3)).unwrap();
1346        assert_eq!(body["v"]["x/y~z"], json!(3));
1347        assert!(set_body_pointer(&mut body, "/filterGroups/1/filters", json!(1)).is_err());
1348        assert!(set_body_pointer(&mut body, "/missing/leaf", json!(1)).is_err());
1349        assert!(set_body_pointer(&mut body, "/v", json!(1)).is_err());
1350        assert!(set_body_pointer(&mut body, "/filterGroups/0/filters/5", json!(1)).is_err());
1351        assert!(set_body_pointer(&mut body, "nope", json!(1)).is_err());
1352    }
1353
1354    #[test]
1355    fn bind_format_bookmark_bool_raw() {
1356        assert_eq!(
1357            format_bookmark(&json!(true), BindFormat::Raw).unwrap(),
1358            "true"
1359        );
1360    }
1361
1362    #[test]
1363    fn bind_format_non_scalar_bookmark_errors() {
1364        // A composite / null bookmark cannot be parsed into an instant.
1365        assert!(format_bookmark(&json!({"a": 1}), BindFormat::Iso8601).is_err());
1366        assert!(format_bookmark(&json!(null), BindFormat::EpochS).is_err());
1367    }
1368
1369    #[test]
1370    fn numeric_bookmarks_keep_the_epoch_binds_unit() {
1371        assert_eq!(
1372            format_bookmark(&json!(1_717_200_000_000_i64), BindFormat::EpochMs).unwrap(),
1373            "1717200000000"
1374        );
1375        assert_eq!(
1376            format_bookmark(&json!(1_717_200_000), BindFormat::EpochS).unwrap(),
1377            "1717200000"
1378        );
1379        assert_eq!(
1380            format_bookmark(&json!(u64::MAX), BindFormat::EpochMs).unwrap(),
1381            u64::MAX.to_string()
1382        );
1383        assert_eq!(
1384            format_bookmark(&json!(1_717_200_000.9), BindFormat::EpochS).unwrap(),
1385            "1717200000"
1386        );
1387    }
1388
1389    #[test]
1390    fn bind_format_out_of_range_epoch_errors() {
1391        // i64::MAX seconds is far outside chrono's representable range.
1392        assert!(format_bookmark(&json!(i64::MAX), BindFormat::Iso8601).is_err());
1393    }
1394}
faucet_core/replication.rs

faucet_core/
replication.rs
