reifydb-core 0.9.1

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

use std::borrow::Cow;

use reifydb_codec::key::{deserializer::KeyDeserializer, encoded::EncodedKey, serializer::KeySerializer};
use reifydb_macro::KeyCodec;
use reifydb_value::value::Value;
use smallvec::{SmallVec, smallvec};

use super::KeyTag;
use crate::{
	interface::catalog::{id::RingBufferId, object::ObjectId, storage::StorageId},
	key::{
		any::{Field, KeyFields, RawEncoding, Width, encode_values},
		bound::{TaggedKeyBoundRange, object_fields},
		catalog::{KeyDeserializerCatalogExt, KeySerializerCatalogExt},
	},
};

#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
#[key(tag = RingBuffer)]
pub struct RingBufferKey {
	pub ringbuffer: RingBufferId,
}

impl RingBufferKey {
	pub fn new(ringbuffer: RingBufferId) -> Self {
		Self {
			ringbuffer,
		}
	}

	pub fn encoded(ringbuffer: impl Into<RingBufferId>) -> EncodedKey {
		Self::new(ringbuffer.into()).encode()
	}

	pub fn full_scan() -> TaggedKeyBoundRange {
		TaggedKeyBoundRange::kind(Self::TAG)
	}
}

#[derive(Debug, Clone, PartialEq, Hash)]
pub struct RingBufferMetadataKey {
	pub storage: StorageId,
	pub partition_values: Vec<Value>,
}

impl RingBufferMetadataKey {
	pub fn new(storage: impl Into<StorageId>) -> Self {
		Self {
			storage: storage.into(),
			partition_values: vec![],
		}
	}

	pub fn encoded(storage: impl Into<StorageId>) -> EncodedKey {
		Self::new(storage).encode()
	}

	pub fn partition(storage: impl Into<StorageId>, partition_values: Vec<Value>) -> Self {
		Self {
			storage: storage.into(),
			partition_values,
		}
	}

	pub fn encoded_partition(storage: impl Into<StorageId>, partition_values: Vec<Value>) -> EncodedKey {
		Self::partition(storage, partition_values).encode()
	}

	pub fn full_scan_for_storage(storage: impl Into<StorageId>) -> TaggedKeyBoundRange {
		TaggedKeyBoundRange::prefix(Self::TAG, object_fields(ObjectId::from(storage.into())))
	}
}

impl RingBufferMetadataKey {
	pub const TAG: KeyTag = KeyTag::RingBufferMetadata;

	pub fn encode(&self) -> EncodedKey {
		let mut serializer = KeySerializer::with_capacity(32);
		serializer.extend_u8(Self::TAG as u8).extend_object_id(self.storage);
		for value in &self.partition_values {
			serializer.extend_value(value);
		}
		serializer.to_encoded_key()
	}

	pub fn decode(key: &EncodedKey) -> Option<Self> {
		let mut de = KeyDeserializer::from_bytes(key.as_slice());

		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
		if kind != Self::TAG {
			return None;
		}

		let storage = StorageId::from_object(de.read_object_id().ok()?)?;

		let mut partition_values = Vec::new();
		while !de.is_empty() {
			partition_values.push(de.read_value().ok()?);
		}

		Some(Self {
			storage,
			partition_values,
		})
	}
}

#[cfg(test)]
mod tests {
	use std::ops::RangeBounds;

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

	#[test]
	fn test_metadata_key_encode_decode_roundtrip() {
		let key = RingBufferMetadataKey::encoded_partition(
			RingBufferId(42),
			vec![Value::Utf8("east".to_string())],
		);
		let mut de = KeyDeserializer::from_bytes(key.as_slice());
		let _ = (de.read_u8(), de.read_object_id());
		let value = de.read_value().unwrap();
		assert_eq!(value, Value::Utf8("east".to_string()));
	}

	#[test]
	fn test_metadata_key_encode_decode_multiple() {
		let key = RingBufferMetadataKey::encoded_partition(
			RingBufferId(7),
			vec![Value::Utf8("us".to_string()), Value::Uint8(42)],
		);
		let mut de = KeyDeserializer::from_bytes(key.as_slice());
		let _ = (de.read_u8(), de.read_object_id());
		assert_eq!(de.read_value().unwrap(), Value::Utf8("us".to_string()));
		assert_eq!(de.read_value().unwrap(), Value::Uint8(42));
	}

	#[test]
	fn test_metadata_key_roundtrip_ringbuffer() {
		// The tag byte is what keeps a ring buffer's metadata out of a view's; a bare id would collide.
		let key = RingBufferMetadataKey {
			storage: StorageId::RingBuffer(RingBufferId(42)),
			partition_values: vec![Value::Utf8("east".to_string())],
		};
		assert_eq!(RingBufferMetadataKey::decode(&key.encode()).unwrap(), key);
	}

	#[test]
	fn test_metadata_key_roundtrip_view() {
		// A ring-buffer-backed view keeps its own metadata under its own id, not a backing object's.
		let key = RingBufferMetadataKey {
			storage: StorageId::View(ViewId(42)),
			partition_values: vec![Value::Utf8("east".to_string())],
		};
		assert_eq!(RingBufferMetadataKey::decode(&key.encode()).unwrap(), key);
	}

	#[test]
	fn test_full_scan_for_storage_excludes_a_view_with_the_same_id() {
		// Ring buffer 42 and view 42 share a numeric id, so only the tag byte separates their scans.
		let range = RingBufferMetadataKey::full_scan_for_storage(RingBufferId(42)).encode();
		let ringbuffer = RingBufferMetadataKey::encoded(RingBufferId(42));
		let view = RingBufferMetadataKey::encoded(ViewId(42));
		assert!(range.contains(&ringbuffer));
		assert!(!range.contains(&view));
	}

	#[test]
	fn test_ring_buffer_key_matches_legacy_byte_layout() {
		for id in [RingBufferId(0), RingBufferId(1), RingBufferId(u64::MAX)] {
			let mut legacy = KeySerializer::with_capacity(9);
			legacy.extend_u8(KeyTag::RingBuffer as u8).extend_u64(id);
			assert_eq!(legacy.to_encoded_key().as_slice(), RingBufferKey::encoded(id).as_slice());
		}
	}
}

impl KeyFields for RingBufferMetadataKey {
	fn fields(&self) -> SmallVec<[Field<'_>; 6]> {
		smallvec![
			Field::UAsc(Width::U8, ObjectId::from(self.storage).type_tag() as u128),
			Field::UDesc(Width::U64, ObjectId::from(self.storage).as_u64() as u128),
			Field::RawAsc(
				RawEncoding::Verbatim,
				Cow::Owned(encode_values(&self.partition_values).as_slice().to_vec())
			),
		]
	}
}