reifydb_core/interface/
cdc.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_codec::{key::encoded::EncodedKey, row::bytes::EncodedBytes};
5use reifydb_value::value::datetime::DateTime;
6use serde::{Deserialize, Serialize};
7
8use crate::{common::CommitVersion, interface::change::Change};
9
10#[repr(transparent)]
11#[derive(Debug, Clone, PartialOrd, PartialEq, Ord, Eq, Hash)]
12pub struct CdcConsumerId(pub(crate) String);
13
14impl CdcConsumerId {
15	const FLOW: &'static str = "__FLOW_COORDINATOR";
16	const SUBSCRIPTION: &'static str = "__SUBSCRIPTION_CONSUMER";
17
18	pub fn new(id: impl Into<String>) -> Self {
19		let id = id.into();
20		assert_ne!(id, Self::FLOW);
21		assert_ne!(id, Self::SUBSCRIPTION);
22		Self(id)
23	}
24
25	pub fn flow_consumer() -> Self {
26		Self(Self::FLOW.to_string())
27	}
28
29	pub fn subscription_consumer() -> Self {
30		Self(Self::SUBSCRIPTION.to_string())
31	}
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub enum ConsumerClass {
36	Pinning,
37	Ephemeral,
38}
39
40impl ConsumerClass {
41	pub fn encode(self) -> u8 {
42		match self {
43			Self::Pinning => 0,
44			Self::Ephemeral => 1,
45		}
46	}
47
48	pub fn decode(byte: u8) -> Option<Self> {
49		match byte {
50			0 => Some(Self::Pinning),
51			1 => Some(Self::Ephemeral),
52			_ => None,
53		}
54	}
55}
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq)]
58pub enum CheckpointState {
59	Valid,
60	Invalidated,
61}
62
63impl CheckpointState {
64	pub fn encode(self) -> u8 {
65		match self {
66			Self::Valid => 0,
67			Self::Invalidated => 1,
68		}
69	}
70
71	pub fn decode(byte: u8) -> Option<Self> {
72		match byte {
73			0 => Some(Self::Valid),
74			1 => Some(Self::Invalidated),
75			_ => None,
76		}
77	}
78}
79
80impl AsRef<str> for CdcConsumerId {
81	fn as_ref(&self) -> &str {
82		&self.0
83	}
84}
85
86#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
87pub enum SystemChange {
88	Insert {
89		key: EncodedKey,
90		post: EncodedBytes,
91	},
92	Update {
93		key: EncodedKey,
94		pre: EncodedBytes,
95		post: EncodedBytes,
96	},
97	Delete {
98		key: EncodedKey,
99		pre: Option<EncodedBytes>,
100	},
101}
102
103impl SystemChange {
104	pub fn key(&self) -> &EncodedKey {
105		match self {
106			SystemChange::Insert {
107				key,
108				..
109			} => key,
110			SystemChange::Update {
111				key,
112				..
113			} => key,
114			SystemChange::Delete {
115				key,
116				..
117			} => key,
118		}
119	}
120
121	pub fn value_bytes(&self) -> usize {
122		match self {
123			SystemChange::Insert {
124				post,
125				..
126			} => post.len(),
127			SystemChange::Update {
128				pre,
129				post,
130				..
131			} => pre.len() + post.len(),
132			SystemChange::Delete {
133				pre,
134				..
135			} => pre.as_ref().map(|p| p.len()).unwrap_or(0),
136		}
137	}
138}
139
140#[derive(Debug, Clone, Serialize, Deserialize)]
141pub struct Cdc {
142	pub version: CommitVersion,
143	pub timestamp: DateTime,
144
145	pub changes: Vec<Change>,
146
147	pub system_changes: Vec<SystemChange>,
148}
149
150impl Cdc {
151	pub fn new(
152		version: CommitVersion,
153		timestamp: DateTime,
154		changes: Vec<Change>,
155		system_changes: Vec<SystemChange>,
156	) -> Self {
157		Self {
158			version,
159			timestamp,
160			changes,
161			system_changes,
162		}
163	}
164}
165
166#[derive(Debug, Clone, PartialEq, Eq)]
167pub struct ConsumerState {
168	pub consumer_id: CdcConsumerId,
169	pub checkpoint: CommitVersion,
170}
171
172#[derive(Debug, Clone)]
173pub struct CdcBatch {
174	pub items: Vec<Cdc>,
175
176	pub has_more: bool,
177}
178
179impl CdcBatch {
180	pub fn empty() -> Self {
181		Self {
182			items: Vec::new(),
183			has_more: false,
184		}
185	}
186
187	pub fn is_empty(&self) -> bool {
188		self.items.is_empty()
189	}
190}
reifydb_core/interface/cdc.rs

reifydb_core/interface/
cdc.rs
