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;
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 CdcChange {
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		visible: bool,
101	},
102}
103
104impl CdcChange {
105	pub fn key(&self) -> &EncodedKey {
106		match self {
107			CdcChange::Insert {
108				key,
109				..
110			} => key,
111			CdcChange::Update {
112				key,
113				..
114			} => key,
115			CdcChange::Delete {
116				key,
117				..
118			} => key,
119		}
120	}
121
122	pub fn value_bytes(&self) -> usize {
123		match self {
124			CdcChange::Insert {
125				post,
126				..
127			} => post.len(),
128			CdcChange::Update {
129				pre,
130				post,
131				..
132			} => pre.len() + post.len(),
133			CdcChange::Delete {
134				pre,
135				..
136			} => pre.as_ref().map(|p| p.len()).unwrap_or(0),
137		}
138	}
139}
140
141#[derive(Debug, Clone, Serialize, Deserialize)]
142pub struct Cdc {
143	pub version: CommitVersion,
144	pub timestamp: DateTime,
145
146	pub changes: Vec<CdcChange>,
147}
148
149impl Cdc {
150	pub fn new(version: CommitVersion, timestamp: DateTime, changes: Vec<CdcChange>) -> Self {
151		Self {
152			version,
153			timestamp,
154			changes,
155		}
156	}
157}
158
159#[derive(Debug, Clone, PartialEq, Eq)]
160pub struct ConsumerState {
161	pub consumer_id: CdcConsumerId,
162	pub checkpoint: CommitVersion,
163}
164
165#[derive(Debug, Clone)]
166pub struct CdcBatch {
167	pub items: Vec<Cdc>,
168
169	pub has_more: bool,
170}
171
172impl CdcBatch {
173	pub fn empty() -> Self {
174		Self {
175			items: Vec::new(),
176			has_more: false,
177		}
178	}
179
180	pub fn is_empty(&self) -> bool {
181		self.items.is_empty()
182	}
183}
reifydb_core/interface/cdc.rs

reifydb_core/interface/
cdc.rs
