atomr_persistence_azure/
entities.rs

1//! Row shapes stored in the journal / snapshot tables. Payload bytes are
2//! base64-encoded because Table Storage doesn't support raw binary
3//! natively, only "typed" properties.
4
5use atomr_persistence::{PersistentRepr, SnapshotMetadata};
6use base64::{engine::general_purpose::STANDARD as B64, Engine as _};
7use serde::{Deserialize, Serialize};
8
9fn row_key(sequence_nr: u64) -> String {
10    format!("{:020}", sequence_nr)
11}
12
13#[derive(Debug, Clone, Serialize, Deserialize)]
14#[serde(rename_all = "PascalCase")]
15pub struct EventEntity {
16    pub partition_key: String,
17    pub row_key: String,
18    pub sequence_nr: i64,
19    pub payload_b64: String,
20    pub manifest: String,
21    pub writer_uuid: String,
22    pub deleted: bool,
23    pub tags_csv: String,
24}
25
26impl EventEntity {
27    pub fn from_repr(repr: &PersistentRepr) -> Self {
28        Self {
29            partition_key: repr.persistence_id.clone(),
30            row_key: row_key(repr.sequence_nr),
31            sequence_nr: repr.sequence_nr as i64,
32            payload_b64: B64.encode(&repr.payload),
33            manifest: repr.manifest.clone(),
34            writer_uuid: repr.writer_uuid.clone(),
35            deleted: repr.deleted,
36            tags_csv: repr.tags.join(","),
37        }
38    }
39
40    pub fn into_repr(self) -> PersistentRepr {
41        PersistentRepr {
42            persistence_id: self.partition_key,
43            sequence_nr: self.sequence_nr as u64,
44            payload: B64.decode(self.payload_b64).unwrap_or_default(),
45            manifest: self.manifest,
46            writer_uuid: self.writer_uuid,
47            deleted: self.deleted,
48            tags: if self.tags_csv.is_empty() {
49                Vec::new()
50            } else {
51                self.tags_csv.split(',').map(|s| s.to_string()).collect()
52            },
53        }
54    }
55}
56
57#[derive(Debug, Clone, Serialize, Deserialize)]
58#[serde(rename_all = "PascalCase")]
59pub struct SnapshotEntity {
60    pub partition_key: String,
61    pub row_key: String,
62    pub sequence_nr: i64,
63    pub payload_b64: String,
64    pub timestamp_ms: i64,
65}
66
67impl SnapshotEntity {
68    pub fn from_meta(meta: &SnapshotMetadata, payload: &[u8]) -> Self {
69        Self {
70            partition_key: meta.persistence_id.clone(),
71            row_key: row_key(meta.sequence_nr),
72            sequence_nr: meta.sequence_nr as i64,
73            payload_b64: B64.encode(payload),
74            timestamp_ms: meta.timestamp as i64,
75        }
76    }
77
78    pub fn into_parts(self) -> (SnapshotMetadata, Vec<u8>) {
79        (
80            SnapshotMetadata {
81                persistence_id: self.partition_key,
82                sequence_nr: self.sequence_nr as u64,
83                timestamp: self.timestamp_ms as u64,
84            },
85            B64.decode(self.payload_b64).unwrap_or_default(),
86        )
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93
94    #[test]
95    fn event_round_trip() {
96        let original = PersistentRepr {
97            persistence_id: "p".into(),
98            sequence_nr: 5,
99            payload: vec![1, 2, 3],
100            manifest: "m".into(),
101            writer_uuid: "w".into(),
102            deleted: false,
103            tags: vec!["a".into(), "b".into()],
104        };
105        let entity = EventEntity::from_repr(&original);
106        let back = entity.into_repr();
107        assert_eq!(back.payload, original.payload);
108        assert_eq!(back.tags, original.tags);
109    }
110
111    #[test]
112    fn row_key_is_zero_padded() {
113        let entity = EventEntity::from_repr(&PersistentRepr {
114            persistence_id: "p".into(),
115            sequence_nr: 7,
116            payload: vec![],
117            manifest: "".into(),
118            writer_uuid: "".into(),
119            deleted: false,
120            tags: vec![],
121        });
122        assert_eq!(entity.row_key, "00000000000000000007");
123    }
124}
atomr_persistence_azure/entities.rs

atomr_persistence_azure/
entities.rs
