reifydb-transaction 0.9.3

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

use std::sync::Arc;

use crossbeam_skiplist::SkipMap;
use reifydb_codec::{
	key::encoded::{EncodedKey, EncodedKeyRange},
	row::bytes::EncodedBytes,
};
use reifydb_core::{delta::Delta, event::EventBus, interface::WithEventBus};
use reifydb_runtime::sync::rwlock::{ArcRwLock, RwLock};
use reifydb_store_single::SingleStore;

pub mod read;
pub mod write;

use read::{KeyReadLock, SingleReadTransaction};
use reifydb_runtime::{actor::system::ActorSystem, context::clock::Clock};
use reifydb_value::Result;
use write::{KeyWriteLock, SingleWriteTransaction};

#[derive(Clone)]
pub struct SingleTransaction {
	inner: Arc<SingleTransactionInner>,
}

pub(crate) struct SingleTransactionInner {
	pub(crate) store: RwLock<SingleStore>,
	pub(crate) event_bus: EventBus,
	pub(crate) key_locks: SkipMap<EncodedKey, ArcRwLock<()>>,
}

impl SingleTransactionInner {
	fn get_or_create_lock(&self, key: &EncodedKey) -> ArcRwLock<()> {
		if let Some(entry) = self.key_locks.get(key) {
			return entry.value().clone();
		}

		self.key_locks.get_or_insert(key.clone(), ArcRwLock::new(())).value().clone()
	}
}

impl SingleTransaction {
	pub fn new(store: SingleStore, event_bus: EventBus) -> Self {
		Self {
			inner: Arc::new(SingleTransactionInner {
				store: RwLock::new(store),
				event_bus,
				key_locks: SkipMap::new(),
			}),
		}
	}

	pub fn read_store(&self) -> SingleStore {
		self.inner.store.read().clone()
	}

	pub fn testing() -> Self {
		let actor_system = ActorSystem::testing(Clock::Real);
		let spawner = actor_system.spawner();
		Self::new(SingleStore::testing_memory(), EventBus::new(&spawner))
	}

	pub fn with_query<'a, I, F, R>(&self, keys: I, f: F) -> Result<R>
	where
		I: IntoIterator<Item = &'a EncodedKey>,
		F: FnOnce(&mut SingleReadTransaction<'_>) -> Result<R>,
	{
		let mut tx = self.begin_query(keys)?;
		f(&mut tx)
	}

	pub fn with_command<'a, I, F, R>(&self, keys: I, f: F) -> Result<R>
	where
		I: IntoIterator<Item = &'a EncodedKey>,
		F: FnOnce(&mut SingleWriteTransaction<'_>) -> Result<R>,
	{
		let mut tx = self.begin_command(keys)?;
		let result = f(&mut tx)?;
		tx.commit()?;
		Ok(result)
	}

	pub fn begin_query<'a, I>(&self, keys: I) -> Result<SingleReadTransaction<'_>>
	where
		I: IntoIterator<Item = &'a EncodedKey>,
	{
		let mut keys_vec: Vec<EncodedKey> = keys.into_iter().cloned().collect();
		assert!(
			!keys_vec.is_empty(),
			"SVL transactions must declare keys upfront - empty keysets are not allowed"
		);

		keys_vec.sort();

		let mut locks = Vec::new();
		for key in &keys_vec {
			let lock = self.inner.get_or_create_lock(key);
			locks.push(KeyReadLock::new(lock));
		}

		Ok(SingleReadTransaction {
			inner: &self.inner,
			keys: keys_vec,
			_key_locks: locks,
		})
	}

	pub fn begin_command<'a, I>(&self, keys: I) -> Result<SingleWriteTransaction<'_>>
	where
		I: IntoIterator<Item = &'a EncodedKey>,
	{
		self.begin_command_ranged(keys, Vec::new())
	}

	pub fn begin_command_ranged<'a, I>(
		&self,
		lock_keys: I,
		ranges: Vec<EncodedKeyRange>,
	) -> Result<SingleWriteTransaction<'_>>
	where
		I: IntoIterator<Item = &'a EncodedKey>,
	{
		let mut keys_vec: Vec<EncodedKey> = lock_keys.into_iter().cloned().collect();
		assert!(
			!keys_vec.is_empty(),
			"SVL transactions must declare keys upfront - empty keysets are not allowed"
		);

		keys_vec.sort();

		let mut locks = Vec::new();
		for key in &keys_vec {
			let lock = self.inner.get_or_create_lock(key);
			locks.push(KeyWriteLock::new(lock));
		}

		Ok(SingleWriteTransaction::new(&self.inner, keys_vec, ranges, locks))
	}
}

impl WithEventBus for SingleTransaction {
	fn event_bus(&self) -> &EventBus {
		&self.inner.event_bus
	}
}

#[cfg(test)]
pub mod tests {
	use std::{
		iter,
		ops::Bound,
		sync::{Arc, Barrier},
		thread,
	};

	use reifydb_core::{interface::catalog::id::QueueId, key::queue::QueueDeduplicationKey};
	use reifydb_value::{util::cowvec::CowVec, value::duration::Duration};

	use super::*;

	fn make_key(s: &str) -> QueueDeduplicationKey {
		QueueDeduplicationKey::new(QueueId(1), s.as_bytes().iter().map(|b| !b).collect::<Vec<u8>>())
	}

	fn make_value(s: &str) -> EncodedBytes {
		EncodedBytes(CowVec::new(s.as_bytes().to_vec()))
	}

	fn create_test_svl() -> SingleTransaction {
		SingleTransaction::testing()
	}

	#[test]
	fn test_allowed_key_query() {
		let svl = create_test_svl();
		let key = make_key("test_key");

		let mut tx = svl.begin_query(vec![&key.encode()]).unwrap();

		let result = tx.get(&key.encode());
		assert!(result.is_ok());
	}

	#[test]
	fn test_disallowed_key_query() {
		let svl = create_test_svl();
		let key1 = make_key("allowed");
		let key2 = make_key("disallowed");

		let mut tx = svl.begin_query(vec![&key1.encode()]).unwrap();

		assert!(tx.get(&key1.encode()).is_ok());

		let result = tx.get(&key2.encode());
		assert!(result.is_err());
		let err = result.unwrap_err();
		assert_eq!(err.0.code, "TXN_010");
	}

	#[test]
	#[should_panic(expected = "SVL transactions must declare keys upfront - empty keysets are not allowed")]
	fn test_empty_keyset_query_panics() {
		let svl = create_test_svl();

		let _tx = svl.begin_query(iter::empty());
	}

	#[test]
	#[should_panic(expected = "SVL transactions must declare keys upfront - empty keysets are not allowed")]
	fn test_empty_keyset_command_panics() {
		let svl = create_test_svl();

		let _tx = svl.begin_command(iter::empty());
	}

	#[test]
	fn test_allowed_key_command() {
		let svl = create_test_svl();
		let key = make_key("test_key");
		let value = make_value("test_value");

		let mut tx = svl.begin_command(vec![&key.encode()]).unwrap();

		assert!(tx.set(&key, value.clone()).is_ok());
		assert!(tx.get(&key).is_ok());
		assert!(tx.commit().is_ok());
	}

	#[test]
	fn test_disallowed_key_command() {
		let svl = create_test_svl();
		let key1 = make_key("allowed");
		let key2 = make_key("disallowed");
		let value = make_value("test_value");

		let mut tx = svl.begin_command(vec![&key1.encode()]).unwrap();

		assert!(tx.set(&key1, value.clone()).is_ok());

		let result = tx.set(&key2, value);
		assert!(result.is_err());
		let err = result.unwrap_err();
		assert_eq!(err.0.code, "TXN_010");
	}

	#[test]
	fn test_ranged_command_allows_keys_in_declared_range() {
		let svl = create_test_svl();
		let lock_key = make_key("lock");
		let in_range = make_key("range_b");
		let value = make_value("test_value");

		// Only the coarse lock key is locked; writes are scoped to the declared range.
		let range = EncodedKeyRange::new(
			Bound::Included(make_key("range_a").encode()),
			Bound::Excluded(make_key("range_z").encode()),
		);
		let mut tx = svl.begin_command_ranged(vec![&lock_key.encode()], vec![range]).unwrap();

		assert!(tx.set(&in_range, value.clone()).is_ok());
		assert!(tx.commit().is_ok());

		let mut rx = svl.begin_query(vec![&in_range.encode()]).unwrap();
		let row = rx.get(&in_range.encode()).unwrap().unwrap();
		assert_eq!(row.bytes, value);
	}

	#[test]
	fn test_ranged_command_rejects_keys_outside_range_and_lock_set() {
		let svl = create_test_svl();
		let lock_key = make_key("lock");
		let outside = make_key("zzz_outside");
		let value = make_value("test_value");

		let range = EncodedKeyRange::new(
			Bound::Included(make_key("range_a").encode()),
			Bound::Excluded(make_key("range_z").encode()),
		);
		let mut tx = svl.begin_command_ranged(vec![&lock_key.encode()], vec![range]).unwrap();

		let result = tx.set(&outside, value);
		assert!(result.is_err());
		let err = result.unwrap_err();
		assert_eq!(err.0.code, "TXN_010");
	}

	#[test]
	fn test_ranged_command_still_allows_exact_lock_keys() {
		let svl = create_test_svl();
		let lock_key = make_key("lock");
		let value = make_value("test_value");

		let range = EncodedKeyRange::new(
			Bound::Included(make_key("range_a").encode()),
			Bound::Excluded(make_key("range_z").encode()),
		);
		let mut tx = svl.begin_command_ranged(vec![&lock_key.encode()], vec![range]).unwrap();

		// A declared lock key stays writable even though it falls outside the range.
		assert!(tx.set(&lock_key, value).is_ok());
		assert!(tx.commit().is_ok());
	}

	#[test]
	fn test_command_commit_with_valid_keys() {
		let svl = create_test_svl();
		let key1 = make_key("key1");
		let key2 = make_key("key2");
		let value1 = make_value("value1");
		let value2 = make_value("value2");

		{
			let mut tx = svl.begin_command(vec![&key1.encode(), &key2.encode()]).unwrap();
			tx.set(&key1, value1.clone()).unwrap();
			tx.set(&key2, value2.clone()).unwrap();
			tx.commit().unwrap();
		}

		{
			let mut tx = svl.begin_query(vec![&key1.encode(), &key2.encode()]).unwrap();
			let result1 = tx.get(&key1.encode()).unwrap();
			let result2 = tx.get(&key2.encode()).unwrap();
			assert!(result1.is_some());
			assert!(result2.is_some());
			assert_eq!(result1.unwrap().bytes, value1);
			assert_eq!(result2.unwrap().bytes, value2);
		}
	}

	#[test]
	fn test_rollback_with_scoped_keys() {
		let svl = create_test_svl();
		let key = make_key("test_key");
		let value = make_value("test_value");

		{
			let mut tx = svl.begin_command(vec![&key.encode()]).unwrap();
			tx.set(&key, value).unwrap();
			tx.rollback().unwrap();
		}

		{
			let mut tx = svl.begin_query(vec![&key.encode()]).unwrap();
			let result = tx.get(&key.encode()).unwrap();
			assert!(result.is_none());
		}
	}

	#[test]
	fn test_concurrent_reads() {
		let svl = Arc::new(create_test_svl());
		let key = make_key("shared_key");
		let value = make_value("shared_value");

		{
			let mut tx = svl.begin_command(vec![&key.encode()]).unwrap();
			tx.set(&key, value.clone()).unwrap();
			tx.commit().unwrap();
		}

		let mut handles = vec![];
		for _ in 0..5 {
			let svl_clone = Arc::clone(&svl);
			let key_clone = key.clone();
			let value_clone = value.clone();

			let handle = thread::spawn(move || {
				let mut tx = svl_clone.begin_query(vec![&key_clone.encode()]).unwrap();
				let result = tx.get(&key_clone.encode()).unwrap();
				assert!(result.is_some());
				assert_eq!(result.unwrap().bytes, value_clone);
			});
			handles.push(handle);
		}

		for handle in handles {
			handle.join().unwrap();
		}
	}

	#[test]
	fn test_concurrent_writers_disjoint_keys() {
		let svl = Arc::new(create_test_svl());

		let mut handles = vec![];
		for i in 0..5 {
			let svl_clone = Arc::clone(&svl);
			let key = make_key(&format!("key_{}", i));
			let value = make_value(&format!("value_{}", i));

			let handle = thread::spawn(move || {
				let mut tx = svl_clone.begin_command(vec![&key.encode()]).unwrap();
				tx.set(&key, value).unwrap();
				tx.commit().unwrap();
			});
			handles.push(handle);
		}

		for handle in handles {
			handle.join().unwrap();
		}

		for i in 0..5 {
			let key = make_key(&format!("key_{}", i));
			let expected_value = make_value(&format!("value_{}", i));

			let mut tx = svl.begin_query(vec![&key.encode()]).unwrap();
			let result = tx.get(&key.encode()).unwrap();
			assert!(result.is_some());
			assert_eq!(result.unwrap().bytes, expected_value);
		}
	}

	#[test]
	fn test_concurrent_readers_and_writer() {
		let svl = Arc::new(create_test_svl());
		let key1 = make_key("key1");
		let key2 = make_key("key2");
		let value1 = make_value("value1");
		let value2 = make_value("value2");

		{
			let mut tx = svl.begin_command(vec![&key1.encode(), &key2.encode()]).unwrap();
			tx.set(&key1, value1.clone()).unwrap();
			tx.set(&key2, value2.clone()).unwrap();
			tx.commit().unwrap();
		}

		let mut handles = vec![];
		for _ in 0..3 {
			let svl_clone = Arc::clone(&svl);
			let key_clone = key1.clone();
			let value_clone = value1.clone();

			let handle = thread::spawn(move || {
				let mut tx = svl_clone.begin_query(vec![&key_clone.encode()]).unwrap();
				let result = tx.get(&key_clone.encode()).unwrap();
				assert!(result.is_some());
				assert_eq!(result.unwrap().bytes, value_clone);
			});
			handles.push(handle);
		}

		// A writer on a different key must not block these readers.
		let svl_clone = Arc::clone(&svl);
		let new_value = make_value("new_value2");
		let handle = thread::spawn(move || {
			let mut tx = svl_clone.begin_command(vec![&key2.encode()]).unwrap();
			tx.set(&key2, new_value).unwrap();
			tx.commit().unwrap();
		});
		handles.push(handle);

		for handle in handles {
			handle.join().unwrap();
		}
	}

	#[test]
	fn test_no_panics_with_rwlock() {
		let svl = Arc::new(create_test_svl());

		let mut handles = vec![];
		for i in 0..10 {
			let svl_clone = Arc::clone(&svl);
			let key = make_key(&format!("key_{}", i % 3)); // keys overlap across threads
			let value = make_value(&format!("value_{}", i));

			let handle = thread::spawn(move || {
				if i % 2 == 0 {
					let mut tx = svl_clone.begin_command(vec![&key.encode()]).unwrap();
					let _ = tx.set(&key, value);
					let _ = tx.commit();
				} else {
					let mut tx = svl_clone.begin_query(vec![&key.encode()]).unwrap();
					let _ = tx.get(&key.encode());
				}
			});
			handles.push(handle);
		}

		for handle in handles {
			handle.join().unwrap();
		}
	}

	#[test]
	fn test_write_blocks_concurrent_write() {
		let svl = Arc::new(create_test_svl());
		let key = make_key("blocking_key");
		let barrier = Arc::new(Barrier::new(2));

		let svl1 = Arc::clone(&svl);
		let key1 = key.clone();
		let barrier1 = Arc::clone(&barrier);
		let handle1 = thread::spawn(move || {
			let mut tx = svl1.begin_command(vec![&key1.encode()]).unwrap();
			tx.set(&key1, make_value("value1")).unwrap();

			// Signal that the write lock is held.
			barrier1.wait();

			// Hold the lock so the second writer has to block on it.
			thread::sleep(Duration::from_milliseconds(100).unwrap().to_std());

			tx.commit().unwrap();
		});

		let svl2 = Arc::clone(&svl);
		let key2 = key.clone();
		let barrier2 = Arc::clone(&barrier);
		let handle2 = thread::spawn(move || {
			barrier2.wait();

			// Give thread 1 time to enter its sleep still holding the lock.
			thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());

			let mut tx = svl2.begin_command(vec![&key2.encode()]).unwrap();
			tx.set(&key2, make_value("value2")).unwrap();
			tx.commit().unwrap();
		});

		handle1.join().unwrap();
		handle2.join().unwrap();

		// Thread 2 could only have written after thread 1 released, so its value must survive.
		let mut tx = svl.begin_query(vec![&key.encode()]).unwrap();
		let result = tx.get(&key.encode()).unwrap();
		assert!(result.is_some());
		assert_eq!(result.unwrap().bytes, make_value("value2"));
	}

	#[test]
	fn test_write_blocks_concurrent_read() {
		let svl = Arc::new(create_test_svl());
		let key = make_key("blocking_key");

		{
			let mut tx = svl.begin_command(vec![&key.encode()]).unwrap();
			tx.set(&key, make_value("initial")).unwrap();
			tx.commit().unwrap();
		}

		let barrier = Arc::new(Barrier::new(2));

		let svl1 = Arc::clone(&svl);
		let key1 = key.clone();
		let barrier1 = Arc::clone(&barrier);
		let handle1 = thread::spawn(move || {
			let mut tx = svl1.begin_command(vec![&key1.encode()]).unwrap();
			tx.set(&key1, make_value("updated")).unwrap();

			// Signal that the write lock is held.
			barrier1.wait();

			// Hold the lock so the reader has to block on it.
			thread::sleep(Duration::from_milliseconds(100).unwrap().to_std());

			tx.commit().unwrap();
		});

		let svl2 = Arc::clone(&svl);
		let key2 = key.clone();
		let barrier2 = Arc::clone(&barrier);
		let handle2 = thread::spawn(move || {
			barrier2.wait();

			// Give thread 1 time to enter its sleep still holding the lock.
			thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());

			let mut tx = svl2.begin_query(vec![&key2.encode()]).unwrap();
			let result = tx.get(&key2.encode()).unwrap();

			// A reader that truly blocked sees the committed value, never the initial one.
			assert!(result.is_some());
			assert_eq!(result.unwrap().bytes, make_value("updated"));
		});

		handle1.join().unwrap();
		handle2.join().unwrap();
	}

	#[test]
	fn test_concurrent_reads_allowed() {
		let svl = Arc::new(create_test_svl());
		let key = make_key("shared_read_key");

		{
			let mut tx = svl.begin_command(vec![&key.encode()]).unwrap();
			tx.set(&key, make_value("shared")).unwrap();
			tx.commit().unwrap();
		}

		let barrier = Arc::new(Barrier::new(3));
		let mut handles = vec![];

		for _ in 0..3 {
			let svl_clone = Arc::clone(&svl);
			let key_clone = key.clone();
			let barrier_clone = Arc::clone(&barrier);

			let handle = thread::spawn(move || {
				let mut tx = svl_clone.begin_query(vec![&key_clone.encode()]).unwrap();

				// Every reader holds its read lock past this point; a mutually exclusive
				// read lock would deadlock here rather than fail an assertion.
				barrier_clone.wait();

				let result = tx.get(&key_clone.encode()).unwrap();
				assert!(result.is_some());
				assert_eq!(result.unwrap().bytes, make_value("shared"));

				thread::sleep(Duration::from_milliseconds(50).unwrap().to_std());
			});
			handles.push(handle);
		}

		for handle in handles {
			handle.join().unwrap();
		}
	}

	#[test]
	fn test_overlapping_keys_different_order() {
		// The two threads declare the same keys in opposite order; begin_command sorts them, so
		// there is no lock-order cycle to deadlock on.
		let svl = Arc::new(create_test_svl());
		let key1 = make_key("deadlock_key1");
		let key2 = make_key("deadlock_key2");
		let barrier = Arc::new(Barrier::new(2));

		let svl1 = Arc::clone(&svl);
		let key1_clone = key1.clone();
		let key2_clone = key2.clone();
		let barrier1 = Arc::clone(&barrier);
		let handle1 = thread::spawn(move || {
			barrier1.wait();
			let mut tx = svl1.begin_command(vec![&key1_clone.encode(), &key2_clone.encode()]).unwrap();
			tx.set(&key1_clone, make_value("from_thread1")).unwrap();
			thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());
			tx.commit().unwrap();
		});

		let svl2 = Arc::clone(&svl);
		let key1_clone2 = key1.clone();
		let key2_clone2 = key2.clone();
		let barrier2 = Arc::clone(&barrier);
		let handle2 = thread::spawn(move || {
			barrier2.wait();
			let mut tx = svl2.begin_command(vec![&key2_clone2.encode(), &key1_clone2.encode()]).unwrap();
			tx.set(&key2_clone2, make_value("from_thread2")).unwrap();
			thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());
			tx.commit().unwrap();
		});

		handle1.join().unwrap();
		handle2.join().unwrap();

		let mut tx = svl.begin_query(vec![&key1.encode(), &key2.encode()]).unwrap();
		let result1 = tx.get(&key1.encode()).unwrap();
		let result2 = tx.get(&key2.encode()).unwrap();
		assert!(result1.is_some());
		assert!(result2.is_some());
	}

	#[test]
	fn test_circular_dependency_three_transactions() {
		// The three key sets chain into a cycle (1,2) (2,3) (3,1); only sorted acquisition keeps
		// that from becoming a deadlock.
		let svl = Arc::new(create_test_svl());
		let key1 = make_key("circular_key1");
		let key2 = make_key("circular_key2");
		let key3 = make_key("circular_key3");
		let barrier = Arc::new(Barrier::new(3));

		let svl1 = Arc::clone(&svl);
		let k1_1 = key1.clone();
		let k2_1 = key2.clone();
		let barrier1 = Arc::clone(&barrier);
		let handle1 = thread::spawn(move || {
			barrier1.wait();
			let mut tx = svl1.begin_command(vec![&k1_1.encode(), &k2_1.encode()]).unwrap();
			tx.set(&k1_1, make_value("t1")).unwrap();
			thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());
			tx.commit().unwrap();
		});

		let svl2 = Arc::clone(&svl);
		let k2_2 = key2.clone();
		let k3_2 = key3.clone();
		let barrier2 = Arc::clone(&barrier);
		let handle2 = thread::spawn(move || {
			barrier2.wait();
			let mut tx = svl2.begin_command(vec![&k2_2.encode(), &k3_2.encode()]).unwrap();
			tx.set(&k2_2, make_value("t2")).unwrap();
			thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());
			tx.commit().unwrap();
		});

		let svl3 = Arc::clone(&svl);
		let barrier3 = Arc::clone(&barrier);
		let handle3 = thread::spawn(move || {
			barrier3.wait();
			let mut tx = svl3.begin_command(vec![&key3.encode(), &key1.encode()]).unwrap();
			tx.set(&key3, make_value("t3")).unwrap();
			thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());
			tx.commit().unwrap();
		});

		handle1.join().unwrap();
		handle2.join().unwrap();
		handle3.join().unwrap();
	}

	#[test]
	fn test_locks_released_on_drop() {
		let svl = Arc::new(create_test_svl());
		let key = make_key("drop_test_key");

		let svl1 = Arc::clone(&svl);
		let key_clone = key.clone();
		let handle1 = thread::spawn(move || {
			let mut tx = svl1.begin_command(vec![&key_clone.encode()]).unwrap();
			tx.set(&key_clone, make_value("dropped")).unwrap();
			// Dropped here without commit.
		});

		handle1.join().unwrap();

		thread::sleep(Duration::from_milliseconds(10).unwrap().to_std());

		// A lock not released on drop makes this block forever rather than fail.
		let svl2 = Arc::clone(&svl);
		let key_clone2 = key.clone();
		let handle2 = thread::spawn(move || {
			let mut tx = svl2.begin_command(vec![&key_clone2.encode()]).unwrap();
			tx.set(&key_clone2, make_value("success")).unwrap();
			tx.commit().unwrap();
		});

		handle2.join().unwrap();

		let mut tx = svl.begin_query(vec![&key.encode()]).unwrap();
		let result = tx.get(&key.encode()).unwrap();
		assert!(result.is_some());
		assert_eq!(result.unwrap().bytes, make_value("success"));
	}
}