pjson_rs_domain/entities/
frame.rs

1//! Frame entity with streaming data
2
3use crate::{
4    DomainError, DomainResult,
5    value_objects::{JsonData, JsonPath, Priority, StreamId},
6};
7use chrono::{DateTime, Utc};
8use serde::{Deserialize, Serialize};
9use std::collections::HashMap;
10
11/// Custom serde for StreamId within entities
12mod serde_stream_id {
13    use crate::value_objects::StreamId;
14    use serde::{Deserialize, Deserializer, Serialize, Serializer};
15
16    pub fn serialize<S>(id: &StreamId, serializer: S) -> Result<S::Ok, S::Error>
17    where
18        S: Serializer,
19    {
20        id.as_uuid().serialize(serializer)
21    }
22
23    pub fn deserialize<'de, D>(deserializer: D) -> Result<StreamId, D::Error>
24    where
25        D: Deserializer<'de>,
26    {
27        let uuid = uuid::Uuid::deserialize(deserializer)?;
28        Ok(StreamId::from_uuid(uuid))
29    }
30}
31
32/// Custom serde for Priority within entities
33mod serde_priority {
34    use crate::value_objects::Priority;
35    use serde::{Deserialize, Deserializer, Serialize, Serializer};
36
37    pub fn serialize<S>(priority: &Priority, serializer: S) -> Result<S::Ok, S::Error>
38    where
39        S: Serializer,
40    {
41        priority.value().serialize(serializer)
42    }
43
44    pub fn deserialize<'de, D>(deserializer: D) -> Result<Priority, D::Error>
45    where
46        D: Deserializer<'de>,
47    {
48        let value = u8::deserialize(deserializer)?;
49        Priority::new(value).map_err(serde::de::Error::custom)
50    }
51}
52
53/// Custom serde for JsonPath within entities
54mod serde_json_path {
55    use crate::value_objects::JsonPath;
56    use serde::{Deserialize, Deserializer, Serialize, Serializer};
57
58    pub fn serialize<S>(path: &JsonPath, serializer: S) -> Result<S::Ok, S::Error>
59    where
60        S: Serializer,
61    {
62        path.as_str().serialize(serializer)
63    }
64
65    pub fn deserialize<'de, D>(deserializer: D) -> Result<JsonPath, D::Error>
66    where
67        D: Deserializer<'de>,
68    {
69        let s = String::deserialize(deserializer)?;
70        JsonPath::new(s).map_err(serde::de::Error::custom)
71    }
72}
73
74/// Frame types for different stages of streaming
75///
76/// # Wire format
77///
78/// `FrameType` is serialized as a bare JSON string (Serde's default for fieldless
79/// variants). Adding a new variant changes the wire format — older deserializers
80/// will fail on unknown frame-type strings. The `#[non_exhaustive]` marker prevents
81/// downstream Rust crates from breaking on `match` exhaustiveness, but it does not
82/// address the wire-format compatibility concern.
83#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
84#[non_exhaustive]
85pub enum FrameType {
86    /// Initial skeleton with structure
87    Skeleton,
88    /// Data patch update
89    Patch,
90    /// Stream completion signal
91    Complete,
92    /// Error notification
93    Error,
94}
95
96/// Individual frame in a priority stream
97#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98pub struct Frame {
99    #[serde(with = "serde_stream_id")]
100    stream_id: StreamId,
101    frame_type: FrameType,
102    #[serde(with = "serde_priority")]
103    priority: Priority,
104    sequence: u64,
105    timestamp: DateTime<Utc>,
106    payload: JsonData,
107    metadata: HashMap<String, String>,
108}
109
110impl std::hash::Hash for Frame {
111    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
112        self.stream_id.hash(state);
113        self.frame_type.hash(state);
114        self.priority.hash(state);
115        self.sequence.hash(state);
116        self.timestamp.hash(state);
117        self.payload.hash(state);
118
119        // For HashMap, sort keys for consistent hashing
120        let mut pairs: Vec<_> = self.metadata.iter().collect();
121        pairs.sort_by_key(|(k, _)| *k);
122        pairs.hash(state);
123    }
124}
125
126impl Frame {
127    /// Create new skeleton frame
128    pub fn skeleton(stream_id: StreamId, sequence: u64, skeleton_data: JsonData) -> Self {
129        Self {
130            stream_id,
131            frame_type: FrameType::Skeleton,
132            priority: Priority::CRITICAL,
133            sequence,
134            timestamp: Utc::now(),
135            payload: skeleton_data,
136            metadata: HashMap::new(),
137        }
138    }
139
140    /// Create new patch frame
141    pub fn patch(
142        stream_id: StreamId,
143        sequence: u64,
144        priority: Priority,
145        patches: Vec<FramePatch>,
146    ) -> DomainResult<Self> {
147        if patches.is_empty() {
148            return Err(DomainError::InvalidFrame(
149                "Patch frame must contain at least one patch".to_string(),
150            ));
151        }
152
153        // Create JsonData payload directly instead of using serde_json
154        let mut payload_obj = HashMap::with_capacity(1);
155        let patches_array: Vec<JsonData> = patches
156            .into_iter()
157            .map(|patch| {
158                let mut patch_obj = HashMap::with_capacity(3);
159                patch_obj.insert("path".into(), JsonData::String(patch.path.to_string()));
160                patch_obj.insert(
161                    "operation".into(),
162                    JsonData::String(
163                        match patch.operation {
164                            PatchOperation::Set => "set",
165                            PatchOperation::Append => "append",
166                            PatchOperation::Merge => "merge",
167                            PatchOperation::Delete => "delete",
168                        }
169                        .into(),
170                    ),
171                );
172                patch_obj.insert("value".into(), patch.value);
173                JsonData::Object(patch_obj)
174            })
175            .collect();
176
177        payload_obj.insert("patches".into(), JsonData::Array(patches_array));
178        let payload = JsonData::Object(payload_obj);
179
180        Ok(Self {
181            stream_id,
182            frame_type: FrameType::Patch,
183            priority,
184            sequence,
185            timestamp: Utc::now(),
186            payload,
187            metadata: HashMap::new(),
188        })
189    }
190
191    /// Create completion frame
192    pub fn complete(stream_id: StreamId, sequence: u64, checksum: Option<String>) -> Self {
193        let payload = if let Some(checksum) = checksum {
194            let mut obj = HashMap::new();
195            obj.insert("checksum".to_string(), JsonData::String(checksum));
196            JsonData::Object(obj)
197        } else {
198            JsonData::Object(HashMap::new())
199        };
200
201        Self {
202            stream_id,
203            frame_type: FrameType::Complete,
204            priority: Priority::CRITICAL,
205            sequence,
206            timestamp: Utc::now(),
207            payload,
208            metadata: HashMap::new(),
209        }
210    }
211
212    /// Create error frame
213    pub fn error(
214        stream_id: StreamId,
215        sequence: u64,
216        error_message: String,
217        error_code: Option<String>,
218    ) -> Self {
219        let payload = if let Some(code) = error_code {
220            let mut obj = HashMap::new();
221            obj.insert("message".to_string(), JsonData::String(error_message));
222            obj.insert("code".to_string(), JsonData::String(code));
223            JsonData::Object(obj)
224        } else {
225            let mut obj = HashMap::new();
226            obj.insert("message".to_string(), JsonData::String(error_message));
227            JsonData::Object(obj)
228        };
229
230        Self {
231            stream_id,
232            frame_type: FrameType::Error,
233            priority: Priority::CRITICAL,
234            sequence,
235            timestamp: Utc::now(),
236            payload,
237            metadata: HashMap::new(),
238        }
239    }
240
241    /// Get stream ID
242    pub fn stream_id(&self) -> StreamId {
243        self.stream_id
244    }
245
246    /// Get frame type
247    pub fn frame_type(&self) -> &FrameType {
248        &self.frame_type
249    }
250
251    /// Get priority
252    pub fn priority(&self) -> Priority {
253        self.priority
254    }
255
256    /// Get sequence number
257    pub fn sequence(&self) -> u64 {
258        self.sequence
259    }
260
261    /// Get timestamp
262    pub fn timestamp(&self) -> DateTime<Utc> {
263        self.timestamp
264    }
265
266    /// Get payload
267    pub fn payload(&self) -> &JsonData {
268        &self.payload
269    }
270
271    /// Add metadata
272    pub fn with_metadata(mut self, key: String, value: String) -> Self {
273        self.metadata.insert(key, value);
274        self
275    }
276
277    /// Get metadata
278    pub fn metadata(&self) -> &HashMap<String, String> {
279        &self.metadata
280    }
281
282    /// Get metadata value
283    pub fn get_metadata(&self, key: &str) -> Option<&String> {
284        self.metadata.get(key)
285    }
286
287    /// Check if frame is critical priority
288    pub fn is_critical(&self) -> bool {
289        self.priority.is_critical()
290    }
291
292    /// Check if frame is high priority or above
293    pub fn is_high_priority(&self) -> bool {
294        self.priority.is_high_or_above()
295    }
296
297    /// Estimate frame size in bytes (for network planning)
298    pub fn estimated_size(&self) -> usize {
299        // Rough estimation: JSON serialization + metadata overhead
300        let payload_size = self.payload.to_string().len();
301        let metadata_size: usize = self
302            .metadata
303            .iter()
304            .map(|(k, v)| k.len() + v.len() + 4) // JSON overhead
305            .sum();
306
307        payload_size + metadata_size + 200 // Base frame overhead
308    }
309
310    /// Validate frame consistency
311    pub fn validate(&self) -> DomainResult<()> {
312        match &self.frame_type {
313            FrameType::Skeleton => {
314                if !self.priority.is_critical() {
315                    return Err(DomainError::InvalidFrame(
316                        "Skeleton frames must have critical priority".to_string(),
317                    ));
318                }
319            }
320            FrameType::Patch => {
321                // Validate patch payload structure
322                if !self.payload.is_object() {
323                    return Err(DomainError::InvalidFrame(
324                        "Patch frames must have object payload".to_string(),
325                    ));
326                }
327
328                if !self.payload.get("patches").is_some_and(|p| p.is_array()) {
329                    return Err(DomainError::InvalidFrame(
330                        "Patch frames must contain patches array".to_string(),
331                    ));
332                }
333            }
334            FrameType::Complete => {
335                if !self.priority.is_critical() {
336                    return Err(DomainError::InvalidFrame(
337                        "Complete frames must have critical priority".to_string(),
338                    ));
339                }
340            }
341            FrameType::Error => {
342                if !self.priority.is_critical() {
343                    return Err(DomainError::InvalidFrame(
344                        "Error frames must have critical priority".to_string(),
345                    ));
346                }
347
348                if !self.payload.get("message").is_some_and(|m| m.is_string()) {
349                    return Err(DomainError::InvalidFrame(
350                        "Error frames must contain message".to_string(),
351                    ));
352                }
353            }
354        }
355
356        Ok(())
357    }
358}
359
360/// Individual patch within a frame
361#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
362pub struct FramePatch {
363    /// JSON path to the target location
364    #[serde(with = "serde_json_path")]
365    pub path: JsonPath,
366    /// Operation to perform at the path
367    pub operation: PatchOperation,
368    /// Value to apply with the operation
369    pub value: JsonData,
370}
371
372/// Patch operation types
373#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
374#[serde(rename_all = "lowercase")]
375#[non_exhaustive]
376pub enum PatchOperation {
377    /// Set a value at the path
378    Set,
379    /// Append to an array at the path
380    Append,
381    /// Merge object at the path
382    Merge,
383    /// Delete value at the path
384    Delete,
385}
386
387impl FramePatch {
388    /// Create set operation patch
389    pub fn set(path: JsonPath, value: JsonData) -> Self {
390        Self {
391            path,
392            operation: PatchOperation::Set,
393            value,
394        }
395    }
396
397    /// Create append operation patch
398    pub fn append(path: JsonPath, value: JsonData) -> Self {
399        Self {
400            path,
401            operation: PatchOperation::Append,
402            value,
403        }
404    }
405
406    /// Create merge operation patch
407    pub fn merge(path: JsonPath, value: JsonData) -> Self {
408        Self {
409            path,
410            operation: PatchOperation::Merge,
411            value,
412        }
413    }
414
415    /// Create delete operation patch
416    pub fn delete(path: JsonPath) -> Self {
417        Self {
418            path,
419            operation: PatchOperation::Delete,
420            value: JsonData::Null,
421        }
422    }
423}
424
425#[cfg(test)]
426mod tests {
427    use super::*;
428
429    #[test]
430    fn test_skeleton_frame_creation() {
431        let stream_id = StreamId::new();
432        let skeleton_data = serde_json::json!({
433            "users": [],
434            "total": 0
435        });
436
437        let frame = Frame::skeleton(stream_id, 1, skeleton_data.clone().into());
438
439        assert_eq!(frame.frame_type(), &FrameType::Skeleton);
440        assert_eq!(frame.priority(), Priority::CRITICAL);
441        assert_eq!(frame.sequence(), 1);
442        assert_eq!(frame.stream_id(), stream_id);
443        assert!(frame.validate().is_ok());
444    }
445
446    #[test]
447    fn test_patch_frame_creation() {
448        let stream_id = StreamId::new();
449        let path = JsonPath::new("$.users[0].name").expect("Failed to create JsonPath in test");
450        let patch = FramePatch::set(path, JsonData::String("John".to_string()));
451
452        let frame = Frame::patch(stream_id, 2, Priority::HIGH, vec![patch])
453            .expect("Failed to create patch frame in test");
454
455        assert_eq!(frame.frame_type(), &FrameType::Patch);
456        assert_eq!(frame.priority(), Priority::HIGH);
457        assert_eq!(frame.sequence(), 2);
458        assert!(frame.validate().is_ok());
459    }
460
461    #[test]
462    fn test_complete_frame_creation() {
463        let stream_id = StreamId::new();
464        let frame = Frame::complete(stream_id, 10, Some("abc123".to_string()));
465
466        assert_eq!(frame.frame_type(), &FrameType::Complete);
467        assert_eq!(frame.priority(), Priority::CRITICAL);
468        assert_eq!(frame.sequence(), 10);
469        assert!(frame.validate().is_ok());
470    }
471
472    #[test]
473    fn test_frame_with_metadata() {
474        let stream_id = StreamId::new();
475        let skeleton_data = serde_json::json!({});
476        let frame = Frame::skeleton(stream_id, 1, skeleton_data.into())
477            .with_metadata("source".to_string(), "api".to_string())
478            .with_metadata("version".to_string(), "1.0".to_string());
479
480        assert_eq!(frame.get_metadata("source"), Some(&"api".to_string()));
481        assert_eq!(frame.get_metadata("version"), Some(&"1.0".to_string()));
482        assert_eq!(frame.metadata().len(), 2);
483    }
484
485    #[test]
486    fn test_empty_patch_validation() {
487        let stream_id = StreamId::new();
488        let result = Frame::patch(stream_id, 1, Priority::MEDIUM, vec![]);
489
490        assert!(result.is_err());
491    }
492}
pjson_rs_domain/entities/frame.rs

pjson_rs_domain/entities/
frame.rs
