reifydb_codec/row/envelope/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_value::{
5	encoding::LeBytes,
6	error::{Error as ValueError, TypeError},
7	util::cowvec::CowVec,
8	value::{datetime::DateTime, partition::Partition},
9};
10use thiserror::Error;
11
12use crate::row::{bytes::EncodedBytes, pod::EncodedPodRow, shape::fingerprint::RowShapeFingerprint};
13
14pub const HAS_CREATED_AT: u8 = 1 << 0;
15
16pub const HAS_UPDATED_AT: u8 = 1 << 1;
17
18pub const HAS_TIME: u8 = 1 << 2;
19
20pub const HAS_FINGERPRINT: u8 = 1 << 3;
21
22pub const HAS_COMMIT_VERSION: u8 = 1 << 4;
23
24pub const HAS_PARTITION: u8 = 1 << 5;
25
26pub const ENVELOPE_FLAGS_SIZE: usize = 1;
27
28pub const ENVELOPE_FIELD_SIZE: usize = DateTime::ENCODED_SIZE;
29
30pub const ENVELOPE_PARTITION_SIZE: usize = 16;
31
32#[inline]
33pub const fn header_size(flags: u8) -> usize {
34	let partition = if flags & HAS_PARTITION != 0 {
35		ENVELOPE_PARTITION_SIZE
36	} else {
37		0
38	};
39	ENVELOPE_FLAGS_SIZE + ENVELOPE_FIELD_SIZE * (flags & !HAS_PARTITION).count_ones() as usize + partition
40}
41
42#[inline]
43const fn field_offset(flags: u8, bit: u8) -> usize {
44	ENVELOPE_FLAGS_SIZE + ENVELOPE_FIELD_SIZE * (flags & (bit - 1)).count_ones() as usize
45}
46
47#[derive(Debug, Error, PartialEq)]
48pub enum EnvelopeError {
49	#[error("envelope row is {len} bytes, too short for the {required} byte header its flags declare")]
50	Truncated {
51		len: usize,
52		required: usize,
53	},
54}
55
56impl From<EnvelopeError> for ValueError {
57	fn from(err: EnvelopeError) -> Self {
58		TypeError::SerdeDeserialize {
59			message: err.to_string(),
60		}
61		.into()
62	}
63}
64
65#[repr(transparent)]
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct Envelope(EncodedBytes);
68
69impl Envelope {
70	pub fn try_view(row: &EncodedPodRow) -> Result<&Self, EnvelopeError> {
71		let bytes = row.bytes();
72		let len = bytes.len();
73		if len < ENVELOPE_FLAGS_SIZE {
74			return Err(EnvelopeError::Truncated {
75				len,
76				required: ENVELOPE_FLAGS_SIZE,
77			});
78		}
79		let required = header_size(bytes[0]);
80		if len < required {
81			return Err(EnvelopeError::Truncated {
82				len,
83				required,
84			});
85		}
86		// SAFETY: Envelope is repr(transparent) over EncodedBytes, so the pointer cast preserves layout,
87		// and the returned reference borrows the same allocation for the same lifetime.
88		Ok(unsafe { &*(bytes as *const EncodedBytes as *const Self) })
89	}
90
91	#[inline]
92	pub fn flags(&self) -> u8 {
93		self.0[0]
94	}
95
96	#[inline]
97	pub fn header_size(&self) -> usize {
98		header_size(self.flags())
99	}
100
101	#[inline]
102	pub fn created_at(&self) -> Option<DateTime> {
103		self.field(HAS_CREATED_AT).map(DateTime::from_le_bytes)
104	}
105
106	#[inline]
107	pub fn updated_at(&self) -> Option<DateTime> {
108		self.field(HAS_UPDATED_AT).map(DateTime::from_le_bytes)
109	}
110
111	#[inline]
112	pub fn time(&self) -> Option<DateTime> {
113		self.field(HAS_TIME).map(DateTime::from_le_bytes)
114	}
115
116	#[inline]
117	pub fn fingerprint(&self) -> Option<RowShapeFingerprint> {
118		self.field(HAS_FINGERPRINT).map(RowShapeFingerprint::from_le_bytes)
119	}
120
121	#[inline]
122	pub fn commit_version(&self) -> Option<u64> {
123		self.field(HAS_COMMIT_VERSION).map(u64::from_le_bytes)
124	}
125
126	#[inline]
127	pub fn partition(&self) -> Option<Partition> {
128		self.field(HAS_PARTITION).map(|bytes| Partition(u128::from_le_bytes(bytes)))
129	}
130
131	#[inline]
132	pub fn body(&self) -> &[u8] {
133		&self.0[self.header_size()..]
134	}
135
136	#[inline]
137	fn field<const N: usize>(&self, bit: u8) -> Option<[u8; N]> {
138		let flags = self.flags();
139		if flags & bit == 0 {
140			return None;
141		}
142		let offset = field_offset(flags, bit);
143		Some(self.0[offset..offset + N].try_into().expect("the envelope header is length-checked"))
144	}
145}
146
147#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
148pub struct EnvelopeBuilder {
149	created_at: Option<DateTime>,
150	updated_at: Option<DateTime>,
151	time: Option<DateTime>,
152	fingerprint: Option<RowShapeFingerprint>,
153	commit_version: Option<u64>,
154	partition: Option<Partition>,
155}
156
157impl EnvelopeBuilder {
158	pub fn new() -> Self {
159		Self::default()
160	}
161
162	pub fn created_at(mut self, created_at: DateTime) -> Self {
163		self.created_at = Some(created_at);
164		self
165	}
166
167	pub fn updated_at(mut self, updated_at: DateTime) -> Self {
168		self.updated_at = Some(updated_at);
169		self
170	}
171
172	pub fn time(mut self, time: DateTime) -> Self {
173		self.time = Some(time);
174		self
175	}
176
177	pub fn fingerprint(mut self, fingerprint: RowShapeFingerprint) -> Self {
178		self.fingerprint = Some(fingerprint);
179		self
180	}
181
182	pub fn commit_version(mut self, commit_version: u64) -> Self {
183		self.commit_version = Some(commit_version);
184		self
185	}
186
187	pub fn partition(mut self, partition: Partition) -> Self {
188		self.partition = Some(partition);
189		self
190	}
191
192	pub fn flags(&self) -> u8 {
193		let mut flags = 0u8;
194		if self.created_at.is_some() {
195			flags |= HAS_CREATED_AT;
196		}
197		if self.updated_at.is_some() {
198			flags |= HAS_UPDATED_AT;
199		}
200		if self.time.is_some() {
201			flags |= HAS_TIME;
202		}
203		if self.fingerprint.is_some() {
204			flags |= HAS_FINGERPRINT;
205		}
206		if self.commit_version.is_some() {
207			flags |= HAS_COMMIT_VERSION;
208		}
209		if self.partition.is_some() {
210			flags |= HAS_PARTITION;
211		}
212		flags
213	}
214
215	pub fn build(self, body: &[u8]) -> EncodedPodRow {
216		let flags = self.flags();
217		let mut buffer = Vec::with_capacity(header_size(flags) + body.len());
218		buffer.push(flags);
219		for stamp in [self.created_at, self.updated_at, self.time].into_iter().flatten() {
220			buffer.extend_from_slice(&stamp.to_le_bytes());
221		}
222		if let Some(fingerprint) = self.fingerprint {
223			buffer.extend_from_slice(&fingerprint.to_le_bytes());
224		}
225		if let Some(commit_version) = self.commit_version {
226			buffer.extend_from_slice(&commit_version.to_le_bytes());
227		}
228		if let Some(partition) = self.partition {
229			buffer.extend_from_slice(&partition.0.to_le_bytes());
230		}
231		buffer.extend_from_slice(body);
232		EncodedPodRow::from(EncodedBytes(CowVec::new(buffer)))
233	}
234}
reifydb_codec/row/envelope/mod.rs

reifydb_codec/row/envelope/
mod.rs
