reddb-io-server 1.24.2

use std::collections::BTreeSet;

use crate::application::CreateEntityOutput;
use crate::json::{
    from_slice as json_from_slice, to_string as json_to_string, Map, Value as JsonValue,
};
use crate::runtime::ScanPage;
use crate::storage::{CrossRef, EntityData, EntityKind, UnifiedEntity};
use reddb_types::Value;

const MAX_JSON_SAFE_INTEGER: i64 = 9_007_199_254_740_991;

pub(crate) fn created_entity_output_json(output: &CreateEntityOutput) -> JsonValue {
    let mut object = Map::new();
    object.insert("ok".to_string(), JsonValue::Bool(true));
    object.insert("rid".to_string(), exact_u64_to_json(output.id.raw()));
    object.insert("id".to_string(), exact_u64_to_json(output.id.raw()));
    object.insert(
        "entity".to_string(),
        output
            .entity
            .as_ref()
            .map(entity_json)
            .unwrap_or(JsonValue::Null),
    );
    JsonValue::Object(object)
}

pub(crate) fn scan_page_json(page: &ScanPage) -> JsonValue {
    let mut object = Map::new();
    object.insert(
        "collection".to_string(),
        JsonValue::String(page.collection.clone()),
    );
    object.insert("total".to_string(), JsonValue::Number(page.total as f64));
    object.insert(
        "next_offset".to_string(),
        match page.next {
            Some(cursor) => JsonValue::Number(cursor.offset as f64),
            None => JsonValue::Null,
        },
    );
    object.insert(
        "items".to_string(),
        JsonValue::Array(page.items.iter().map(entity_json).collect()),
    );
    JsonValue::Object(object)
}

pub(crate) use crate::entity_render::{
    base_entity_object, compact_entity_json, compact_entity_json_string, named_fields_json,
};

pub(crate) fn entity_json(entity: &UnifiedEntity) -> JsonValue {
    let mut object = base_entity_object(entity);
    object.insert("identity".to_string(), entity_kind_json(&entity.kind));
    object.insert("data".to_string(), entity_data_json(&entity.data));
    object.insert(
        "cross_refs".to_string(),
        JsonValue::Array(entity.cross_refs().iter().map(cross_ref_json).collect()),
    );
    JsonValue::Object(object)
}

pub(crate) fn storage_value_to_json(value: &Value) -> JsonValue {
    match value {
        Value::Null => JsonValue::Null,
        Value::Integer(value) => exact_i64_to_json(*value),
        Value::UnsignedInteger(value) => exact_u64_to_json(*value),
        Value::Float(value) if value.is_finite() => JsonValue::Number(*value),
        // Canonical query envelopes use the same lossless non-finite tags
        // understood by RedDB drivers. Bare NaN/Infinity are not RFC 8259
        // JSON and cannot be read back by `crate::json::from_slice`.
        Value::Float(value) => {
            let token = if value.is_nan() {
                "NaN"
            } else if value.is_sign_positive() {
                "Infinity"
            } else {
                "-Infinity"
            };
            single_key_object("$float", JsonValue::String(token.to_string()))
        }
        Value::Text(value) => JsonValue::String(value.to_string()),
        Value::Blob(value) => JsonValue::String(hex::encode(value)),
        Value::Boolean(value) => JsonValue::Bool(*value),
        Value::Timestamp(value) => exact_i64_to_json(*value),
        Value::Duration(value) => exact_i64_to_json(*value),
        Value::IpAddr(value) => JsonValue::String(value.to_string()),
        Value::MacAddr(value) => JsonValue::String(format_mac(value)),
        Value::Vector(value) => JsonValue::Array(
            value
                .iter()
                .map(|entry| JsonValue::Number(*entry as f64))
                .collect(),
        ),
        Value::Json(value) => storage_json_bytes_to_json(value),
        Value::Uuid(value) => JsonValue::String(hex::encode(value)),
        Value::NodeRef(value) => JsonValue::String(value.clone()),
        Value::EdgeRef(value) => JsonValue::String(value.clone()),
        Value::VectorRef(collection, id) => {
            let mut object = Map::new();
            object.insert(
                "collection".to_string(),
                JsonValue::String(collection.clone()),
            );
            object.insert("id".to_string(), exact_u64_to_json(*id));
            JsonValue::Object(object)
        }
        Value::RowRef(table, row_id) => {
            let mut object = Map::new();
            object.insert("table".to_string(), JsonValue::String(table.to_string()));
            object.insert("row_id".to_string(), exact_u64_to_json(*row_id));
            JsonValue::Object(object)
        }
        Value::Color([r, g, b]) => JsonValue::String(format!("#{:02X}{:02X}{:02X}", r, g, b)),
        Value::Email(s) => JsonValue::String(s.clone()),
        Value::Url(s) => JsonValue::String(s.clone()),
        Value::Phone(n) => exact_u64_to_json(*n),
        Value::Semver(packed) => JsonValue::String(format!(
            "{}.{}.{}",
            packed / 1_000_000,
            (packed / 1_000) % 1_000,
            packed % 1_000
        )),
        Value::Cidr(ip, prefix) => JsonValue::String(format!(
            "{}.{}.{}.{}/{}",
            (ip >> 24) & 0xFF,
            (ip >> 16) & 0xFF,
            (ip >> 8) & 0xFF,
            ip & 0xFF,
            prefix
        )),
        Value::Date(days) => JsonValue::Integer(i64::from(*days)),
        Value::Time(ms) => JsonValue::Integer(i64::from(*ms)),
        Value::Decimal(v) => {
            exact_decimal_to_json(reddb_types::Value::Decimal(*v).display_string())
        }
        Value::DecimalText(v) => exact_decimal_to_json(v.clone()),
        Value::EnumValue(i) => JsonValue::Integer(i64::from(*i)),
        Value::Array(elems) => JsonValue::Array(elems.iter().map(storage_value_to_json).collect()),
        Value::TimestampMs(ms) => exact_i64_to_json(*ms),
        Value::Ipv4(ip) => JsonValue::String(format!(
            "{}.{}.{}.{}",
            (ip >> 24) & 0xFF,
            (ip >> 16) & 0xFF,
            (ip >> 8) & 0xFF,
            ip & 0xFF
        )),
        Value::Ipv6(bytes) => JsonValue::String(format!("{}", std::net::Ipv6Addr::from(*bytes))),
        Value::Subnet(ip, mask) => {
            let prefix = mask.leading_ones();
            JsonValue::String(format!(
                "{}.{}.{}.{}/{}",
                (ip >> 24) & 0xFF,
                (ip >> 16) & 0xFF,
                (ip >> 8) & 0xFF,
                ip & 0xFF,
                prefix
            ))
        }
        Value::Port(p) => JsonValue::Integer(i64::from(*p)),
        Value::Latitude(micro) => JsonValue::Number(*micro as f64 / 1_000_000.0),
        Value::Longitude(micro) => JsonValue::Number(*micro as f64 / 1_000_000.0),
        Value::GeoPoint(lat, lon) => JsonValue::String(format!(
            "{:.6},{:.6}",
            *lat as f64 / 1_000_000.0,
            *lon as f64 / 1_000_000.0
        )),
        Value::Country2(c) => JsonValue::String(String::from_utf8_lossy(c).to_string()),
        Value::Country3(c) => JsonValue::String(String::from_utf8_lossy(c).to_string()),
        Value::Lang2(c) => JsonValue::String(String::from_utf8_lossy(c).to_string()),
        Value::Lang5(c) => JsonValue::String(String::from_utf8_lossy(c).to_string()),
        Value::Currency(c) => JsonValue::String(String::from_utf8_lossy(c).to_string()),
        Value::AssetCode(code) => JsonValue::String(code.clone()),
        Value::Money {
            asset_code,
            minor_units,
            scale,
        } => {
            let mut object = Map::new();
            object.insert(
                "asset_code".to_string(),
                JsonValue::String(asset_code.clone()),
            );
            object.insert("minor_units".to_string(), exact_i64_to_json(*minor_units));
            object.insert("scale".to_string(), exact_i64_to_json(i64::from(*scale)));
            JsonValue::Object(object)
        }
        Value::ColorAlpha([r, g, b, a]) => {
            JsonValue::String(format!("#{:02X}{:02X}{:02X}{:02X}", r, g, b, a))
        }
        Value::BigInt(v) => exact_i64_to_json(*v),
        Value::KeyRef(col, key) => {
            let mut object = Map::new();
            object.insert("collection".to_string(), JsonValue::String(col.clone()));
            object.insert("key".to_string(), JsonValue::String(key.clone()));
            JsonValue::Object(object)
        }
        Value::DocRef(col, id) => {
            let mut object = Map::new();
            object.insert("collection".to_string(), JsonValue::String(col.clone()));
            object.insert("id".to_string(), exact_u64_to_json(*id));
            JsonValue::Object(object)
        }
        Value::TableRef(name) => JsonValue::String(name.clone()),
        Value::PageRef(page_id) => exact_u64_to_json(u64::from(*page_id)),
        // Secrets and passwords are always masked in JSON output.
        // SELECT decryption (if the vault is unsealed) happens earlier
        // in the query pipeline and replaces Value::Secret with
        // Value::Text before serialization.
        Value::Secret(_) => JsonValue::String("***".to_string()),
        Value::Password(_) => JsonValue::String("***".to_string()),
    }
}

fn exact_i64_to_json(value: i64) -> JsonValue {
    if (-MAX_JSON_SAFE_INTEGER..=MAX_JSON_SAFE_INTEGER).contains(&value) {
        JsonValue::Integer(value)
    } else {
        single_key_object("$int", JsonValue::String(value.to_string()))
    }
}

fn exact_u64_to_json(value: u64) -> JsonValue {
    if value <= MAX_JSON_SAFE_INTEGER as u64 {
        JsonValue::Integer(value as i64)
    } else {
        single_key_object("$uint", JsonValue::String(value.to_string()))
    }
}

fn exact_decimal_to_json(value: String) -> JsonValue {
    single_key_object("$decimal", JsonValue::String(value))
}

fn single_key_object(key: &str, value: JsonValue) -> JsonValue {
    JsonValue::Object([(key.to_string(), value)].into_iter().collect())
}

pub(crate) fn storage_json_bytes_to_json(bytes: &[u8]) -> JsonValue {
    // A document body may be stored as the native binary container (PRD-1398);
    // decode it back to JSON for the wire. Plain-JSON bytes fall through.
    if let Some(body) = crate::document_body::decode_container_to_json(bytes) {
        return body;
    }
    json_from_slice::<JsonValue>(bytes).unwrap_or_else(|_| {
        JsonValue::Object(
            [
                (
                    "code".to_string(),
                    JsonValue::String("INVALID_JSON".to_string()),
                ),
                ("hex".to_string(), JsonValue::String(hex::encode(bytes))),
            ]
            .into_iter()
            .collect(),
        )
    })
}

fn entity_kind_json(kind: &EntityKind) -> JsonValue {
    let mut object = Map::new();
    match kind {
        EntityKind::TableRow { table, row_id } => {
            object.insert("table".to_string(), JsonValue::String(table.to_string()));
            object.insert("row_id".to_string(), JsonValue::Number(*row_id as f64));
        }
        EntityKind::GraphNode(ref node) => {
            object.insert("label".to_string(), JsonValue::String(node.label.clone()));
            object.insert(
                "node_type".to_string(),
                JsonValue::String(node.node_type.clone()),
            );
        }
        EntityKind::GraphEdge(ref edge) => {
            object.insert("label".to_string(), JsonValue::String(edge.label.clone()));
            object.insert(
                "from_node".to_string(),
                JsonValue::String(edge.from_node.clone()),
            );
            object.insert(
                "to_node".to_string(),
                JsonValue::String(edge.to_node.clone()),
            );
            object.insert("weight".to_string(), JsonValue::Number(edge.weight as f64));
        }
        EntityKind::Vector { collection } => {
            object.insert(
                "collection".to_string(),
                JsonValue::String(collection.clone()),
            );
        }
        EntityKind::TimeSeriesPoint(ref ts) => {
            object.insert("series".to_string(), JsonValue::String(ts.series.clone()));
            object.insert("metric".to_string(), JsonValue::String(ts.metric.clone()));
        }
        EntityKind::QueueMessage { queue, position } => {
            object.insert("queue".to_string(), JsonValue::String(queue.clone()));
            object.insert("position".to_string(), JsonValue::Number(*position as f64));
        }
    }
    JsonValue::Object(object)
}

fn entity_data_json(data: &EntityData) -> JsonValue {
    let mut object = Map::new();
    match data {
        EntityData::Row(row) => {
            object.insert(
                "columns".to_string(),
                JsonValue::Array(row.columns.iter().map(storage_value_to_json).collect()),
            );
            object.insert(
                "named".to_string(),
                match &row.named {
                    Some(named) => JsonValue::Object(named_fields_json(named)),
                    None => JsonValue::Null,
                },
            );
        }
        EntityData::Node(node) => {
            object.insert(
                "properties".to_string(),
                JsonValue::Object(
                    node.properties
                        .iter()
                        .map(|(key, value)| (key.clone(), storage_value_to_json(value)))
                        .collect(),
                ),
            );
        }
        EntityData::Edge(edge) => {
            object.insert("weight".to_string(), JsonValue::Number(edge.weight as f64));
            object.insert(
                "properties".to_string(),
                JsonValue::Object(
                    edge.properties
                        .iter()
                        .map(|(key, value)| (key.clone(), storage_value_to_json(value)))
                        .collect(),
                ),
            );
        }
        EntityData::Vector(vector) => {
            object.insert(
                "dense".to_string(),
                JsonValue::Array(
                    vector
                        .dense
                        .iter()
                        .map(|value| JsonValue::Number(*value as f64))
                        .collect(),
                ),
            );
            object.insert(
                "sparse".to_string(),
                match &vector.sparse {
                    Some(sparse) => {
                        let mut sparse_object = Map::new();
                        sparse_object.insert(
                            "indices".to_string(),
                            JsonValue::Array(
                                sparse
                                    .indices
                                    .iter()
                                    .map(|value| JsonValue::Number(*value as f64))
                                    .collect(),
                            ),
                        );
                        sparse_object.insert(
                            "values".to_string(),
                            JsonValue::Array(
                                sparse
                                    .values
                                    .iter()
                                    .map(|value| JsonValue::Number(*value as f64))
                                    .collect(),
                            ),
                        );
                        JsonValue::Object(sparse_object)
                    }
                    None => JsonValue::Null,
                },
            );
            object.insert(
                "content".to_string(),
                match &vector.content {
                    Some(content) => JsonValue::String(content.clone()),
                    None => JsonValue::Null,
                },
            );
        }
        EntityData::TimeSeries(ts) => {
            object.insert("metric".to_string(), JsonValue::String(ts.metric.clone()));
            object.insert(
                "timestamp_ns".to_string(),
                JsonValue::Number(ts.timestamp_ns as f64),
            );
            object.insert("value".to_string(), JsonValue::Number(ts.value));
            object.insert(
                "tags".to_string(),
                JsonValue::Object(
                    ts.tags
                        .iter()
                        .map(|(k, v)| {
                            (
                                k.clone(),
                                crate::runtime::query_exec::decode_stored_tag_value(v),
                            )
                        })
                        .collect(),
                ),
            );
        }
        EntityData::QueueMessage(msg) => {
            object.insert("payload".to_string(), storage_value_to_json(&msg.payload));
            if let Some(priority) = msg.priority {
                object.insert("priority".to_string(), JsonValue::Number(priority as f64));
            }
            object.insert(
                "enqueued_at_ns".to_string(),
                JsonValue::Number(msg.enqueued_at_ns as f64),
            );
            object.insert(
                "attempts".to_string(),
                JsonValue::Number(msg.attempts as f64),
            );
            object.insert(
                "max_attempts".to_string(),
                JsonValue::Number(msg.max_attempts as f64),
            );
            object.insert("acked".to_string(), JsonValue::Bool(msg.acked));
        }
    }

    JsonValue::Object(object)
}

fn cross_ref_json(cross_ref: &CrossRef) -> JsonValue {
    let mut object = Map::new();
    object.insert(
        "source".to_string(),
        JsonValue::Number(cross_ref.source.raw() as f64),
    );
    object.insert(
        "target".to_string(),
        JsonValue::Number(cross_ref.target.raw() as f64),
    );
    object.insert(
        "target_collection".to_string(),
        JsonValue::String(cross_ref.target_collection.clone()),
    );
    object.insert(
        "ref_type".to_string(),
        JsonValue::String(format!("{:?}", cross_ref.ref_type)),
    );
    object.insert(
        "weight".to_string(),
        JsonValue::Number(cross_ref.weight as f64),
    );
    object.insert(
        "created_at".to_string(),
        JsonValue::Number(cross_ref.created_at as f64),
    );
    JsonValue::Object(object)
}

fn format_mac(bytes: &[u8; 6]) -> String {
    bytes
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<Vec<_>>()
        .join(":")
}

#[cfg(test)]
mod tests {
    use std::net::{IpAddr, Ipv4Addr};

    use super::*;

    #[test]
    fn storage_value_to_json_preserves_exact_numbers() {
        assert_eq!(
            storage_value_to_json(&Value::Integer(9_007_199_254_740_993)),
            single_key_object("$int", JsonValue::String("9007199254740993".to_string()))
        );
        assert_eq!(
            storage_value_to_json(&Value::UnsignedInteger(i64::MAX as u64 + 1)),
            single_key_object(
                "$uint",
                JsonValue::String("9223372036854775808".to_string())
            )
        );
        assert_eq!(
            storage_value_to_json(&Value::Decimal(1_234_567)),
            single_key_object("$decimal", JsonValue::String("123.4567".to_string()))
        );
        assert_eq!(
            storage_value_to_json(&Value::DecimalText(
                "3.141592653589793238462643383279".to_string()
            )),
            single_key_object(
                "$decimal",
                JsonValue::String("3.141592653589793238462643383279".to_string())
            )
        );
    }

    #[test]
    fn keystone_json_is_byte_identical_to_presentation_json() {
        let document_fields = [
            Value::Integer(7),
            Value::DecimalText("18446744073709551616.0001".to_string()),
        ];
        let mut document_body = Vec::new();
        reddb_types::document_body_codec::encode(
            &[
                ("integer", &document_fields[0]),
                ("decimal", &document_fields[1]),
            ],
            &mut document_body,
        )
        .expect("representative document body should encode");

        let corpus = [
            Value::Null,
            Value::Integer(9_007_199_254_740_992),
            Value::UnsignedInteger(u64::MAX),
            Value::Float(1.25),
            Value::text("hello"),
            Value::Blob(vec![0x00, 0xff]),
            Value::Boolean(true),
            Value::Timestamp(1_700_000_000),
            Value::TimestampMs(1_700_000_000_123),
            Value::Duration(-500),
            Value::IpAddr(IpAddr::V4(Ipv4Addr::LOCALHOST)),
            Value::MacAddr([0, 1, 2, 3, 4, 5]),
            Value::Vector(vec![1.5, -2.0]),
            Value::Json(br#"{"nested":[1,2.5]}"#.to_vec()),
            Value::Json(vec![0xff]),
            Value::Json(document_body),
            Value::Uuid([0; 16]),
            Value::VectorRef("embeddings".to_string(), 7),
            Value::RowRef("users".to_string(), 8),
            Value::Cidr(0xc0a8_0100, 24),
            Value::Decimal(1_234_567),
            Value::DecimalText("3.141592653589793238462643383279".to_string()),
            Value::Array(vec![
                Value::Boolean(false),
                Value::Array(vec![Value::Integer(2)]),
            ]),
            Value::GeoPoint(12_345_678, -87_654_321),
            Value::Money {
                asset_code: "BRL".to_string(),
                minor_units: 12_345,
                scale: 2,
            },
            Value::KeyRef("settings".to_string(), "theme".to_string()),
            Value::DocRef("profiles".to_string(), 9),
            Value::Secret(vec![1, 2, 3]),
            Value::Password("hash".to_string()),
        ];

        for value in corpus {
            let legacy = storage_value_to_json(&value).to_string_compact();
            let keystone = value.to_json().to_string_compact();
            assert_eq!(keystone.as_bytes(), legacy.as_bytes(), "{value:?}");
        }
    }
}