reifydb-transaction 0.9.3

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

use std::{
	ops::Bound,
	sync::atomic::{AtomicU64, Ordering},
};

use reifydb_codec::{
	key::encoded::EncodedKey,
	row::{bytes::EncodedBytes, pod::EncodedPodRow},
};
use reifydb_core::{
	interface::store::{SingleVersionGet, SingleVersionRange},
	key::catalog::{DictionaryEntryIndexKey, DictionaryEntryKey, DictionaryKey},
};
use reifydb_store_single::{
	SingleStore,
	tier::{RangeCursor, TierStorage},
};
use reifydb_value::{Result, value::dictionary::DictionaryId};

use crate::single::SingleTransaction;

pub trait DictionaryStore: Send + Sync {
	fn read_committed(&self, key: &EncodedKey) -> Result<Option<EncodedBytes>>;

	fn max_index_id(&self, dictionary: DictionaryId) -> Result<Option<u128>>;

	fn commit_entries(&self, dictionary: DictionaryId, writes: &[DictEntryWrite]) -> Result<()>;
}

pub struct DictEntryWrite {
	pub entry_key: DictionaryEntryKey,
	pub entry_value: EncodedPodRow,
	pub index_key: DictionaryEntryIndexKey,
	pub index_value: EncodedPodRow,
}

pub fn durable_max_index_id(store: &SingleStore, dictionary: DictionaryId) -> Result<Option<u128>> {
	let range = DictionaryEntryIndexKey::full_scan(dictionary).encode();
	match store.persistent() {
		Some(tier) => {
			let mut cursor = RangeCursor::new();
			let batch = tier.range_next(
				&mut cursor,
				bound_as_slice(&range.start),
				bound_as_slice(&range.end),
				1,
			)?;
			Ok(batch.entries
				.first()
				.and_then(|entry| DictionaryEntryIndexKey::decode(&entry.key).map(|key| key.id)))
		}
		None => {
			let batch = SingleVersionRange::range_batch(store, range, 1)?;
			Ok(batch.items
				.first()
				.and_then(|row| DictionaryEntryIndexKey::decode(&row.key).map(|key| key.id)))
		}
	}
}

fn bound_as_slice(bound: &Bound<EncodedKey>) -> Bound<&[u8]> {
	match bound {
		Bound::Included(key) => Bound::Included(key.as_slice()),
		Bound::Excluded(key) => Bound::Excluded(key.as_slice()),
		Bound::Unbounded => Bound::Unbounded,
	}
}

pub struct SingleDictionaryStore {
	single: SingleTransaction,
	reads: AtomicU64,
}

impl SingleDictionaryStore {
	pub fn new(single: SingleTransaction) -> Self {
		Self {
			single,
			reads: AtomicU64::new(0),
		}
	}

	pub fn read_count(&self) -> u64 {
		self.reads.load(Ordering::Relaxed)
	}
}

impl DictionaryStore for SingleDictionaryStore {
	fn read_committed(&self, key: &EncodedKey) -> Result<Option<EncodedBytes>> {
		self.reads.fetch_add(1, Ordering::Relaxed);
		let store = self.single.read_store();
		Ok(SingleVersionGet::get(&store, key)?.map(|bytes| bytes.bytes))
	}

	fn max_index_id(&self, dictionary: DictionaryId) -> Result<Option<u128>> {
		let store = self.single.read_store();
		let batch = SingleVersionRange::range_batch(
			&store,
			DictionaryEntryIndexKey::full_scan(dictionary).encode(),
			1,
		)?;
		match batch.items.first() {
			Some(row) => Ok(DictionaryEntryIndexKey::decode(&row.key).map(|key| key.id)),
			None => Ok(None),
		}
	}

	fn commit_entries(&self, dictionary: DictionaryId, writes: &[DictEntryWrite]) -> Result<()> {
		debug_assert!(!writes.is_empty(), "commit_entries must not be called with no writes");

		let lock_key = DictionaryKey::encoded(dictionary);
		let ranges = vec![
			DictionaryEntryKey::full_scan(dictionary).encode(),
			DictionaryEntryIndexKey::full_scan(dictionary).encode(),
		];
		let mut txn = self.single.begin_command_ranged([&lock_key], ranges)?;
		for write in writes {
			txn.set(&write.index_key, write.index_value.clone())?;
			txn.set(&write.entry_key, write.entry_value.clone())?;
		}
		txn.commit()?;
		Ok(())
	}
}

#[cfg(test)]
mod tests {
	use std::sync::Arc;

	use postcard::to_stdvec;
	use reifydb_core::interface::catalog::{dictionary::Dictionary, id::NamespaceId};
	use reifydb_value::value::{Value, dictionary::DictionaryId, value_type::ValueType};

	use super::SingleDictionaryStore;
	use crate::{dictionary::DictionaryAllocatorRegistry, single::SingleTransaction};

	fn mints() -> Dictionary {
		Dictionary {
			id: DictionaryId(16385),
			namespace: NamespaceId::SYSTEM,
			name: "mints".to_string(),
			value_type: ValueType::Utf8,
			id_type: ValueType::Uint4,
		}
	}

	#[test]
	fn a_cold_registry_resolves_an_interned_value_through_the_single_store() {
		// A registry with an empty cache must find the committed entry in the single store rather
		// than mint a second id for the same value.
		let single = SingleTransaction::testing();
		let dictionary = mints();
		let value = Value::Utf8("GvUCjmWSXA5hrTh9smmNA1AU55YCtP9mDLQcrKA1pump".to_string());

		let warm = DictionaryAllocatorRegistry::new(Arc::new(SingleDictionaryStore::new(single.clone())));
		let first = warm.intern(&dictionary, &value).unwrap();
		assert!(first.created, "first intern must create a new entry");

		let cold = DictionaryAllocatorRegistry::new(Arc::new(SingleDictionaryStore::new(single.clone())));
		let second = cold.intern(&dictionary, &value).unwrap();

		assert_eq!(
			second.id, first.id,
			"a committed entry must be visible to a cold registry through the single store"
		);
		assert!(!second.created, "a cold registry must resolve the durable entry, not remint it");

		let bytes = cold.get(&dictionary, first.id.to_u128()).unwrap().expect("index row must resolve");
		assert_eq!(
			bytes.as_ref(),
			to_stdvec(&value).unwrap().as_slice(),
			"the index row must decode back to the interned value"
		);
	}
}