reifydb-core 0.10.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

use std::{
	collections::{BTreeMap, BTreeSet},
	ops::Deref,
	sync::{Arc, OnceLock},
};

use reifydb_codec::{key::encoded::EncodedKey, row::bytes::EncodedBytes};
use reifydb_value::value::{datetime::DateTime, row_number::RowNumber};
use serde::{Deserialize, Deserializer, Serialize, Serializer};

use crate::{
	common::{ChangeVersion, CommitVersion},
	interface::{catalog::object::ObjectId, change::Change},
};

#[repr(transparent)]
#[derive(Debug, Clone, PartialOrd, PartialEq, Ord, Eq, Hash)]
pub struct CdcConsumerId(pub(crate) String);

impl CdcConsumerId {
	const FLOW: &'static str = "__FLOW_COORDINATOR";
	const SUBSCRIPTION: &'static str = "__SUBSCRIPTION_CONSUMER";

	pub fn new(id: impl Into<String>) -> Self {
		let id = id.into();
		assert_ne!(id, Self::FLOW);
		assert_ne!(id, Self::SUBSCRIPTION);
		Self(id)
	}

	pub fn flow_consumer() -> Self {
		Self(Self::FLOW.to_string())
	}

	pub fn subscription_consumer() -> Self {
		Self(Self::SUBSCRIPTION.to_string())
	}
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConsumerClass {
	Pinning,
	Ephemeral,
}

impl ConsumerClass {
	pub fn encode(self) -> u8 {
		match self {
			Self::Pinning => 0,
			Self::Ephemeral => 1,
		}
	}

	pub fn decode(byte: u8) -> Option<Self> {
		match byte {
			0 => Some(Self::Pinning),
			1 => Some(Self::Ephemeral),
			_ => None,
		}
	}
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CheckpointState {
	Valid,
	Invalidated,
}

impl CheckpointState {
	pub fn encode(self) -> u8 {
		match self {
			Self::Valid => 0,
			Self::Invalidated => 1,
		}
	}

	pub fn decode(byte: u8) -> Option<Self> {
		match byte {
			0 => Some(Self::Valid),
			1 => Some(Self::Invalidated),
			_ => None,
		}
	}
}

impl AsRef<str> for CdcConsumerId {
	fn as_ref(&self) -> &str {
		&self.0
	}
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum CdcChange {
	Insert {
		key: EncodedKey,
		post: EncodedBytes,
	},
	Update {
		key: EncodedKey,
		pre: EncodedBytes,
		post: EncodedBytes,
	},
	Delete {
		key: EncodedKey,
		pre: Option<EncodedBytes>,
		visible: bool,
	},
}

impl CdcChange {
	pub fn key(&self) -> &EncodedKey {
		match self {
			CdcChange::Insert {
				key,
				..
			} => key,
			CdcChange::Update {
				key,
				..
			} => key,
			CdcChange::Delete {
				key,
				..
			} => key,
		}
	}

	pub fn value_bytes(&self) -> usize {
		match self {
			CdcChange::Insert {
				post,
				..
			} => post.len(),
			CdcChange::Update {
				pre,
				post,
				..
			} => pre.len() + post.len(),
			CdcChange::Delete {
				pre,
				..
			} => pre.as_ref().map(|p| p.len()).unwrap_or(0),
		}
	}
}

#[derive(Debug, Clone)]
pub struct Cdc(Arc<CdcRecord>);

#[derive(Debug, Serialize, Deserialize)]
pub struct CdcRecord {
	pub version: ChangeVersion,
	pub timestamp: DateTime,

	pub changes: Vec<CdcChange>,

	#[serde(skip)]
	changed_objects: OnceLock<BTreeSet<ObjectId>>,

	#[serde(skip)]
	lift: OnceLock<CdcLift>,
}

#[derive(Debug, Clone, Default)]
pub struct CdcLift {
	pub objects: BTreeMap<ObjectId, ObjectLift>,
}

#[derive(Debug, Clone, Default)]
pub struct ObjectLift {
	pub rows: Vec<(usize, RowNumber)>,
	pub first_missing_pre: Option<usize>,
	pub change: OnceLock<Option<Change>>,
}

impl Cdc {
	pub fn new(version: ChangeVersion, timestamp: DateTime, changes: Vec<CdcChange>) -> Self {
		Self(Arc::new(CdcRecord {
			version,
			timestamp,
			changes,
			changed_objects: OnceLock::new(),
			lift: OnceLock::new(),
		}))
	}

	pub fn changed_objects_with(&self, compute: impl FnOnce(&Cdc) -> BTreeSet<ObjectId>) -> &BTreeSet<ObjectId> {
		self.0.changed_objects.get_or_init(|| compute(self))
	}

	pub fn lift_with(&self, build: impl FnOnce(&Cdc) -> CdcLift) -> &CdcLift {
		self.0.lift.get_or_init(|| build(self))
	}
}

impl Deref for Cdc {
	type Target = CdcRecord;

	fn deref(&self) -> &CdcRecord {
		&self.0
	}
}

impl Serialize for Cdc {
	fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
		self.0.serialize(serializer)
	}
}

impl<'de> Deserialize<'de> for Cdc {
	fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
		CdcRecord::deserialize(deserializer).map(|record| Self(Arc::new(record)))
	}
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumerState {
	pub consumer_id: CdcConsumerId,
	pub checkpoint: CommitVersion,
}

#[derive(Debug, Clone)]
pub struct CdcBatch {
	pub items: Vec<Cdc>,

	pub has_more: bool,
}

impl CdcBatch {
	pub fn empty() -> Self {
		Self {
			items: Vec::new(),
			has_more: false,
		}
	}

	pub fn is_empty(&self) -> bool {
		self.items.is_empty()
	}
}

#[cfg(test)]
mod tests {
	use std::cell::Cell;

	use super::*;
	use crate::interface::catalog::id::TableId;

	#[test]
	fn changed_objects_are_computed_once_per_commit() {
		// Every flow actor asks the same shared commit for its objects; a second compute would redo the key
		// decode per flow.
		let cdc = Cdc::new(ChangeVersion::from(CommitVersion(1)), DateTime::from_nanos(1), Vec::new());
		let calls = Cell::new(0);
		let first = BTreeSet::from([ObjectId::Table(TableId(1))]);

		let a = cdc.changed_objects_with(|_| {
			calls.set(calls.get() + 1);
			first.clone()
		});
		assert_eq!(a, &first);
		let b = cdc.changed_objects_with(|_| {
			calls.set(calls.get() + 1);
			BTreeSet::from([ObjectId::Table(TableId(2))])
		});

		assert_eq!(calls.get(), 1, "the second call recomputed instead of reading the memo");
		assert_eq!(b, &first, "the memoised set was replaced by the second compute");
	}
}