nautilus_model/data/

registry.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Registries for custom data: JSON (de)serialization and Arrow encode/decode.
//!
//! Mirrors Python's `register_custom_data_class` surface in `custom.py`.
//! The registry only stores type name -> callbacks for lookup; each type provides
//! its own deserialize/encode/decode via the trait or registration.

use std::sync::Arc;

#[cfg(feature = "arrow")]
use arrow::{
    datatypes::{DataType as ArrowDataType, Field, Schema},
    record_batch::RecordBatch,
};
use dashmap::{DashMap, mapref::entry::Entry};
use nautilus_core::Params;
#[cfg(feature = "python")]
use pyo3::types::PyAnyMethods;

use crate::data::{CustomData, CustomDataTrait, Data, DataType};

pub type JsonDeserializer =
    Box<dyn Fn(serde_json::Value) -> Result<Arc<dyn CustomDataTrait>, anyhow::Error> + Send + Sync>;
#[cfg(feature = "arrow")]
pub type ArrowEncoder =
    Box<dyn Fn(&[Arc<dyn CustomDataTrait>]) -> Result<RecordBatch, anyhow::Error> + Send + Sync>;
#[cfg(feature = "arrow")]
pub type ArrowDecoder = Box<
    dyn Fn(
            &std::collections::HashMap<String, String>,
            RecordBatch,
        ) -> Result<Vec<Data>, anyhow::Error>
        + Send
        + Sync,
>;

/// Validates that a custom Arrow write schema contains no unsupported opaque byte fields.
///
/// `allow_binary` is reserved for the Rust macro's documented `Vec<u8>` exemption. Python
/// schemas cannot prove that provenance and must pass `false`.
///
/// # Errors
///
/// Returns an error if an unsupported opaque byte field is found. The error names the first such
/// field.
#[cfg(feature = "arrow")]
pub fn validate_custom_arrow_schema(
    type_name: &str,
    schema: &Schema,
    allow_binary: bool,
) -> anyhow::Result<()> {
    for field in schema.fields() {
        validate_custom_arrow_field(type_name, field, allow_binary)?;
    }
    Ok(())
}

#[cfg(feature = "arrow")]
fn validate_custom_arrow_field(
    type_name: &str,
    field: &Field,
    allow_binary: bool,
) -> anyhow::Result<()> {
    match field.data_type() {
        ArrowDataType::Binary if allow_binary => Ok(()),
        ArrowDataType::Binary
        | ArrowDataType::LargeBinary
        | ArrowDataType::BinaryView
        | ArrowDataType::FixedSizeBinary(_) => anyhow::bail!(
            "custom write schema `{type_name}` contains opaque byte field `{}`: {}",
            field.name(),
            field.data_type(),
        ),
        ArrowDataType::List(child)
        | ArrowDataType::LargeList(child)
        | ArrowDataType::ListView(child)
        | ArrowDataType::LargeListView(child)
        | ArrowDataType::FixedSizeList(child, _)
        | ArrowDataType::Map(child, _) => {
            validate_custom_arrow_field(type_name, child, allow_binary)
        }
        ArrowDataType::Struct(children) => {
            for child in children {
                validate_custom_arrow_field(type_name, child, allow_binary)?;
            }
            Ok(())
        }
        ArrowDataType::Dictionary(_, value) => {
            let child = Field::new(field.name(), value.as_ref().clone(), field.is_nullable());
            validate_custom_arrow_field(type_name, &child, allow_binary)
        }
        _ => Ok(()),
    }
}

struct Registries {
    json: DashMap<String, JsonDeserializer>,
    #[cfg(feature = "arrow")]
    arrow: DashMap<String, (Arc<Schema>, ArrowEncoder, ArrowDecoder)>,
}

fn registries() -> &'static Registries {
    static REGISTRIES: std::sync::OnceLock<Registries> = std::sync::OnceLock::new();
    REGISTRIES.get_or_init(|| Registries {
        json: DashMap::new(),
        #[cfg(feature = "arrow")]
        arrow: DashMap::new(),
    })
}

/// Registers a JSON deserializer for the given custom data type name.
/// When `Data::deserialize` sees this type name, it will call this function.
///
/// # Errors
/// Returns an error if the type is already registered.
pub fn register_json_deserializer(
    type_name: &str,
    deserializer: JsonDeserializer,
) -> Result<(), anyhow::Error> {
    let reg = registries();
    match reg.json.entry(type_name.to_string()) {
        Entry::Occupied(_) => {
            anyhow::bail!("Custom data type \"{type_name}\" is already registered for JSON");
        }
        Entry::Vacant(v) => {
            v.insert(deserializer);
            Ok(())
        }
    }
}

/// Registers a JSON deserializer for the given custom data type name if not already registered.
/// If the type is already registered, returns `Ok(())` without overwriting (idempotent).
/// Use this where repeated registration can occur (e.g. module init).
///
/// Does not return an error (idempotent insert into `DashMap`).
#[allow(
    clippy::missing_errors_doc,
    reason = "idempotent registration only inserts into a DashMap"
)]
pub fn ensure_json_deserializer_registered(
    type_name: &str,
    deserializer: JsonDeserializer,
) -> Result<(), anyhow::Error> {
    let reg = registries();
    reg.json
        .entry(type_name.to_string())
        .or_insert_with(|| deserializer);
    Ok(())
}

/// Parses a "`data_type`" JSON object into `DataType` (`type_name`, metadata, identifier).
fn parse_data_type_from_value(value: &serde_json::Value) -> anyhow::Result<Option<DataType>> {
    let Some(obj) = value.get("data_type").and_then(|v| v.as_object()) else {
        return Ok(None);
    };

    let Some(type_name) = obj.get("type_name").and_then(|v| v.as_str()) else {
        return Ok(None);
    };

    let metadata = obj.get("metadata").and_then(|m| {
        if m.is_null() {
            None
        } else {
            let p: Params = serde_json::from_value(m.clone()).ok()?;
            if p.is_empty() { None } else { Some(p) }
        }
    });
    let identifier = obj
        .get("identifier")
        .and_then(|v| v.as_str())
        .map(String::from);
    DataType::try_new(type_name, metadata, identifier).map(Some)
}

/// Parses the canonical `CustomData` JSON envelope `{ type, data_type, payload }` and returns
/// the payload value to pass to the registered type deserializer. Does not depend on
/// user payload field names.
fn parse_envelope_payload(value: &serde_json::Value) -> Result<serde_json::Value, anyhow::Error> {
    let payload = value
        .get("payload")
        .cloned()
        .ok_or_else(|| anyhow::anyhow!("CustomData JSON missing 'payload' field"))?;
    Ok(payload)
}

/// Looks up and runs the JSON deserializer for the given type name.
/// Returns `None` if the type is not registered.
///
/// # Errors
///
/// Returns an error if metadata contains the reserved `identifier` key, the payload is missing,
/// or the deserializer fails.
pub fn deserialize_custom_from_json(
    type_name: &str,
    value: &serde_json::Value,
) -> Result<Option<Data>, anyhow::Error> {
    let reg = registries();
    let deserializer_ref = match reg.json.get(type_name) {
        Some(d) => d,
        None => return Ok(None),
    };

    let data_type = parse_data_type_from_value(value)?;
    let payload = parse_envelope_payload(value)?;
    let arc = deserializer_ref.value()(payload)?;
    let custom = match data_type {
        Some(dt) => CustomData::new(arc, dt),
        None => CustomData::from_arc(arc),
    };
    Ok(Some(Data::Custom(custom)))
}

/// Registers Arrow schema, encoder, and decoder for the given custom data type name.
///
/// # Errors
/// Returns an error if the type is already registered for Arrow.
#[cfg(feature = "arrow")]
pub fn register_arrow(
    type_name: &str,
    schema: Arc<Schema>,
    encoder: ArrowEncoder,
    decoder: ArrowDecoder,
) -> Result<(), anyhow::Error> {
    let reg = registries();
    match reg.arrow.entry(type_name.to_string()) {
        Entry::Occupied(_) => {
            anyhow::bail!("Custom data type \"{type_name}\" is already registered for Arrow");
        }
        Entry::Vacant(v) => {
            v.insert((schema, encoder, decoder));
            Ok(())
        }
    }
}

/// Registers Arrow schema, encoder, and decoder for the given custom data type name if not already
/// registered. If the type is already registered, returns `Ok(())` without overwriting (idempotent).
/// Use this where repeated registration can occur (e.g. module init).
///
/// Does not return an error (idempotent insert into `DashMap`).
#[allow(
    clippy::missing_errors_doc,
    reason = "idempotent registration only inserts into a DashMap"
)]
#[cfg(feature = "arrow")]
pub fn ensure_arrow_registered(
    type_name: &str,
    schema: Arc<Schema>,
    encoder: ArrowEncoder,
    decoder: ArrowDecoder,
) -> Result<(), anyhow::Error> {
    let reg = registries();
    reg.arrow
        .entry(type_name.to_string())
        .or_insert_with(|| (schema, encoder, decoder));
    Ok(())
}

/// Returns the Arrow schema for the given custom type name, if registered.
#[must_use]
#[cfg(feature = "arrow")]
pub fn get_arrow_schema(type_name: &str) -> Option<Arc<Schema>> {
    let reg = registries();
    reg.arrow
        .get(type_name)
        .map(|entry| Arc::clone(&entry.value().0))
}

/// Encodes a slice of custom data trait objects to a `RecordBatch` using the registered encoder.
///
/// # Errors
/// Returns an error if the type is not registered or encoding fails.
#[cfg(feature = "arrow")]
pub fn encode_custom_to_arrow(
    type_name: &str,
    items: &[Arc<dyn CustomDataTrait>],
) -> Result<Option<RecordBatch>, anyhow::Error> {
    let reg = registries();
    let entry = match reg.arrow.get(type_name) {
        Some(e) => e,
        None => return Ok(None),
    };
    let encoder = &entry.value().1;
    encoder(items).map(Some)
}

/// Decodes a `RecordBatch` into `Vec<Data>` using the registered decoder.
///
/// # Errors
/// Returns an error if the type is not registered or decoding fails.
#[expect(
    clippy::implicit_hasher,
    reason = "callers always use the default hasher"
)]
#[cfg(feature = "arrow")]
pub fn decode_custom_from_arrow(
    type_name: &str,
    metadata: &std::collections::HashMap<String, String>,
    record_batch: RecordBatch,
) -> Result<Option<Vec<Data>>, anyhow::Error> {
    let reg = registries();
    let entry = match reg.arrow.get(type_name) {
        Some(e) => e,
        None => return Ok(None),
    };
    let decoder = &entry.value().2;
    decoder(metadata, record_batch).map(Some)
}

#[cfg(feature = "python")]
pub type PyExtractor = Box<
    dyn for<'a> Fn(&pyo3::Bound<'a, pyo3::PyAny>) -> Option<Arc<dyn CustomDataTrait>> + Send + Sync,
>;

#[cfg(feature = "python")]
fn py_extractors() -> &'static DashMap<String, PyExtractor> {
    static PY_EXTRACTORS: std::sync::OnceLock<DashMap<String, PyExtractor>> =
        std::sync::OnceLock::new();
    PY_EXTRACTORS.get_or_init(DashMap::new)
}

/// Registers a `PyExtractor` for the given custom data type name.
/// Used by `CustomData` constructor to convert Python objects to `Arc<dyn CustomDataTrait>`.
///
/// # Errors
/// Returns an error if the type is already registered.
#[cfg(feature = "python")]
pub fn register_py_extractor(type_name: &str, extractor: PyExtractor) -> Result<(), anyhow::Error> {
    let reg = py_extractors();
    match reg.entry(type_name.to_string()) {
        Entry::Occupied(_) => {
            anyhow::bail!(
                "Custom data type \"{type_name}\" is already registered for Python extraction"
            );
        }
        Entry::Vacant(v) => {
            v.insert(extractor);
            Ok(())
        }
    }
}

/// Registers a `PyExtractor` for the given custom data type name if not already registered.
/// If the type is already registered, returns `Ok(())` without overwriting (idempotent).
/// Use this where repeated registration can occur (e.g. module init).
///
/// Does not return an error (idempotent insert into `DashMap`).
#[allow(
    clippy::missing_errors_doc,
    reason = "idempotent registration only inserts into a DashMap"
)]
#[cfg(feature = "python")]
pub fn ensure_py_extractor_registered(
    type_name: &str,
    extractor: PyExtractor,
) -> Result<(), anyhow::Error> {
    let reg = py_extractors();
    reg.entry(type_name.to_string())
        .or_insert_with(|| extractor);
    Ok(())
}

/// Tries to extract `Arc<dyn CustomDataTrait>` from a Python object using the registered extractor.
/// Returns None if no extractor is registered or extraction fails.
#[cfg(feature = "python")]
#[must_use]
pub fn try_extract_from_py(
    type_name: &str,
    obj: &pyo3::Bound<'_, pyo3::PyAny>,
) -> Option<Arc<dyn CustomDataTrait>> {
    let reg = py_extractors();
    let entry = reg.get(type_name)?;
    let extractor = entry.value();
    extractor(obj)
}

#[cfg(feature = "python")]
type RustExtractorFactory = Box<dyn Fn() -> PyExtractor + Send + Sync>;

#[cfg(feature = "python")]
fn rust_extractor_factories() -> &'static DashMap<String, RustExtractorFactory> {
    static RUST_EXTRACTOR_FACTORIES: std::sync::OnceLock<DashMap<String, RustExtractorFactory>> =
        std::sync::OnceLock::new();
    RUST_EXTRACTOR_FACTORIES.get_or_init(DashMap::new)
}

#[cfg(feature = "python")]
type RustClassCheck = Box<dyn Fn(&pyo3::Bound<'_, pyo3::PyAny>) -> bool + Send + Sync>;

#[cfg(feature = "python")]
fn rust_class_checks() -> &'static DashMap<String, RustClassCheck> {
    static RUST_CLASS_CHECKS: std::sync::OnceLock<DashMap<String, RustClassCheck>> =
        std::sync::OnceLock::new();
    RUST_CLASS_CHECKS.get_or_init(DashMap::new)
}

/// Registers a factory that produces a `PyExtractor` for the given type name.
/// Crates (e.g. persistence) call this at load time for each Rust custom data type.
/// When `register_custom_data_class(cls)` is called with that type's class, the factory is invoked
/// and the extractor is registered in the main `PyExtractor` registry.
///
/// # Errors
/// Returns an error if the type name is already registered.
#[cfg(feature = "python")]
pub fn register_rust_extractor_factory(
    type_name: &str,
    factory: RustExtractorFactory,
) -> Result<(), anyhow::Error> {
    let reg = rust_extractor_factories();
    match reg.entry(type_name.to_string()) {
        Entry::Occupied(_) => {
            anyhow::bail!("Rust extractor factory for \"{type_name}\" is already registered");
        }
        Entry::Vacant(v) => {
            v.insert(factory);
            Ok(())
        }
    }
}

/// Registers a factory that produces a `PyExtractor` for the given type name if not already
/// registered. If the type is already registered, returns `Ok(())` without overwriting (idempotent).
/// Use this where repeated registration can occur (e.g. module load).
///
/// Does not return an error (idempotent insert into `DashMap`).
#[allow(
    clippy::missing_errors_doc,
    reason = "idempotent registration only inserts into a DashMap"
)]
#[cfg(feature = "python")]
pub fn ensure_rust_extractor_factory_registered(
    type_name: &str,
    factory: RustExtractorFactory,
) -> Result<(), anyhow::Error> {
    let reg = rust_extractor_factories();
    reg.entry(type_name.to_string()).or_insert_with(|| factory);
    Ok(())
}

/// Registers a Rust custom data type for Python extraction. Call once per type at module load
/// (e.g. in the persistence PyO3 module). Uses [`register_rust_extractor_factory`] with a
/// factory that builds the extractor for `T`.
///
/// # Errors
/// Returns an error if the type name is already registered.
#[cfg(feature = "python")]
pub fn register_rust_extractor<T>() -> Result<(), anyhow::Error>
where
    T: CustomDataTrait
        + pyo3::PyTypeInfo
        + for<'a, 'py> pyo3::FromPyObject<'a, 'py>
        + Send
        + Sync
        + 'static,
{
    let type_name = T::type_name_static();
    ensure_rust_class_registered::<T>(type_name);

    let factory: RustExtractorFactory = Box::new(|| {
        Box::new(|obj: &pyo3::Bound<'_, pyo3::PyAny>| {
            obj.extract::<T>()
                .ok()
                .map(|x| Arc::new(x) as Arc<dyn CustomDataTrait>)
        })
    });
    register_rust_extractor_factory(type_name, factory)
}

/// Registers a Rust custom data type for Python extraction if not already registered.
/// If the type is already registered, returns `Ok(())` without overwriting (idempotent).
/// Use this where repeated registration can occur (e.g. module load).
///
/// Does not return an error (idempotent insert into `DashMap`).
#[allow(
    clippy::missing_errors_doc,
    reason = "idempotent registration only inserts into a DashMap"
)]
#[cfg(feature = "python")]
pub fn ensure_rust_extractor_registered<T>() -> Result<(), anyhow::Error>
where
    T: CustomDataTrait
        + pyo3::PyTypeInfo
        + for<'a, 'py> pyo3::FromPyObject<'a, 'py>
        + Send
        + Sync
        + 'static,
{
    let type_name = T::type_name_static();
    ensure_rust_class_registered::<T>(type_name);

    let factory: RustExtractorFactory = Box::new(|| {
        Box::new(|obj: &pyo3::Bound<'_, pyo3::PyAny>| {
            obj.extract::<T>()
                .ok()
                .map(|x| Arc::new(x) as Arc<dyn CustomDataTrait>)
        })
    });
    ensure_rust_extractor_factory_registered(type_name, factory)
}

/// Calls the registered factory for the given type name and returns the extractor, if any.
#[cfg(feature = "python")]
#[must_use]
pub fn get_rust_extractor(type_name: &str) -> Option<PyExtractor> {
    let reg = rust_extractor_factories();
    let factory_ref = reg.get(type_name)?;
    Some(factory_ref.value()())
}

// Returns `true` when a Rust type registered with a different PyO3 class uses `type_name`
#[cfg(feature = "python")]
#[must_use]
pub(crate) fn conflicts_with_rust_class(
    type_name: &str,
    data_class: &pyo3::Bound<'_, pyo3::PyAny>,
) -> bool {
    rust_class_checks()
        .get(type_name)
        .is_some_and(|check| !check.value()(data_class))
}

#[cfg(feature = "python")]
fn ensure_rust_class_registered<T: pyo3::PyTypeInfo>(type_name: &str) {
    rust_class_checks()
        .entry(type_name.to_string())
        .or_insert_with(|| Box::new(|data_class| data_class.is(T::type_object(data_class.py()))));
}

#[cfg(test)]
mod tests {
    use std::{
        hash::{DefaultHasher, Hash, Hasher},
        sync::atomic::{AtomicBool, Ordering},
    };

    use nautilus_core::UnixNanos;
    use rstest::rstest;
    use serde::{Deserialize, Serialize};

    use super::*;
    use crate::data::{
        CustomData,
        custom::{ensure_custom_data_json_registered, register_custom_data_json},
    };

    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
    struct TestRegCustomData {
        ts_init: UnixNanos,
    }

    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
    #[serde(deny_unknown_fields)]
    struct StrictRegCustomData {
        ts_init: UnixNanos,
    }

    impl crate::data::HasTsInit for TestRegCustomData {
        fn ts_init(&self) -> UnixNanos {
            self.ts_init
        }
    }

    impl crate::data::custom::CustomDataTrait for TestRegCustomData {
        fn type_name(&self) -> &'static str {
            "TestRegCustomData"
        }
        fn type_name_static() -> &'static str {
            "TestRegCustomData"
        }
        fn as_any(&self) -> &dyn std::any::Any {
            self
        }
        fn ts_event(&self) -> nautilus_core::UnixNanos {
            self.ts_init
        }
        fn to_json(&self) -> anyhow::Result<String> {
            Ok(serde_json::to_string(self)?)
        }
        fn clone_arc(&self) -> Arc<dyn crate::data::CustomDataTrait> {
            Arc::new(self.clone())
        }
        fn eq_arc(&self, other: &dyn crate::data::CustomDataTrait) -> bool {
            other.as_any().downcast_ref::<Self>() == Some(self)
        }
        fn from_json(
            value: serde_json::Value,
        ) -> anyhow::Result<Arc<dyn crate::data::CustomDataTrait>> {
            let t: Self = serde_json::from_value(value)?;
            Ok(Arc::new(t))
        }
    }

    impl crate::data::HasTsInit for StrictRegCustomData {
        fn ts_init(&self) -> UnixNanos {
            self.ts_init
        }
    }

    impl crate::data::custom::CustomDataTrait for StrictRegCustomData {
        fn type_name(&self) -> &'static str {
            "StrictRegCustomData"
        }
        fn type_name_static() -> &'static str {
            "StrictRegCustomData"
        }
        fn as_any(&self) -> &dyn std::any::Any {
            self
        }
        fn ts_event(&self) -> nautilus_core::UnixNanos {
            self.ts_init
        }
        fn to_json(&self) -> anyhow::Result<String> {
            Ok(serde_json::to_string(self)?)
        }
        fn clone_arc(&self) -> Arc<dyn crate::data::CustomDataTrait> {
            Arc::new(self.clone())
        }
        fn eq_arc(&self, other: &dyn crate::data::CustomDataTrait) -> bool {
            other.as_any().downcast_ref::<Self>() == Some(self)
        }
        fn from_json(
            value: serde_json::Value,
        ) -> anyhow::Result<Arc<dyn crate::data::CustomDataTrait>> {
            let t: Self = serde_json::from_value(value)?;
            Ok(Arc::new(t))
        }
    }

    #[rstest]
    fn json_registry_roundtrip() {
        let _ = register_custom_data_json::<TestRegCustomData>();

        let data = Data::Custom(CustomData::from_arc(Arc::new(TestRegCustomData {
            ts_init: UnixNanos::from(100),
        })));

        let json = serde_json::to_string(&data).unwrap();
        let back: Data = serde_json::from_str(&json).unwrap();

        match (&data, &back) {
            (Data::Custom(a), Data::Custom(b)) => {
                assert_eq!(a.data.type_name(), b.data.type_name());
                assert_eq!(a.data.ts_init(), b.data.ts_init());
            }
            _ => panic!("expected Custom variant"),
        }
    }

    #[rstest]
    fn json_registry_roundtrip_with_deny_unknown_fields() {
        let _ = register_custom_data_json::<StrictRegCustomData>();

        let data = Data::Custom(CustomData::from_arc(Arc::new(StrictRegCustomData {
            ts_init: UnixNanos::from(200),
        })));

        let json = serde_json::to_string(&data).unwrap();
        let back: Data = serde_json::from_str(&json).unwrap();

        match (&data, &back) {
            (Data::Custom(a), Data::Custom(b)) => {
                assert_eq!(a.data.type_name(), b.data.type_name());
                assert_eq!(a.data.ts_init(), b.data.ts_init());
            }
            _ => panic!("expected Custom variant"),
        }
    }

    #[rstest]
    #[case("ReservedMetadataString", serde_json::json!("shadow"))]
    #[case("ReservedMetadataNull", serde_json::Value::Null)]
    fn json_registry_rejects_reserved_metadata_before_callback(
        #[case] type_name: &str,
        #[case] value: serde_json::Value,
    ) {
        let called = Arc::new(AtomicBool::new(false));
        let called_by_callback = called.clone();
        register_json_deserializer(
            type_name,
            Box::new(move |_| {
                called_by_callback.store(true, Ordering::SeqCst);
                anyhow::bail!("invalid metadata reached callback")
            }),
        )
        .unwrap();

        let custom = CustomData::from_arc(Arc::new(TestRegCustomData {
            ts_init: UnixNanos::from(317),
        }));

        let mut envelope = serde_json::to_value(custom).unwrap();
        envelope["type"] = serde_json::json!(type_name);
        envelope["data_type"]["type_name"] = serde_json::json!(type_name);
        envelope["data_type"]["metadata"]["identifier"] = value;

        let error = deserialize_custom_from_json(type_name, &envelope).unwrap_err();

        assert_eq!(
            error.to_string(),
            "`identifier` is reserved for DataType catalog identity"
        );
        assert!(!called.load(Ordering::SeqCst));
    }

    #[rstest]
    #[case::missing(None)]
    #[case::null(Some(serde_json::Value::Null))]
    #[case::empty(Some(serde_json::json!({})))]
    #[case::populated(Some(serde_json::json!({"source": "feed", "level": 3})))]
    fn json_registry_preserves_envelope_metadata(#[case] metadata: Option<serde_json::Value>) {
        ensure_custom_data_json_registered::<StrictRegCustomData>().unwrap();
        let expected_metadata: Option<Params> = metadata
            .clone()
            .filter(|v| !v.is_null() && v.as_object().is_some_and(|m| !m.is_empty()))
            .map(|v| serde_json::from_value(v).unwrap());

        let data_type = DataType::new(
            "StrictRegCustomData",
            expected_metadata.clone(),
            Some("catalog/317".to_string()),
        );

        let expected = CustomData::new(
            Arc::new(StrictRegCustomData {
                ts_init: UnixNanos::from(317),
            }),
            data_type,
        );

        let mut envelope = serde_json::to_value(&expected).unwrap();
        if let Some(metadata) = metadata {
            envelope["data_type"]["metadata"] = metadata;
        } else {
            envelope["data_type"]
                .as_object_mut()
                .unwrap()
                .remove("metadata");
        }

        let actual = deserialize_custom_from_json("StrictRegCustomData", &envelope)
            .unwrap()
            .unwrap();

        let Data::Custom(actual) = actual else {
            panic!("expected Custom variant");
        };

        assert_eq!(
            serde_json::to_value(&actual).unwrap(),
            serde_json::to_value(&expected).unwrap()
        );
        assert_eq!(actual.data_type.metadata(), expected_metadata.as_ref());
        assert_eq!(actual.data_type.identifier(), Some("catalog/317"));
    }

    #[rstest]
    #[case::missing(None)]
    #[case::null(Some(serde_json::Value::Null))]
    #[case::missing_type_name(Some(serde_json::json!({"metadata": {"source": "feed"}})))]
    fn json_registry_preserves_missing_descriptor_fallback(
        #[case] descriptor: Option<serde_json::Value>,
    ) {
        ensure_custom_data_json_registered::<StrictRegCustomData>().unwrap();

        let expected = CustomData::from_arc(Arc::new(StrictRegCustomData {
            ts_init: UnixNanos::from(317),
        }));

        let mut envelope = serde_json::to_value(&expected).unwrap();
        if let Some(descriptor) = descriptor {
            envelope["data_type"] = descriptor;
        } else {
            envelope.as_object_mut().unwrap().remove("data_type");
        }

        let actual = deserialize_custom_from_json("StrictRegCustomData", &envelope)
            .unwrap()
            .unwrap();

        assert_eq!(
            serde_json::to_value(actual).unwrap(),
            serde_json::to_value(expected).unwrap()
        );
    }

    #[rstest]
    fn data_type_registry_result_hashes_like_equal_values_from_all_routes() {
        fn hash_data_type(data_type: &DataType) -> u64 {
            let mut hasher = DefaultHasher::new();
            data_type.hash(&mut hasher);
            hasher.finish()
        }

        let metadata = serde_json::json!({"key": "value"});
        let constructed = DataType::new(
            "ExampleType",
            Some(serde_json::from_value(metadata.clone()).unwrap()),
            Some("catalog/path".to_string()),
        );
        let persistence_json = serde_json::json!({
            "type_name": constructed.type_name(),
            "metadata": metadata,
            "identifier": constructed.identifier(),
        });
        let persisted = DataType::from_persistence_json(&persistence_json.to_string()).unwrap();
        let registry_envelope = serde_json::json!({"data_type": persistence_json});
        let registered = parse_data_type_from_value(&registry_envelope)
            .unwrap()
            .unwrap();
        let deserialization_payload = serde_json::json!({
            "type_name": constructed.type_name(),
            "metadata": constructed.metadata(),
            "topic": constructed.topic(),
            "hash": constructed.precomputed_hash() ^ u64::MAX,
            "identifier": constructed.identifier(),
        });
        let deserialized: DataType = serde_json::from_value(deserialization_payload).unwrap();

        // Eq and Hash both observe `topic` alone, so they cannot detect a route that drops or
        // rewrites the other fields. Assert those explicitly as well, or a registry regression
        // that lost `identifier` would still satisfy this test.
        for data_type in [&deserialized, &persisted, &registered] {
            assert_eq!(data_type, &constructed);
            assert_eq!(hash_data_type(data_type), hash_data_type(&constructed));
            assert_eq!(data_type.type_name(), constructed.type_name());
            assert_eq!(data_type.metadata(), constructed.metadata());
            assert_eq!(data_type.identifier(), constructed.identifier());
            assert_eq!(data_type.topic(), constructed.topic());
        }
    }

    #[rstest]
    fn ensure_json_deserializer_registered_is_idempotent() {
        let deserializer: JsonDeserializer = Box::new(|value| {
            let t: TestRegCustomData = serde_json::from_value(value)?;
            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
        });
        let r1 = ensure_json_deserializer_registered("IdempotentTestJson", deserializer);
        assert!(r1.is_ok(), "first registration should succeed");
        let deserializer2: JsonDeserializer = Box::new(|value| {
            let t: TestRegCustomData = serde_json::from_value(value)?;
            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
        });
        let r2 = ensure_json_deserializer_registered("IdempotentTestJson", deserializer2);
        assert!(
            r2.is_ok(),
            "second registration with same type_name should succeed (idempotent)"
        );
    }

    #[rstest]
    fn register_json_deserializer_fails_on_duplicate() {
        let deserializer: JsonDeserializer = Box::new(|value| {
            let t: TestRegCustomData = serde_json::from_value(value)?;
            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
        });
        let r1 = register_json_deserializer("StrictDuplicateTestJson", deserializer);
        assert!(r1.is_ok());
        let deserializer2: JsonDeserializer = Box::new(|value| {
            let t: TestRegCustomData = serde_json::from_value(value)?;
            Ok(Arc::new(t) as Arc<dyn crate::data::CustomDataTrait>)
        });
        let r2 = register_json_deserializer("StrictDuplicateTestJson", deserializer2);
        assert!(r2.is_err());
        let err_msg = r2.unwrap_err().to_string();
        assert!(
            err_msg.contains("already registered"),
            "expected 'already registered' in error, found: {err_msg}"
        );
    }

    #[rstest]
    #[cfg(feature = "arrow")]
    fn ensure_arrow_registered_is_idempotent() {
        let schema = Arc::new(arrow::datatypes::Schema::empty());
        let encoder: ArrowEncoder = Box::new(|_| {
            Ok(arrow::record_batch::RecordBatch::new_empty(Arc::new(
                arrow::datatypes::Schema::empty(),
            )))
        });
        let decoder: ArrowDecoder = Box::new(|_, _| Ok(Vec::new()));

        let r1 = ensure_arrow_registered("IdempotentTestArrow", schema, encoder, decoder);
        assert!(r1.is_ok(), "first Arrow registration should succeed");

        let schema2 = Arc::new(arrow::datatypes::Schema::empty());
        let encoder2: ArrowEncoder = Box::new(|_| {
            Ok(arrow::record_batch::RecordBatch::new_empty(Arc::new(
                arrow::datatypes::Schema::empty(),
            )))
        });
        let decoder2: ArrowDecoder = Box::new(|_, _| Ok(Vec::new()));

        let r2 = ensure_arrow_registered("IdempotentTestArrow", schema2, encoder2, decoder2);
        assert!(
            r2.is_ok(),
            "second Arrow registration with same type_name should be idempotent"
        );
    }

    #[rstest]
    #[cfg(feature = "arrow")]
    fn python_custom_arrow_schema_rejects_opaque_bytes() {
        let schema = Schema::new(vec![Field::new("payload", ArrowDataType::Binary, false)]);

        let error = validate_custom_arrow_schema("PythonPayload", &schema, false).unwrap_err();

        assert_eq!(
            error.to_string(),
            "custom write schema `PythonPayload` contains opaque byte field `payload`: Binary",
        );
    }
}