kglite-bolt-server 0.19.7

//! `kglite::api::Value` ↔ `boltr::types::BoltValue` adapter.
//!
//! `to_bolt` handles ALL outbound variants (scalars, collections,
//! temporal/spatial, Node, Relationship, Path). `Value::NodeRef` and malformed
//! paths return errors: internal placeholders and invalid graph structure must
//! not reach the boundary. `from_bolt` rejects inbound Node/Rel/Path (drivers
//! don't pass those as parameters), inbound time-of-day variants
//! (kglite has no standalone time-of-day variant), inbound Bytes (no Value
//! variant), and inbound Point3D (kglite has only Point2D).
//!
//! # Mapping table (the implementer's spec)
//!
//! | kglite `Value`                          | `boltr::BoltValue`            | Notes                                              |
//! |-----------------------------------------|-------------------------------|----------------------------------------------------|
//! | `Null`                                  | `Null`                        |                                                    |
//! | `Boolean(b)`                            | `Boolean(b)`                  |                                                    |
//! | `Int64(n)`                              | `Integer(n)`                  |                                                    |
//! | `UniqueId(n)`                           | `Integer(n as i64)`           | u32 → i64 widen (always non-negative)              |
//! | `Float64(f)`                            | `Float(f)`                    | IEEE 754 double; both sides preserve NaN bit-pat   |
//! | `String(s)`                             | `String(s.clone())`           | UTF-8                                              |
//! | `List(items)`                           | `List(items.map(to_bolt))`    | Recursive                                          |
//! | `Map(entries)`                          | `Dict(entries.map(...))`      | `BoltDict = HashMap<String, BoltValue>`            |
//! | `DateTime(NaiveDate)`                   | `Date(BoltDate { days })`     | Days since Unix epoch (1970-01-01)                 |
//! | `Timestamp(NaiveDateTime)`             | `LocalDateTime`               | Epoch seconds and fractional nanoseconds preserved |
//! | `Duration { months, days, seconds }`    | `Duration(BoltDuration {..})` | All i64; kglite has second precision (`nanoseconds: 0`) |
//! | `Point { lat, lon }`                    | `Point2D(BoltPoint2D { .. })` | srid=4326 for WGS84; Bolt convention x=lon, y=lat  |
//! | `Node { id, labels, properties }`       | `Node(BoltNode { id, labels, properties, element_id })` | `element_id = id.to_string()` |
//! | `Relationship { id, start_id, end_id, rel_type, properties }` | `Relationship(BoltRelationship { ... })` | `element_id` / `start_element_id` / `end_element_id` = stringified ids |
//! | `Path { nodes, rels }`                  | `Path(BoltPath { nodes, rels: UnboundRel*, indices })` | `indices`: 1-based signed rel idx + 0-based node idx pairs |
//! | `NodeRef(_)`                            | `Err(BoltError::Backend)`     | Internal placeholder — leaking here is an executor bug |
//!
//! `from_bolt` is the inbound direction for parameters. The graph-structure
//! variants (`Node`/`Relationship`/`Path`) on input would mean a driver
//! passed a node *as a parameter* — Neo4j drivers don't do this, and
//! `from_bolt` rejects them with `BoltError::Protocol`.

use std::collections::HashMap;

use boltr::error::BoltError;
use boltr::types::{
    BoltDate, BoltDict, BoltDuration, BoltLocalDateTime, BoltNode, BoltPath, BoltPoint2D,
    BoltRelationship, BoltUnboundRelationship, BoltValue,
};

use kglite::api::Value;

/// kglite → Bolt. Called by `execute`'s record emission.
///
/// Returns `Err(BoltError::Backend)` rather than panicking so a query
/// that touches an unimplemented variant doesn't orphan the tokio
/// connection task; the client gets a clean Bolt FAILURE instead.
pub fn to_bolt(value: &Value) -> Result<BoltValue, BoltError> {
    match value {
        // ---- Scalars -----------------------------------------------------
        Value::Null => Ok(BoltValue::Null),
        Value::Boolean(b) => Ok(BoltValue::Boolean(*b)),
        Value::Int64(n) => Ok(BoltValue::Integer(*n)),
        Value::UniqueId(n) => Ok(BoltValue::Integer(i64::from(*n))),
        Value::Float64(f) => Ok(BoltValue::Float(*f)),
        Value::String(s) => Ok(BoltValue::String(s.clone())),

        // ---- Recursive containers ---------------------------------------
        Value::List(items) => items
            .iter()
            .map(to_bolt)
            .collect::<Result<Vec<_>, _>>()
            .map(BoltValue::List),
        Value::Map(entries) => entries
            .iter()
            .map(|(k, v)| to_bolt(v).map(|bv| (k.to_string(), bv)))
            .collect::<Result<HashMap<_, _>, _>>()
            .map(BoltValue::Dict),

        // ---- Temporal / spatial / duration ------------------------------
        Value::DateTime(date) => {
            // kglite's `DateTime` is a `NaiveDate` (date only);
            // Bolt's BoltDate is also days-since-Unix-epoch.
            // `signed_duration_since` is unambiguous; the bare `-` op
            // resolves to the wrong impl in some chrono versions.
            // SAFETY: 1970-01-01 is a valid Gregorian date — the only way
            // `from_ymd_opt` returns None is on out-of-range/invalid input
            // (month 13, day 32, etc.). This expect is infallible.
            let epoch =
                chrono::NaiveDate::from_ymd_opt(1970, 1, 1).expect("1970-01-01 is a valid date");
            Ok(BoltValue::Date(BoltDate {
                days: date.signed_duration_since(epoch).num_days(),
            }))
        }
        Value::Timestamp(dt) => {
            // Floor seconds plus a nonnegative fraction also preserves instants
            // before the epoch; Duration::num_seconds would truncate toward zero.
            let utc = dt.and_utc();
            if utc.timestamp_subsec_nanos() >= 1_000_000_000 {
                return Err(BoltError::Backend(
                    "Timestamp leap seconds cannot be represented as Bolt LocalDateTime".into(),
                ));
            }
            Ok(BoltValue::LocalDateTime(BoltLocalDateTime {
                seconds: utc.timestamp(),
                nanoseconds: i64::from(utc.timestamp_subsec_nanos()),
            }))
        }
        Value::Point { lat, lon } => Ok(BoltValue::Point2D(BoltPoint2D {
            // SRID 4326 = WGS84 (geographic lat/lon). Bolt convention
            // is x=longitude, y=latitude. kglite stores them named, so
            // the cross-naming is intentional, not a bug.
            srid: 4326,
            x: *lon,
            y: *lat,
        })),
        Value::Duration {
            months,
            days,
            seconds,
        } => Ok(BoltValue::Duration(BoltDuration {
            months: i64::from(*months),
            days: i64::from(*days),
            seconds: *seconds,
            nanoseconds: 0,
        })),

        // ---- Node / Relationship / Path ---------------------------------
        Value::Node(node) => {
            let properties = props_to_bolt_dict(&node.properties)?;
            Ok(BoltValue::Node(BoltNode {
                id: i64::from(node.id),
                labels: node.labels.clone(),
                properties,
                // Bolt 5.x element_id: a stable string identifier.
                // Neo4j uses a UUID-like string; for kglite we use the
                // numeric id stringified — it's stable within one
                // server lifetime, which is the contract drivers care
                // about. (Across reloads the id may change; drivers
                // shouldn't persist element_ids long-term.)
                element_id: node.id.to_string(),
            }))
        }
        Value::Relationship(rel) => {
            let properties = props_to_bolt_dict(&rel.properties)?;
            Ok(BoltValue::Relationship(BoltRelationship {
                id: i64::from(rel.id),
                start_node_id: i64::from(rel.start_id),
                end_node_id: i64::from(rel.end_id),
                rel_type: rel.rel_type.clone(),
                properties,
                element_id: rel.id.to_string(),
                start_element_id: rel.start_id.to_string(),
                end_element_id: rel.end_id.to_string(),
            }))
        }
        Value::Path(path) => path_to_bolt_path(path.as_ref()).map(BoltValue::Path),

        // ---- Executor bug ----------------------------------------------
        Value::NodeRef(_) => Err(BoltError::Backend(
            "internal Value::NodeRef leaked through projection — please file a \
             bug against kglite (this variant is supposed to be materialized \
             before reaching the Bolt boundary)"
                .into(),
        )),
    }
}

/// kglite → Bolt, consuming the value: strings, labels and containers move
/// into the Bolt value instead of being cloned, and the engine's copy is freed
/// as the Bolt copy is built. Same mapping as [`to_bolt`], which it falls back
/// to for the variants that own no heap data.
pub fn to_bolt_owned(value: Value) -> Result<BoltValue, BoltError> {
    match value {
        Value::String(s) => Ok(BoltValue::String(s)),
        Value::List(items) => items
            .into_iter()
            .map(to_bolt_owned)
            .collect::<Result<Vec<_>, _>>()
            .map(BoltValue::List),
        Value::Map(entries) => props_into_bolt_dict(entries).map(BoltValue::Dict),
        Value::Node(node) => {
            let kglite::api::NodeValue {
                id,
                labels,
                properties,
            } = *node;
            Ok(BoltValue::Node(BoltNode {
                id: i64::from(id),
                labels,
                properties: props_into_bolt_dict(properties)?,
                element_id: id.to_string(),
            }))
        }
        Value::Relationship(rel) => {
            let rel = *rel;
            Ok(BoltValue::Relationship(BoltRelationship {
                id: i64::from(rel.id),
                start_node_id: i64::from(rel.start_id),
                end_node_id: i64::from(rel.end_id),
                element_id: rel.id.to_string(),
                start_element_id: rel.start_id.to_string(),
                end_element_id: rel.end_id.to_string(),
                rel_type: rel.rel_type,
                properties: props_into_bolt_dict(rel.properties)?,
            }))
        }
        Value::Path(path) => path_into_bolt_path(*path).map(BoltValue::Path),
        other => to_bolt(&other),
    }
}

fn props_into_bolt_dict(props: kglite::datatypes::PropMap) -> Result<BoltDict, BoltError> {
    let pairs = props.into_pairs();
    let mut dict = HashMap::with_capacity(pairs.len());
    for (key, value) in pairs {
        dict.insert(key.to_string(), to_bolt_owned(value)?);
    }
    Ok(dict)
}

/// Convert a kglite property map to a Bolt dict. Recursive through
/// `to_bolt` so nested lists / maps round-trip; surface conversion
/// failure on the first bad value.
fn props_to_bolt_dict(props: &kglite::datatypes::PropMap) -> Result<BoltDict, BoltError> {
    props
        .iter()
        .map(|(k, v)| to_bolt(v).map(|bv| (k.to_string(), bv)))
        .collect::<Result<HashMap<_, _>, _>>()
}

/// Build a `BoltPath` from a kglite `PathValue`. Encodes the
/// Neo4j Bolt-protocol `indices` scheme: pairs of (signed-rel-index,
/// next-node-index) where the rel index is 1-based with sign
/// (+ = traversed in the rel's natural direction, - = traversed in
/// reverse) and the node index is 0-based into `nodes`.
///
/// kglite's `PathValue` stores parallel `nodes` (k+1) + `rels` (k)
/// vectors with no deduplication and no direction sign — direction is
/// inferred per rel by comparing `rel.start_id` / `rel.end_id` against
/// the node ids before and after the rel in the traversal order. We
/// emit nodes 1:1 (no dedup) and one (signed_rel, next_node) pair
/// per rel.
fn path_to_bolt_path(p: &kglite::api::PathValue) -> Result<BoltPath, BoltError> {
    let indices = path_indices(p)?;
    let nodes: Vec<BoltNode> = p
        .nodes
        .iter()
        .map(|nv| {
            Ok::<BoltNode, BoltError>(BoltNode {
                id: i64::from(nv.id),
                labels: nv.labels.clone(),
                properties: props_to_bolt_dict(&nv.properties)?,
                element_id: nv.id.to_string(),
            })
        })
        .collect::<Result<Vec<_>, _>>()?;

    let rels: Vec<BoltUnboundRelationship> = p
        .rels
        .iter()
        .map(|rv| {
            Ok::<BoltUnboundRelationship, BoltError>(BoltUnboundRelationship {
                id: i64::from(rv.id),
                rel_type: rv.rel_type.clone(),
                properties: props_to_bolt_dict(&rv.properties)?,
                element_id: rv.id.to_string(),
            })
        })
        .collect::<Result<Vec<_>, _>>()?;

    Ok(BoltPath {
        nodes,
        rels,
        indices,
    })
}

/// [`path_to_bolt_path`] consuming the path.
fn path_into_bolt_path(p: kglite::api::PathValue) -> Result<BoltPath, BoltError> {
    let indices = path_indices(&p)?;
    let nodes = p
        .nodes
        .into_iter()
        .map(|nv| {
            Ok::<BoltNode, BoltError>(BoltNode {
                id: i64::from(nv.id),
                element_id: nv.id.to_string(),
                labels: nv.labels,
                properties: props_into_bolt_dict(nv.properties)?,
            })
        })
        .collect::<Result<Vec<_>, _>>()?;
    let rels = p
        .rels
        .into_iter()
        .map(|rv| {
            Ok::<BoltUnboundRelationship, BoltError>(BoltUnboundRelationship {
                id: i64::from(rv.id),
                element_id: rv.id.to_string(),
                rel_type: rv.rel_type,
                properties: props_into_bolt_dict(rv.properties)?,
            })
        })
        .collect::<Result<Vec<_>, _>>()?;
    Ok(BoltPath {
        nodes,
        rels,
        indices,
    })
}

/// Validate a path's shape and compute its Bolt `indices`.
fn path_indices(p: &kglite::api::PathValue) -> Result<Vec<i64>, BoltError> {
    // Sanity check: kglite paths are linear, so |nodes| = |rels| + 1.
    if p.nodes.len() != p.rels.len() + 1 {
        return Err(BoltError::Backend(format!(
            "kglite PathValue invariant violated: {} nodes vs {} rels (expected {} vs {})",
            p.nodes.len(),
            p.rels.len(),
            p.rels.len() + 1,
            p.rels.len()
        )));
    }
    let mut indices: Vec<i64> = Vec::with_capacity(p.rels.len() * 2);
    for (i, rel) in p.rels.iter().enumerate() {
        let node_before = &p.nodes[i];
        let node_after = &p.nodes[i + 1];
        // 1-based rel index; sign indicates traversal direction.
        let rel_idx_1based = (i + 1) as i64;
        let signed_rel: i64 = if rel.start_id == node_before.id && rel.end_id == node_after.id {
            rel_idx_1based // outgoing in path's traversal
        } else if rel.start_id == node_after.id && rel.end_id == node_before.id {
            -rel_idx_1based // incoming (traversed in reverse)
        } else {
            return Err(BoltError::Backend(format!(
                "kglite PathValue relationship {} ({} -> {}) does not connect path nodes {} -> {}",
                rel.id, rel.start_id, rel.end_id, node_before.id, node_after.id
            )));
        };
        indices.push(signed_rel);
        indices.push((i + 1) as i64); // 0-based next-node index
    }
    Ok(indices)
}

/// Bolt → kglite. Called by `execute`'s parameter decoding.
///
/// Returns `BoltError::Protocol` on inbound variants that don't make
/// sense in a parameter context: graph structures (Node/Relationship/
/// Path — drivers never pass these), time-of-day temporals (no standalone
/// time-of-day `Value` variant), Bytes (no `Value` variant), or Point3D
/// (kglite has only 2D points).
pub fn from_bolt(value: &BoltValue) -> Result<Value, BoltError> {
    match value {
        // ---- Scalars -----------------------------------------------------
        BoltValue::Null => Ok(Value::Null),
        BoltValue::Boolean(b) => Ok(Value::Boolean(*b)),
        BoltValue::Integer(n) => Ok(Value::Int64(*n)),
        // PackStream carries NaN and ±Infinity as ordinary float64 values and
        // every result path returns them unchanged, so a parameter takes
        // the same value a result would.
        BoltValue::Float(f) => Ok(Value::Float64(*f)),
        BoltValue::String(s) => Ok(Value::String(s.clone())),

        // ---- Recursive containers ---------------------------------------
        BoltValue::List(items) => items
            .iter()
            .map(from_bolt)
            .collect::<Result<Vec<_>, _>>()
            .map(Value::List),
        BoltValue::Dict(entries) => entries
            .iter()
            .map(|(k, v)| from_bolt(v).map(|kv| (k.as_str(), kv)))
            .collect::<Result<kglite::datatypes::PropMap, _>>()
            .map(Value::Map),

        // ---- Temporal / spatial ----------------------------------------
        BoltValue::Date(d) => {
            // SAFETY: 1970-01-01 is a valid Gregorian date — the only way
            // `from_ymd_opt` returns None is on out-of-range/invalid input
            // (month 13, day 32, etc.). This expect is infallible.
            let epoch =
                chrono::NaiveDate::from_ymd_opt(1970, 1, 1).expect("1970-01-01 is a valid date");
            let date = epoch
                .checked_add_signed(chrono::Duration::days(d.days))
                .ok_or_else(|| {
                    BoltError::Protocol(format!(
                        "Bolt Date out of range for kglite NaiveDate: days={}",
                        d.days
                    ))
                })?;
            Ok(Value::DateTime(date))
        }
        BoltValue::Duration(d) => Ok(Value::Duration {
            months: i32::try_from(d.months).map_err(|_| {
                BoltError::Protocol(format!(
                    "Bolt Duration.months out of i32 range: {}",
                    d.months
                ))
            })?,
            days: i32::try_from(d.days).map_err(|_| {
                BoltError::Protocol(format!("Bolt Duration.days out of i32 range: {}", d.days))
            })?,
            seconds: d.seconds,
            // kglite's `Value::Duration` carries second precision; if the
            // driver sent sub-second nanoseconds we silently truncate
            // them. The asymmetry is documented in the mapping table.
        }),
        BoltValue::Point2D(p) => {
            // SRID 4326 is WGS84 (geographic lat/lon). Other SRIDs (e.g.
            // 7203 for Cartesian) aren't representable as kglite's
            // `Point { lat, lon }`.
            if p.srid != 4326 {
                return Err(BoltError::Protocol(format!(
                    "Bolt Point2D with SRID {} not supported — kglite \
                     only represents WGS84 lat/lon (SRID 4326)",
                    p.srid
                )));
            }
            // Bolt convention: x=longitude, y=latitude.
            Ok(Value::Point { lat: p.y, lon: p.x })
        }

        // ---- Variants kglite can't represent ----------------------------
        BoltValue::Bytes(_) => Err(BoltError::Protocol(
            "Bolt Bytes parameter not supported — kglite has no byte-string Value variant".into(),
        )),
        BoltValue::LocalDateTime(dt) => {
            let stamp = u32::try_from(dt.nanoseconds)
                .ok()
                .filter(|nanos| *nanos < 1_000_000_000)
                .and_then(|nanos| chrono::DateTime::from_timestamp(dt.seconds, nanos))
                .ok_or_else(|| {
                    BoltError::Protocol(format!(
                        "Bolt LocalDateTime out of range: seconds={}, nanoseconds={}",
                        dt.seconds, dt.nanoseconds
                    ))
                })?;
            Ok(Value::Timestamp(stamp.naive_utc()))
        }
        BoltValue::Time(_)
        | BoltValue::LocalTime(_)
        | BoltValue::DateTime(_)
        | BoltValue::DateTimeZoneId(_) => Err(BoltError::Protocol(
            "Bolt zoned timestamp / time-of-day parameters not supported — kglite's \
             temporal Values are zoneless (use LocalDateTime / Date)"
                .into(),
        )),
        BoltValue::Point3D(_) => Err(BoltError::Protocol(
            "Bolt Point3D parameter not supported — kglite represents only 2D points".into(),
        )),

        // ---- Inbound graph structures (drivers don't pass these) -------
        BoltValue::Node(_) | BoltValue::Relationship(_) | BoltValue::Path(_) => {
            Err(BoltError::Protocol(
                "Bolt Node/Relationship/Path is not a valid parameter type — \
                 drivers should serialize property values instead"
                    .into(),
            ))
        }
        BoltValue::UnboundRelationship(_) => Err(BoltError::Protocol(
            "Bolt UnboundRelationship only appears inside Path structures — \
             cannot be a standalone parameter"
                .into(),
        )),
    }
}

#[cfg(test)]
mod tests {

    use kglite::api::{NodeValue, PathValue, RelValue};

    use super::*;

    fn node(id: u32) -> NodeValue {
        NodeValue {
            id,
            labels: vec!["N".into()],
            properties: kglite::datatypes::PropMap::new(),
        }
    }

    fn rel(id: u32, start_id: u32, end_id: u32) -> RelValue {
        RelValue {
            incarnation: None,
            id,
            start_id,
            end_id,
            rel_type: "R".into(),
            properties: kglite::datatypes::PropMap::new(),
        }
    }

    #[test]
    fn non_finite_float_parameters_bind_unchanged() {
        for f in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY, -0.0] {
            let Ok(Value::Float64(bound)) = from_bolt(&BoltValue::Float(f)) else {
                panic!("{f} must bind as a Float64 parameter");
            };
            assert_eq!(bound.to_bits(), f.to_bits());
        }
    }

    #[test]
    fn path_adapter_preserves_reverse_traversal_direction() {
        let path = PathValue {
            nodes: vec![node(2), node(1)],
            rels: vec![rel(7, 1, 2)],
        };

        let encoded = path_to_bolt_path(&path).expect("valid reverse path");
        assert_eq!(encoded.indices, vec![-1, 1]);
    }

    #[test]
    fn path_adapter_rejects_relationship_disconnected_from_segment() {
        let path = PathValue {
            nodes: vec![node(1), node(2)],
            rels: vec![rel(7, 3, 4)],
        };

        let error = path_to_bolt_path(&path).expect_err("disconnected relationship must fail");
        assert!(error
            .to_string()
            .contains("does not connect path nodes 1 -> 2"));
    }

    #[test]
    fn path_adapter_rejects_non_linear_shape() {
        let path = PathValue {
            nodes: vec![node(1)],
            rels: vec![rel(7, 1, 2)],
        };

        let error =
            path_to_bolt_path(&path).expect_err("invalid node/relationship count must fail");
        assert!(error.to_string().contains("invariant violated"));
    }
}

#[cfg(test)]
mod fractional_timestamp_contract_tests {
    use super::*;
    #[test]
    fn local_datetime_roundtrip_preserves_fraction_and_negative_epoch_floor() {
        for (text, seconds, nanos) in [
            ("1970-01-01T00:00:00", 0, 0),
            ("1970-01-01T00:00:00.123456789", 0, 123456789),
            ("1969-12-31T23:59:59.500", -1, 500000000),
        ] {
            let value = Value::Timestamp(
                chrono::NaiveDateTime::parse_from_str(text, "%Y-%m-%dT%H:%M:%S%.f").unwrap(),
            );
            let bolt = BoltValue::LocalDateTime(BoltLocalDateTime {
                seconds,
                nanoseconds: nanos,
            });
            assert_eq!(to_bolt(&value).unwrap(), bolt);
            assert_eq!(from_bolt(&bolt).unwrap(), value);
            let nested = Value::List(vec![value]);
            assert_eq!(from_bolt(&to_bolt(&nested).unwrap()).unwrap(), nested);
        }
    }
}

#[cfg(test)]
mod timestamp_representability_tests {
    use super::*;
    #[test]
    fn chrono_leap_second_is_refused_without_silent_normalization() {
        let leap = chrono::NaiveDate::from_ymd_opt(2016, 12, 31)
            .unwrap()
            .and_hms_nano_opt(23, 59, 59, 1_500_000_000)
            .unwrap();
        let value = Value::Timestamp(leap);
        assert!(
            matches!(to_bolt(&value), Err(BoltError::Backend(message)) if message.contains("leap seconds"))
        );
        assert!(matches!(
            to_bolt(&Value::List(vec![value])),
            Err(BoltError::Backend(_))
        ));
        let incompatible = BoltValue::LocalDateTime(BoltLocalDateTime {
            seconds: leap.and_utc().timestamp(),
            nanoseconds: 1_500_000_000,
        });
        assert!(matches!(
            from_bolt(&incompatible),
            Err(BoltError::Protocol(_))
        ));
    }
}

#[cfg(test)]
mod owned_conversion_tests {
    use kglite::api::{NodeValue, PathValue, RelValue};
    use kglite::datatypes::PropMap;

    use super::*;

    fn props() -> PropMap {
        PropMap::from_pairs(vec![
            ("name".into(), Value::String("Ada".into())),
            (
                "tags".into(),
                Value::List(vec![Value::Int64(1), Value::Null]),
            ),
            (
                "nested".into(),
                Value::Map(PropMap::from_pairs(vec![(
                    "k".into(),
                    Value::Boolean(true),
                )])),
            ),
        ])
    }

    fn node(id: u32) -> NodeValue {
        NodeValue {
            id,
            labels: vec!["Person".into(), "Admin".into()],
            properties: props(),
        }
    }

    fn rel(id: u32, start_id: u32, end_id: u32) -> RelValue {
        RelValue {
            incarnation: None,
            id,
            start_id,
            end_id,
            rel_type: "KNOWS".into(),
            properties: props(),
        }
    }

    /// `to_bolt_owned` must produce exactly what `to_bolt` does, for every variant.
    #[test]
    fn owned_conversion_matches_borrowed_conversion() {
        let values = vec![
            Value::Null,
            Value::Boolean(true),
            Value::Int64(-7),
            Value::UniqueId(9),
            Value::Float64(1.5),
            Value::String("text".into()),
            Value::List(vec![
                Value::String("a".into()),
                Value::List(vec![Value::Int64(1)]),
            ]),
            Value::Map(props()),
            Value::DateTime(chrono::NaiveDate::from_ymd_opt(2024, 2, 29).unwrap()),
            Value::Timestamp(
                chrono::NaiveDate::from_ymd_opt(2024, 2, 29)
                    .unwrap()
                    .and_hms_nano_opt(1, 2, 3, 456)
                    .unwrap(),
            ),
            Value::Point { lat: 1.0, lon: 2.0 },
            Value::Duration {
                months: 1,
                days: 2,
                seconds: 3,
            },
            Value::Node(Box::new(node(4))),
            Value::Relationship(Box::new(rel(5, 4, 6))),
            Value::Path(Box::new(PathValue {
                nodes: vec![node(2), node(1), node(3)],
                rels: vec![rel(7, 1, 2), rel(8, 1, 3)],
            })),
        ];
        for value in values {
            assert_eq!(
                to_bolt_owned(value.clone()).unwrap(),
                to_bolt(&value).unwrap(),
                "{value:?}"
            );
        }
    }

    #[test]
    fn owned_conversion_refuses_what_borrowed_conversion_refuses() {
        let leap = chrono::NaiveDate::from_ymd_opt(2016, 12, 31)
            .unwrap()
            .and_hms_nano_opt(23, 59, 59, 1_500_000_000)
            .unwrap();
        for value in [
            Value::NodeRef(1),
            Value::List(vec![Value::Timestamp(leap)]),
            Value::Path(Box::new(PathValue {
                nodes: vec![node(1)],
                rels: vec![rel(7, 1, 2)],
            })),
            Value::Path(Box::new(PathValue {
                nodes: vec![node(1), node(2)],
                rels: vec![rel(7, 3, 4)],
            })),
        ] {
            assert!(to_bolt(&value).is_err(), "{value:?}");
            assert!(to_bolt_owned(value.clone()).is_err(), "{value:?}");
        }
    }
}