nautilus_model/data/
data_type.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{
17    collections::HashMap,
18    fmt::{Debug, Display, Write as _},
19    hash::{Hash, Hasher},
20    str::FromStr,
21};
22
23use indexmap::IndexMap;
24use nautilus_core::{Params, UnixNanos};
25use serde::{
26    Deserialize, Serialize,
27    de::{self, IgnoredAny, MapAccess, SeqAccess, Visitor},
28};
29use serde_json::Value as JsonValue;
30
31use crate::identifiers::{InstrumentId, Venue};
32
33/// Separates a custom data topic from its identifier.
34pub const IDENTIFIER_TOPIC_SUFFIX: &str = ".identifier=";
35
36/// Prefix reserved for metadata strings that would otherwise parse as JSON topic values.
37const STRING_TOPIC_PREFIX: &str = "~s~";
38const ESCAPED_STRING_TOPIC_PREFIX: &str = "~x~";
39
40/// Builds a topic value while preserving the distinction between JSON strings and scalar values.
41fn value_to_topic_string(v: &JsonValue) -> String {
42    if let Some(s) = v.as_str() {
43        let has_pair_shaped_segment = s.split('.').skip(1).any(|segment| segment.contains('='));
44        if has_pair_shaped_segment || s.starts_with(ESCAPED_STRING_TOPIC_PREFIX) {
45            let mut escaped =
46                String::with_capacity(ESCAPED_STRING_TOPIC_PREFIX.len() + s.len() * 2);
47            escaped.push_str(ESCAPED_STRING_TOPIC_PREFIX);
48
49            for byte in s.as_bytes() {
50                write!(escaped, "{byte:02x}").expect("writing to String cannot fail");
51            }
52            return escaped;
53        }
54
55        if serde_json::from_str::<JsonValue>(s).is_ok() || s.starts_with(STRING_TOPIC_PREFIX) {
56            return format!("{STRING_TOPIC_PREFIX}{s}");
57        }
58        return s.to_string();
59    }
60
61    if let Some(n) = v.as_u64() {
62        return n.to_string();
63    }
64
65    if let Some(n) = v.as_i64() {
66        return n.to_string();
67    }
68
69    if let Some(b) = v.as_bool() {
70        return b.to_string();
71    }
72
73    if let Some(f) = v.as_f64() {
74        let value = f.to_string();
75        return if value.contains('.') || value.contains('e') || value.contains('E') {
76            value
77        } else {
78            format!("{value}.0")
79        };
80    }
81
82    if v.is_null() {
83        return "null".to_string();
84    }
85    serde_json::to_string(v).unwrap_or_default()
86}
87
88/// Decodes one reserved string prefix before attempting JSON topic-value parsing.
89fn topic_string_to_value(s: &str) -> anyhow::Result<JsonValue> {
90    if let Some(value) = s.strip_prefix(ESCAPED_STRING_TOPIC_PREFIX) {
91        anyhow::ensure!(
92            value.len().is_multiple_of(2) && value.bytes().all(|byte| byte.is_ascii_hexdigit()),
93            "Invalid escaped topic string"
94        );
95        let bytes = (0..value.len())
96            .step_by(2)
97            .map(|index| u8::from_str_radix(&value[index..index + 2], 16))
98            .collect::<Result<Vec<_>, _>>()?;
99        return Ok(JsonValue::String(String::from_utf8(bytes)?));
100    }
101
102    if let Some(value) = s.strip_prefix(STRING_TOPIC_PREFIX) {
103        return Ok(JsonValue::String(value.to_string()));
104    }
105
106    Ok(serde_json::from_str(s).unwrap_or_else(|_| JsonValue::String(s.to_string())))
107}
108
109fn parse_topic_metadata(metadata_topic: &str) -> anyhow::Result<Params> {
110    let mut metadata = IndexMap::new();
111    if metadata_topic.is_empty() {
112        return Ok(Params::new(metadata));
113    }
114
115    // Values may contain dots: a segment with '=' starts a new pair, otherwise it
116    // extends the current value (or the first key before any pair has started).
117    let mut pending_key: Option<String> = None;
118    let mut current: Option<(String, String)> = None;
119
120    for segment in metadata_topic.split('.') {
121        if let Some((key, value)) = segment.split_once('=') {
122            let key = if let Some(prefix) = pending_key.take() {
123                format!("{prefix}.{key}")
124            } else {
125                if key.is_empty() {
126                    anyhow::bail!("Invalid empty metadata topic key");
127                }
128                key.to_string()
129            };
130
131            if let Some((prev_key, prev_value)) = current.replace((key, value.to_string())) {
132                metadata.insert(prev_key, topic_string_to_value(&prev_value)?);
133            }
134        } else if let Some((_, value)) = current.as_mut() {
135            value.push('.');
136            value.push_str(segment);
137        } else if let Some(prefix) = pending_key.as_mut() {
138            prefix.push('.');
139            prefix.push_str(segment);
140        } else {
141            pending_key = Some(segment.to_string());
142        }
143    }
144
145    let Some((key, value)) = current else {
146        anyhow::bail!("Invalid metadata topic pair: {metadata_topic}");
147    };
148    metadata.insert(key, topic_string_to_value(&value)?);
149
150    Ok(Params::new(metadata))
151}
152
153/// Builds the topic suffix from Params (string-only view: key=value joined by ".").
154fn params_to_topic_suffix(params: &Params) -> String {
155    let mut entries = params.iter().collect::<Vec<_>>();
156    entries.sort_by_key(|(key, _)| *key);
157
158    entries
159        .into_iter()
160        .map(|(k, v)| format!("{k}={}", value_to_topic_string(v)))
161        .collect::<Vec<_>>()
162        .join(".")
163}
164
165/// Represents a data type including metadata.
166#[derive(Clone, Serialize)]
167#[cfg_attr(
168    feature = "python",
169    pyo3::pyclass(module = "nautilus_trader.model", from_py_object)
170)]
171#[cfg_attr(
172    feature = "python",
173    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.model")
174)]
175pub struct DataType {
176    type_name: String,
177    metadata: Option<Params>,
178    topic: String,
179    hash: u64,
180    identifier: Option<String>,
181}
182
183impl<'de> Deserialize<'de> for DataType {
184    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
185    where
186        D: serde::Deserializer<'de>,
187    {
188        const FIELDS: &[&str] = &["type_name", "metadata", "topic", "hash", "identifier"];
189
190        #[derive(Deserialize)]
191        #[serde(field_identifier, rename_all = "snake_case")]
192        enum Field {
193            TypeName,
194            Metadata,
195            Topic,
196            Hash,
197            Identifier,
198            #[serde(other)]
199            Other,
200        }
201
202        fn finish(
203            type_name: &str,
204            metadata: Option<Params>,
205            topic: Option<String>,
206            identifier: Option<String>,
207        ) -> anyhow::Result<DataType> {
208            let mut data_type = DataType::try_new(type_name, metadata, identifier)?;
209            if let Some(topic) = topic {
210                data_type.hash = calculate_hash(&topic);
211                data_type.topic = topic;
212            }
213            Ok(data_type)
214        }
215
216        struct DataTypeVisitor;
217
218        impl<'de> Visitor<'de> for DataTypeVisitor {
219            type Value = DataType;
220
221            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
222                formatter.write_str("struct DataType")
223            }
224
225            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
226            where
227                A: SeqAccess<'de>,
228            {
229                let type_name: String = seq
230                    .next_element()?
231                    .ok_or_else(|| de::Error::invalid_length(0, &self))?;
232                let metadata = seq
233                    .next_element()?
234                    .ok_or_else(|| de::Error::invalid_length(1, &self))?;
235                let topic = seq
236                    .next_element()?
237                    .ok_or_else(|| de::Error::invalid_length(2, &self))?;
238                let _hash: u64 = seq
239                    .next_element()?
240                    .ok_or_else(|| de::Error::invalid_length(3, &self))?;
241                let identifier = seq
242                    .next_element()?
243                    .ok_or_else(|| de::Error::invalid_length(4, &self))?;
244
245                finish(&type_name, metadata, Some(topic), identifier).map_err(de::Error::custom)
246            }
247
248            fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
249            where
250                A: MapAccess<'de>,
251            {
252                let mut type_name: Option<String> = None;
253                let mut metadata: Option<Option<Params>> = None;
254                let mut topic: Option<Option<String>> = None;
255                let mut hash_seen = false;
256                let mut identifier: Option<Option<String>> = None;
257
258                while let Some(key) = map.next_key()? {
259                    match key {
260                        Field::TypeName => {
261                            if type_name.is_some() {
262                                return Err(de::Error::duplicate_field("type_name"));
263                            }
264                            type_name = Some(map.next_value()?);
265                        }
266                        Field::Metadata => {
267                            if metadata.is_some() {
268                                return Err(de::Error::duplicate_field("metadata"));
269                            }
270                            metadata = Some(map.next_value()?);
271                        }
272                        Field::Topic => {
273                            if topic.is_some() {
274                                return Err(de::Error::duplicate_field("topic"));
275                            }
276                            topic = Some(map.next_value()?);
277                        }
278                        Field::Hash => {
279                            if hash_seen {
280                                return Err(de::Error::duplicate_field("hash"));
281                            }
282                            hash_seen = true;
283                            let _: Option<u64> = map.next_value()?;
284                        }
285                        Field::Identifier => {
286                            if identifier.is_some() {
287                                return Err(de::Error::duplicate_field("identifier"));
288                            }
289                            identifier = Some(map.next_value()?);
290                        }
291                        Field::Other => {
292                            let _: IgnoredAny = map.next_value()?;
293                        }
294                    }
295                }
296
297                let type_name = type_name.ok_or_else(|| de::Error::missing_field("type_name"))?;
298                finish(
299                    &type_name,
300                    metadata.unwrap_or(None),
301                    topic.unwrap_or(None),
302                    identifier.unwrap_or(None),
303                )
304                .map_err(de::Error::custom)
305            }
306        }
307
308        deserializer.deserialize_struct("DataType", FIELDS, DataTypeVisitor)
309    }
310}
311
312fn calculate_hash(topic: &str) -> u64 {
313    let mut hasher = std::collections::hash_map::DefaultHasher::new();
314    topic.hash(&mut hasher);
315    hasher.finish()
316}
317
318impl DataType {
319    /// Creates a new [`DataType`] instance.
320    ///
321    /// # Panics
322    ///
323    /// Panics if `metadata` contains the reserved `identifier` key.
324    #[must_use]
325    pub fn new(type_name: &str, metadata: Option<Params>, identifier: Option<String>) -> Self {
326        Self::try_new(type_name, metadata, identifier)
327            .expect("DataType metadata must not contain the reserved `identifier` key")
328    }
329
330    /// Creates a new [`DataType`] instance after validating reserved metadata keys.
331    ///
332    /// # Errors
333    ///
334    /// Returns an error if `metadata` contains the reserved `identifier` key.
335    pub fn try_new(
336        type_name: &str,
337        metadata: Option<Params>,
338        identifier: Option<String>,
339    ) -> anyhow::Result<Self> {
340        anyhow::ensure!(
341            metadata
342                .as_ref()
343                .is_none_or(|metadata| !metadata.contains_key("identifier")),
344            "`identifier` is reserved for DataType catalog identity",
345        );
346        let base_topic = if let Some(ref meta) = metadata {
347            if meta.is_empty() {
348                type_name.to_string()
349            } else {
350                format!("{type_name}.{}", params_to_topic_suffix(meta))
351            }
352        } else {
353            type_name.to_string()
354        };
355        let topic = identifier
356            .as_ref()
357            .map_or(base_topic.clone(), |identifier| {
358                format!("{base_topic}{IDENTIFIER_TOPIC_SUFFIX}{identifier}")
359            });
360
361        let hash = calculate_hash(&topic);
362
363        Ok(Self {
364            type_name: type_name.to_owned(),
365            metadata,
366            topic,
367            hash,
368            identifier,
369        })
370    }
371
372    /// Serializes to JSON for persistence (`type_name`, metadata, identifier; no topic, no hash).
373    ///
374    /// # Errors
375    ///
376    /// Returns an error if the data cannot be serialized.
377    pub fn to_persistence_json(&self) -> Result<String, serde_json::Error> {
378        let mut map = serde_json::Map::new();
379        map.insert(
380            "type_name".to_string(),
381            serde_json::Value::String(self.type_name.clone()),
382        );
383        map.insert(
384            "metadata".to_string(),
385            self.metadata.as_ref().map_or(serde_json::Value::Null, |m| {
386                serde_json::to_value(m).unwrap_or(serde_json::Value::Null)
387            }),
388        );
389
390        if let Some(ref id) = self.identifier {
391            map.insert(
392                "identifier".to_string(),
393                serde_json::Value::String(id.clone()),
394            );
395        }
396        serde_json::to_string(&serde_json::Value::Object(map))
397    }
398
399    /// Deserializes from JSON produced by `to_persistence_json`.
400    /// Accepts legacy JSON with `topic` (ignored); topic is rebuilt from `type_name` + metadata.
401    ///
402    /// # Errors
403    ///
404    /// Returns an error if the string is not valid JSON or missing required fields.
405    pub fn from_persistence_json(s: &str) -> Result<Self, anyhow::Error> {
406        let value: serde_json::Value =
407            serde_json::from_str(s).map_err(|e| anyhow::anyhow!("Invalid data_type JSON: {e}"))?;
408        let obj = value
409            .as_object()
410            .ok_or_else(|| anyhow::anyhow!("data_type must be a JSON object"))?;
411        let type_name = obj
412            .get("type_name")
413            .and_then(|v| v.as_str())
414            .ok_or_else(|| anyhow::anyhow!("data_type must have type_name"))?
415            .to_string();
416        let metadata = obj.get("metadata").and_then(|m| {
417            if m.is_null() {
418                None
419            } else {
420                let p: Params = serde_json::from_value(m.clone()).ok()?;
421                if p.is_empty() { None } else { Some(p) }
422            }
423        });
424        let identifier = obj
425            .get("identifier")
426            .and_then(|v| v.as_str())
427            .map(String::from);
428        Self::try_new(&type_name, metadata, identifier)
429    }
430
431    /// Returns the type name for the data type.
432    #[must_use]
433    pub fn type_name(&self) -> &str {
434        self.type_name.as_str()
435    }
436
437    /// Returns the metadata for the data type.
438    #[must_use]
439    pub fn metadata(&self) -> Option<&Params> {
440        self.metadata.as_ref()
441    }
442
443    /// Returns a string representation of the metadata.
444    #[must_use]
445    pub fn metadata_str(&self) -> String {
446        self.metadata.as_ref().map_or_else(
447            || "null".to_string(),
448            |metadata| {
449                let mut entries = metadata.iter().collect::<Vec<_>>();
450                entries.sort_by_key(|(key, _)| *key);
451
452                let mut metadata_map = serde_json::Map::new();
453                for (key, value) in entries {
454                    metadata_map.insert(key.clone(), value.clone());
455                }
456
457                serde_json::to_string(&metadata_map).unwrap_or_default()
458            },
459        )
460    }
461
462    /// Returns metadata as a string-only map (e.g. for Arrow schema metadata).
463    #[must_use]
464    pub fn metadata_string_map(&self) -> Option<HashMap<String, String>> {
465        self.metadata.as_ref().map(|p| {
466            p.iter()
467                .map(|(k, v)| (k.clone(), value_to_topic_string(v)))
468                .collect()
469        })
470    }
471
472    /// Returns the precomputed hash for this data type.
473    #[must_use]
474    pub fn precomputed_hash(&self) -> u64 {
475        self.hash
476    }
477
478    /// Returns the messaging topic for the data type.
479    #[must_use]
480    pub fn topic(&self) -> &str {
481        self.topic.as_str()
482    }
483
484    /// Returns the optional catalog path identifier.
485    #[must_use]
486    pub fn identifier(&self) -> Option<&str> {
487        self.identifier.as_deref()
488    }
489
490    /// Returns an optional [`InstrumentId`] parsed from the metadata.
491    ///
492    /// # Errors
493    ///
494    /// Returns an error if the `instrument_id` metadata value is invalid.
495    pub fn instrument_id(&self) -> anyhow::Result<Option<InstrumentId>> {
496        let Some(instrument_id) = self
497            .metadata
498            .as_ref()
499            .and_then(|metadata| metadata.get_str("instrument_id"))
500        else {
501            return Ok(None);
502        };
503
504        InstrumentId::from_str(instrument_id)
505            .map(Some)
506            .map_err(|e| anyhow::anyhow!("Invalid instrument_id metadata `{instrument_id}`: {e}"))
507    }
508
509    /// Returns an optional [`Venue`] parsed from the metadata.
510    ///
511    /// # Errors
512    ///
513    /// Returns an error if the `venue` metadata value is invalid.
514    pub fn venue(&self) -> anyhow::Result<Option<Venue>> {
515        let Some(venue) = self
516            .metadata
517            .as_ref()
518            .and_then(|metadata| metadata.get_str("venue"))
519        else {
520            return Ok(None);
521        };
522
523        Venue::new_checked(venue)
524            .map(Some)
525            .map_err(|e| anyhow::anyhow!("Invalid venue metadata `{venue}`: {e}"))
526    }
527
528    /// Returns an optional [`UnixNanos`] parsed from the metadata `start` field.
529    ///
530    /// # Errors
531    ///
532    /// Returns an error if the `start` metadata value is invalid.
533    pub fn start(&self) -> anyhow::Result<Option<UnixNanos>> {
534        self.parse_unix_nanos("start")
535    }
536
537    /// Returns an optional [`UnixNanos`] parsed from the metadata `end` field.
538    ///
539    /// # Errors
540    ///
541    /// Returns an error if the `end` metadata value is invalid.
542    pub fn end(&self) -> anyhow::Result<Option<UnixNanos>> {
543        self.parse_unix_nanos("end")
544    }
545
546    /// Returns an optional row limit parsed from the metadata.
547    ///
548    /// # Errors
549    ///
550    /// Returns an error if the `limit` metadata value is invalid.
551    pub fn limit(&self) -> anyhow::Result<Option<usize>> {
552        let Some(metadata) = self.metadata.as_ref() else {
553            return Ok(None);
554        };
555
556        if let Some(limit) = metadata.get_usize("limit") {
557            return Ok(Some(limit));
558        }
559        let Some(limit) = metadata.get_str("limit") else {
560            return Ok(None);
561        };
562
563        limit
564            .parse::<usize>()
565            .map(Some)
566            .map_err(|e| anyhow::anyhow!("Invalid limit metadata `{limit}`: {e}"))
567    }
568
569    fn parse_unix_nanos(&self, key: &str) -> anyhow::Result<Option<UnixNanos>> {
570        let Some(value) = self
571            .metadata
572            .as_ref()
573            .and_then(|metadata| metadata.get_str(key))
574        else {
575            return Ok(None);
576        };
577
578        UnixNanos::from_str(value)
579            .map(Some)
580            .map_err(|e| anyhow::anyhow!("Invalid {key} metadata `{value}`: {e}"))
581    }
582}
583
584impl PartialEq for DataType {
585    fn eq(&self, other: &Self) -> bool {
586        self.topic == other.topic
587    }
588}
589
590impl Eq for DataType {}
591
592impl PartialOrd for DataType {
593    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
594        Some(self.cmp(other))
595    }
596}
597
598impl Ord for DataType {
599    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
600        self.topic.cmp(&other.topic)
601    }
602}
603
604impl Hash for DataType {
605    fn hash<H: Hasher>(&self, state: &mut H) {
606        self.hash.hash(state);
607    }
608}
609
610impl FromStr for DataType {
611    type Err = anyhow::Error;
612
613    fn from_str(s: &str) -> Result<Self, Self::Err> {
614        let (base_topic, identifier) = s
615            .rsplit_once(IDENTIFIER_TOPIC_SUFFIX)
616            .map_or((s, None), |(base, identifier)| {
617                (base, Some(identifier.to_string()))
618            });
619        let (type_name, metadata) = match base_topic.split_once('.') {
620            None => (base_topic, None),
621            Some((type_name, metadata_topic)) => {
622                let metadata = parse_topic_metadata(metadata_topic)?;
623                (type_name, Some(metadata))
624            }
625        };
626
627        Self::try_new(type_name, metadata, identifier)
628    }
629}
630
631impl Display for DataType {
632    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
633        write!(f, "{}", self.topic)
634    }
635}
636
637impl Debug for DataType {
638    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
639        write!(
640            f,
641            "DataType(type_name={}, metadata={:?}, identifier={:?})",
642            self.type_name, self.metadata, self.identifier
643        )
644    }
645}
646
647#[cfg(test)]
648mod tests {
649    use std::{
650        collections::hash_map::DefaultHasher,
651        hash::{Hash, Hasher},
652    };
653
654    use rstest::*;
655    use serde_json::json;
656
657    use super::*;
658
659    fn params_from_json(value: serde_json::Value) -> Params {
660        serde_json::from_value(value).expect("valid Params JSON")
661    }
662
663    #[rstest]
664    #[case("aéa")]
665    #[case("éaa")]
666    #[case("zz")]
667    #[case("0")]
668    fn test_from_str_rejects_invalid_escaped_metadata(#[case] value: &str) {
669        let topic = format!("Example.value=~x~{value}");
670        let error = DataType::from_str(&topic).unwrap_err();
671
672        assert_eq!(error.to_string(), "Invalid escaped topic string");
673    }
674
675    #[rstest]
676    #[case("Example.identifier=A.identifier=B")]
677    #[case("Example.value=1.identifier=A.identifier=B")]
678    #[case("Example.identifier=.identifier=B")]
679    fn test_from_str_rejects_repeated_identifier(#[case] topic: &str) {
680        let error = DataType::from_str(topic).unwrap_err();
681
682        assert_eq!(
683            error.to_string(),
684            "`identifier` is reserved for DataType catalog identity"
685        );
686    }
687
688    #[rstest]
689    fn test_data_type_creation_with_metadata() {
690        let metadata = Some(params_from_json(
691            json!({"key1": "value1", "key2": "value2"}),
692        ));
693        let data_type = DataType::new("ExampleType", metadata.clone(), None);
694
695        assert_eq!(data_type.type_name(), "ExampleType");
696        assert_eq!(data_type.topic(), "ExampleType.key1=value1.key2=value2");
697        assert_eq!(data_type.metadata(), metadata.as_ref());
698    }
699
700    #[rstest]
701    fn test_data_type_topic_roundtrip_escapes_ambiguous_metadata_string() {
702        let metadata = Some(params_from_json(json!({"selector": "x.y=z"})));
703        let data_type = DataType::new("ExampleType", metadata.clone(), None);
704
705        let roundtrip = DataType::from_str(data_type.topic()).unwrap();
706
707        assert_eq!(roundtrip.metadata(), metadata.as_ref());
708        assert_eq!(roundtrip.topic(), data_type.topic());
709    }
710
711    #[rstest]
712    fn test_data_type_topic_identity_uses_canonical_metadata_order() {
713        let metadata1 =
714            Params::new([("b".to_string(), json!(2)), ("a".to_string(), json!(1))].into());
715        let metadata2 =
716            Params::new([("a".to_string(), json!(1)), ("b".to_string(), json!(2))].into());
717
718        let data_type1 = DataType::new("ExampleType", Some(metadata1), None);
719        let data_type2 = DataType::new("ExampleType", Some(metadata2), None);
720        let mut hasher1 = DefaultHasher::new();
721        data_type1.hash(&mut hasher1);
722        let hash1 = hasher1.finish();
723        let mut hasher2 = DefaultHasher::new();
724        data_type2.hash(&mut hasher2);
725        let hash2 = hasher2.finish();
726
727        assert_eq!(data_type1.topic(), "ExampleType.a=1.b=2");
728        assert_eq!(data_type1.topic(), data_type2.topic());
729        assert_eq!(data_type1, data_type2);
730        assert_eq!(hash1, hash2);
731        assert_eq!(format!("{data_type1}"), format!("{data_type2}"));
732        assert_eq!(data_type1.metadata_str(), r#"{"a":1,"b":2}"#);
733        assert_eq!(data_type1.metadata_str(), data_type2.metadata_str());
734    }
735
736    #[rstest]
737    fn test_data_type_deserialization_recomputes_hash_from_topic() {
738        let payload = json!({
739            "type_name": "ExampleType",
740            "metadata": {"key": "value"},
741            "topic": "custom.topic",
742            "hash": calculate_hash("custom.topic") ^ u64::MAX,
743            "identifier": "catalog/path",
744        });
745
746        let deserialized: DataType = serde_json::from_value(payload).unwrap();
747
748        assert_eq!(deserialized.type_name(), "ExampleType");
749        assert_eq!(
750            deserialized.metadata(),
751            Some(&params_from_json(json!({"key": "value"})))
752        );
753        assert_eq!(deserialized.identifier(), Some("catalog/path"));
754        assert_eq!(deserialized.topic(), "custom.topic");
755        assert_eq!(
756            deserialized.precomputed_hash(),
757            calculate_hash("custom.topic")
758        );
759    }
760
761    #[rstest]
762    fn test_data_type_deserialization_without_cache_fields_uses_constructor() {
763        let payload = json!({
764            "type_name": "ExampleType",
765            "metadata": {"z": 9, "a": 1},
766            "identifier": "catalog/path",
767        });
768        let expected = DataType::new(
769            "ExampleType",
770            Some(params_from_json(json!({"z": 9, "a": 1}))),
771            Some("catalog/path".to_string()),
772        );
773
774        let deserialized: DataType = serde_json::from_value(payload).unwrap();
775
776        assert_eq!(deserialized.type_name(), expected.type_name());
777        assert_eq!(deserialized.metadata(), expected.metadata());
778        assert_eq!(deserialized.identifier(), expected.identifier());
779        assert_eq!(deserialized.topic(), expected.topic());
780        assert_eq!(deserialized.precomputed_hash(), expected.precomputed_hash());
781    }
782
783    #[rstest]
784    fn test_data_type_deserialization_preserves_topic_without_hash() {
785        let payload = json!({
786            "type_name": "ExampleType",
787            "metadata": null,
788            "topic": "custom.topic",
789        });
790
791        let deserialized: DataType = serde_json::from_value(payload).unwrap();
792
793        assert_eq!(deserialized.topic(), "custom.topic");
794        assert_eq!(
795            deserialized.precomputed_hash(),
796            calculate_hash("custom.topic")
797        );
798    }
799
800    #[rstest]
801    fn test_data_type_deserialization_ignores_hash_without_topic() {
802        let expected = DataType::new("ExampleType", None, None);
803        let payload = json!({
804            "type_name": "ExampleType",
805            "metadata": null,
806            "hash": expected.precomputed_hash() ^ u64::MAX,
807        });
808
809        let deserialized: DataType = serde_json::from_value(payload).unwrap();
810
811        assert_eq!(deserialized, expected);
812        assert_eq!(deserialized.precomputed_hash(), expected.precomputed_hash());
813    }
814
815    #[rstest]
816    fn test_data_type_deserialization_rejects_duplicate_map_key() {
817        let payload = r#"{"type_name":"ExampleType","topic":"first","topic":"second"}"#;
818
819        let e = serde_json::from_str::<DataType>(payload).unwrap_err();
820
821        assert!(e.to_string().contains("duplicate field `topic`"));
822    }
823
824    #[rstest]
825    #[case(
826        r#"{"type_name":"ExampleType","topic":null,"topic":"second"}"#,
827        "duplicate field `topic`"
828    )]
829    #[case(
830        r#"{"type_name":"ExampleType","hash":null,"hash":7}"#,
831        "duplicate field `hash`"
832    )]
833    #[case(
834        r#"{"type_name":"ExampleType","metadata":null,"metadata":{"a":1}}"#,
835        "duplicate field `metadata`"
836    )]
837    #[case(
838        r#"{"type_name":"ExampleType","identifier":null,"identifier":"second"}"#,
839        "duplicate field `identifier`"
840    )]
841    fn test_data_type_deserialization_rejects_duplicate_map_key_after_null(
842        #[case] payload: &str,
843        #[case] expected: &str,
844    ) {
845        let e = serde_json::from_str::<DataType>(payload).unwrap_err();
846
847        assert!(e.to_string().contains(expected));
848    }
849
850    #[rstest]
851    fn test_data_type_serde_roundtrip_preserves_fields_and_repairs_hash() {
852        let payload = json!({
853            "type_name": "ExampleType",
854            "metadata": {"key": "value"},
855            "topic": "custom.topic",
856            "hash": calculate_hash("custom.topic") ^ u64::MAX,
857            "identifier": "catalog/path",
858        });
859        let deserialized: DataType = serde_json::from_value(payload).unwrap();
860
861        let json = serde_json::to_string(&deserialized).unwrap();
862        let roundtripped: DataType = serde_json::from_str(&json).unwrap();
863
864        assert_eq!(roundtripped.type_name(), "ExampleType");
865        assert_eq!(
866            roundtripped.metadata(),
867            Some(&params_from_json(json!({"key": "value"})))
868        );
869        assert_eq!(roundtripped.identifier(), Some("catalog/path"));
870        assert_eq!(roundtripped.topic(), "custom.topic");
871        assert_eq!(
872            roundtripped.precomputed_hash(),
873            calculate_hash("custom.topic")
874        );
875    }
876
877    #[rstest]
878    fn test_data_type_serialized_cache_fields_remain_wire_compatible() {
879        #[derive(Deserialize)]
880        struct LegacyDataType {
881            type_name: String,
882            metadata: Option<Params>,
883            topic: String,
884            hash: u64,
885            identifier: Option<String>,
886        }
887
888        let expected = DataType::new(
889            "ExampleType",
890            Some(params_from_json(json!({"key": "value"}))),
891            Some("catalog/path".to_string()),
892        );
893        let mut payload = serde_json::to_value(&expected).unwrap();
894        payload["hash"] = json!(expected.precomputed_hash() ^ u64::MAX);
895        let repaired: DataType = serde_json::from_value(payload).unwrap();
896
897        let legacy: LegacyDataType =
898            serde_json::from_value(serde_json::to_value(&repaired).unwrap()).unwrap();
899
900        assert_eq!(legacy.type_name, expected.type_name());
901        assert_eq!(legacy.metadata.as_ref(), expected.metadata());
902        assert_eq!(legacy.topic, expected.topic());
903        assert_eq!(legacy.hash, expected.precomputed_hash());
904        assert_eq!(legacy.identifier.as_deref(), expected.identifier());
905    }
906
907    #[rstest]
908    fn test_data_type_topic_distinguishes_numeric_strings_from_numbers() {
909        let string_value = DataType::new(
910            "ExampleType",
911            Some(params_from_json(json!({"value": "123"}))),
912            None,
913        );
914        let numeric_value = DataType::new(
915            "ExampleType",
916            Some(params_from_json(json!({"value": 123}))),
917            None,
918        );
919
920        assert_eq!(string_value.topic(), "ExampleType.value=~s~123");
921        assert_eq!(numeric_value.topic(), "ExampleType.value=123");
922        assert_ne!(string_value, numeric_value);
923        assert_eq!(
924            DataType::from_str(string_value.topic()).unwrap().metadata(),
925            string_value.metadata()
926        );
927        assert_eq!(
928            DataType::from_str(numeric_value.topic())
929                .unwrap()
930                .metadata(),
931            numeric_value.metadata()
932        );
933    }
934
935    #[rstest]
936    fn test_data_type_topic_distinguishes_whole_floats_from_integers() {
937        let integer_value = DataType::new(
938            "ExampleType",
939            Some(params_from_json(json!({"value": 1}))),
940            None,
941        );
942        let float_value = DataType::new(
943            "ExampleType",
944            Some(params_from_json(json!({"value": 1.0}))),
945            None,
946        );
947
948        assert_eq!(integer_value.topic(), "ExampleType.value=1");
949        assert_eq!(float_value.topic(), "ExampleType.value=1.0");
950        assert_ne!(integer_value, float_value);
951        assert_eq!(
952            DataType::from_str(float_value.topic()).unwrap().metadata(),
953            float_value.metadata()
954        );
955    }
956
957    #[rstest]
958    fn test_data_type_topic_escapes_reserved_string_prefix() {
959        let data_type = DataType::new(
960            "ExampleType",
961            Some(params_from_json(json!({"value": "~s~123"}))),
962            None,
963        );
964
965        assert_eq!(data_type.topic(), "ExampleType.value=~s~~s~123");
966        assert_eq!(
967            DataType::from_str(data_type.topic()).unwrap().metadata(),
968            data_type.metadata()
969        );
970    }
971
972    #[rstest]
973    fn test_data_type_rejects_identifier_metadata_key() {
974        let error = DataType::try_new(
975            "ExampleType",
976            Some(params_from_json(json!({"identifier": "A"}))),
977            None,
978        )
979        .unwrap_err();
980
981        assert_eq!(
982            error.to_string(),
983            "`identifier` is reserved for DataType catalog identity"
984        );
985    }
986
987    #[rstest]
988    fn test_data_type_persistence_json_rejects_identifier_metadata_key() {
989        let error = DataType::from_persistence_json(
990            r#"{"type_name":"ExampleType","metadata":{"identifier":"A"}}"#,
991        )
992        .unwrap_err();
993
994        assert_eq!(
995            error.to_string(),
996            "`identifier` is reserved for DataType catalog identity"
997        );
998    }
999
1000    #[rstest]
1001    fn test_data_type_creation_without_metadata() {
1002        let data_type = DataType::new("ExampleType", None, None);
1003
1004        assert_eq!(data_type.type_name(), "ExampleType");
1005        assert_eq!(data_type.topic(), "ExampleType");
1006        assert_eq!(data_type.metadata(), None);
1007    }
1008
1009    #[rstest]
1010    fn test_data_type_equality() {
1011        let metadata1 = Some(params_from_json(json!({"key1": "value1"})));
1012        let metadata2 = Some(params_from_json(json!({"key1": "value1"})));
1013
1014        let data_type1 = DataType::new("ExampleType", metadata1, None);
1015        let data_type2 = DataType::new("ExampleType", metadata2, None);
1016
1017        assert_eq!(data_type1, data_type2);
1018    }
1019
1020    #[rstest]
1021    fn test_data_type_inequality() {
1022        let metadata1 = Some(params_from_json(json!({"key1": "value1"})));
1023        let metadata2 = Some(params_from_json(json!({"key2": "value2"})));
1024
1025        let data_type1 = DataType::new("ExampleType", metadata1, None);
1026        let data_type2 = DataType::new("ExampleType", metadata2, None);
1027
1028        assert_ne!(data_type1, data_type2);
1029    }
1030
1031    #[rstest]
1032    fn test_data_type_ordering() {
1033        let metadata1 = Some(params_from_json(json!({"key1": "value1"})));
1034        let metadata2 = Some(params_from_json(json!({"key2": "value2"})));
1035
1036        let data_type1 = DataType::new("ExampleTypeA", metadata1, None);
1037        let data_type2 = DataType::new("ExampleTypeB", metadata2, None);
1038
1039        assert!(data_type1 < data_type2);
1040    }
1041
1042    #[rstest]
1043    fn test_data_type_hash() {
1044        let metadata = Some(params_from_json(json!({"key1": "value1"})));
1045
1046        let data_type1 = DataType::new("ExampleType", metadata.clone(), None);
1047        let data_type2 = DataType::new("ExampleType", metadata, None);
1048
1049        let mut hasher1 = DefaultHasher::new();
1050        data_type1.hash(&mut hasher1);
1051        let hash1 = hasher1.finish();
1052
1053        let mut hasher2 = DefaultHasher::new();
1054        data_type2.hash(&mut hasher2);
1055        let hash2 = hasher2.finish();
1056
1057        assert_eq!(hash1, hash2);
1058    }
1059
1060    #[rstest]
1061    fn test_data_type_identifier_is_part_of_topic_identity_and_hash() {
1062        let data_type1 = DataType::new("ExampleType", None, Some("A".to_string()));
1063        let data_type2 = DataType::new("ExampleType", None, Some("B".to_string()));
1064
1065        let mut hasher1 = DefaultHasher::new();
1066        data_type1.hash(&mut hasher1);
1067        let hash1 = hasher1.finish();
1068        let mut hasher2 = DefaultHasher::new();
1069        data_type2.hash(&mut hasher2);
1070        let hash2 = hasher2.finish();
1071
1072        assert_eq!(data_type1.topic(), "ExampleType.identifier=A");
1073        assert_eq!(data_type2.topic(), "ExampleType.identifier=B");
1074        assert_ne!(data_type1, data_type2);
1075        assert_ne!(hash1, hash2);
1076    }
1077
1078    #[rstest]
1079    fn test_data_type_display() {
1080        let metadata = Some(params_from_json(json!({"key1": "value1"})));
1081        let data_type = DataType::new("ExampleType", metadata, None);
1082
1083        assert_eq!(format!("{data_type}"), "ExampleType.key1=value1");
1084    }
1085
1086    #[rstest]
1087    fn test_data_type_debug() {
1088        let metadata = Some(params_from_json(json!({"key1": "value1"})));
1089        let data_type = DataType::new("ExampleType", metadata.clone(), None);
1090
1091        assert_eq!(
1092            format!("{data_type:?}"),
1093            format!("DataType(type_name=ExampleType, metadata={metadata:?}, identifier=None)")
1094        );
1095    }
1096
1097    #[rstest]
1098    fn test_parse_instrument_id_from_metadata() {
1099        let instrument_id_str = "MSFT.XNAS";
1100        let metadata = Some(params_from_json(
1101            json!({"instrument_id": instrument_id_str}),
1102        ));
1103        let data_type = DataType::new("InstrumentAny", metadata, None);
1104
1105        assert_eq!(
1106            data_type.instrument_id().unwrap().unwrap(),
1107            InstrumentId::from_str(instrument_id_str).unwrap()
1108        );
1109    }
1110
1111    #[rstest]
1112    fn test_parse_venue_from_metadata() {
1113        let venue_str = "BINANCE";
1114        let metadata = Some(params_from_json(json!({"venue": venue_str})));
1115        let data_type = DataType::new(stringify!(InstrumentAny), metadata, None);
1116
1117        assert_eq!(data_type.venue().unwrap().unwrap(), Venue::new(venue_str));
1118    }
1119
1120    #[rstest]
1121    fn test_parse_start_from_metadata() {
1122        let start_ns = 1_600_054_595_844_758_000;
1123        let metadata = Some(params_from_json(json!({"start": start_ns.to_string()})));
1124        let data_type = DataType::new(stringify!(TradeTick), metadata, None);
1125
1126        assert_eq!(
1127            data_type.start().unwrap().unwrap(),
1128            UnixNanos::from(start_ns),
1129        );
1130    }
1131
1132    #[rstest]
1133    fn test_parse_end_from_metadata() {
1134        let end_ns = 1_720_954_595_844_758_000;
1135        let metadata = Some(params_from_json(json!({"end": end_ns.to_string()})));
1136        let data_type = DataType::new(stringify!(TradeTick), metadata, None);
1137
1138        assert_eq!(data_type.end().unwrap().unwrap(), UnixNanos::from(end_ns),);
1139    }
1140
1141    #[rstest]
1142    fn test_parse_limit_from_metadata() {
1143        let limit = 1000;
1144        let metadata = Some(params_from_json(json!({"limit": limit})));
1145        let data_type = DataType::new(stringify!(TradeTick), metadata, None);
1146
1147        assert_eq!(data_type.limit().unwrap().unwrap(), limit);
1148    }
1149
1150    #[rstest]
1151    fn test_data_type_metadata_accessors_return_none_without_metadata() {
1152        let data_type = DataType::new(stringify!(TradeTick), None, None);
1153
1154        assert_eq!(data_type.instrument_id().unwrap(), None);
1155        assert_eq!(data_type.venue().unwrap(), None);
1156        assert_eq!(data_type.start().unwrap(), None);
1157        assert_eq!(data_type.end().unwrap(), None);
1158        assert_eq!(data_type.limit().unwrap(), None);
1159    }
1160
1161    #[rstest]
1162    fn test_data_type_metadata_accessors_reject_invalid_values() {
1163        let data_type = DataType::new(
1164            stringify!(TradeTick),
1165            Some(params_from_json(json!({
1166                "instrument_id": "invalid",
1167                "venue": "",
1168                "start": "invalid",
1169                "end": "invalid",
1170                "limit": "invalid",
1171            }))),
1172            None,
1173        );
1174
1175        assert_eq!(
1176            data_type.instrument_id().unwrap_err().to_string(),
1177            "Invalid instrument_id metadata `invalid`: invalid `InstrumentId` value 'invalid': missing '.' separator between symbol and venue components"
1178        );
1179        assert_eq!(
1180            data_type.venue().unwrap_err().to_string(),
1181            "Invalid venue metadata ``: invalid string for 'value', was empty"
1182        );
1183        assert_eq!(
1184            data_type.start().unwrap_err().to_string(),
1185            "Invalid start metadata `invalid`: Invalid format: invalid"
1186        );
1187        assert_eq!(
1188            data_type.end().unwrap_err().to_string(),
1189            "Invalid end metadata `invalid`: Invalid format: invalid"
1190        );
1191        assert_eq!(
1192            data_type.limit().unwrap_err().to_string(),
1193            "Invalid limit metadata `invalid`: invalid digit found in string"
1194        );
1195    }
1196
1197    #[rstest]
1198    fn test_data_type_persistence_json_with_identifier() {
1199        let data_type = DataType::new("MyCustomType", None, Some("SYMBOL.VENUE".to_string()));
1200        let json = data_type.to_persistence_json().unwrap();
1201        assert!(!json.contains("topic"));
1202        assert!(json.contains("\"identifier\":\"SYMBOL.VENUE\""));
1203        let restored = DataType::from_persistence_json(&json).unwrap();
1204        assert_eq!(restored.type_name(), "MyCustomType");
1205        assert_eq!(restored.identifier(), Some("SYMBOL.VENUE"));
1206        assert_eq!(restored.topic(), "MyCustomType.identifier=SYMBOL.VENUE");
1207    }
1208
1209    #[rstest]
1210    fn test_data_type_from_str_parses_metadata_and_identifier() {
1211        let data_type = DataType::from_str(
1212            "MyCustomType.instrument_id=SYMBOL.VENUE.strike=1.23.identifier=SYMBOL.VENUE",
1213        )
1214        .unwrap();
1215
1216        assert_eq!(data_type.type_name(), "MyCustomType");
1217        assert_eq!(
1218            data_type.metadata(),
1219            Some(&params_from_json(
1220                json!({"instrument_id": "SYMBOL.VENUE", "strike": 1.23})
1221            ))
1222        );
1223        assert_eq!(data_type.identifier(), Some("SYMBOL.VENUE"));
1224        assert_eq!(
1225            data_type.topic(),
1226            "MyCustomType.instrument_id=SYMBOL.VENUE.strike=1.23.identifier=SYMBOL.VENUE"
1227        );
1228    }
1229
1230    #[rstest]
1231    fn test_data_type_from_persistence_json_rebuilds_canonical_topic() {
1232        let json = r#"{
1233            "type_name": "ExampleType",
1234            "topic": "ExampleType.z=9.a=1",
1235            "metadata": {"z": 9, "a": 1}
1236        }"#;
1237
1238        let restored = DataType::from_persistence_json(json).unwrap();
1239
1240        assert_eq!(restored.topic(), "ExampleType.a=1.z=9");
1241    }
1242
1243    #[rstest]
1244    fn test_data_type_persistence_result_hashes_like_equal_deserialized_value() {
1245        let persistence_json = r#"{
1246            "type_name": "ExampleType",
1247            "topic": "ignored.legacy.topic",
1248            "metadata": {"z": 9, "a": 1},
1249            "identifier": "catalog/path"
1250        }"#;
1251        let persisted = DataType::from_persistence_json(persistence_json).unwrap();
1252        let payload = json!({
1253            "type_name": persisted.type_name(),
1254            "metadata": persisted.metadata(),
1255            "topic": persisted.topic(),
1256            "hash": persisted.precomputed_hash() ^ u64::MAX,
1257            "identifier": persisted.identifier(),
1258        });
1259        let deserialized: DataType = serde_json::from_value(payload).unwrap();
1260        let hash = |data_type: &DataType| {
1261            let mut hasher = DefaultHasher::new();
1262            data_type.hash(&mut hasher);
1263            hasher.finish()
1264        };
1265
1266        assert_eq!(
1267            persisted.topic(),
1268            "ExampleType.a=1.z=9.identifier=catalog/path"
1269        );
1270        assert_eq!(persisted.identifier(), Some("catalog/path"));
1271        assert_eq!(deserialized, persisted);
1272        assert_eq!(hash(&deserialized), hash(&persisted));
1273    }
1274
1275    #[rstest]
1276    fn test_data_type_identifier_getter() {
1277        let data_type = DataType::new("T", None, Some("id".to_string()));
1278        assert_eq!(data_type.identifier(), Some("id"));
1279        let data_type_no_id = DataType::new("T", None, None);
1280        assert_eq!(data_type_no_id.identifier(), None);
1281    }
1282}
nautilus_model/data/data_type.rs

nautilus_model/data/
data_type.rs
