reifydb_engine/transaction/operation/
queue.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_codec::row::bytes::EncodedBytes;
5use reifydb_core::{delta::SetPre, interface::catalog::queue::Queue, key::row::RowKey};
6use reifydb_transaction::{
7	change::{QueueRowInsertion, RowChange},
8	transaction::{Transaction, admin::AdminTransaction, command::CommandTransaction, write::Write},
9};
10use reifydb_value::value::{datetime::DateTime, row_number::RowNumber};
11
12use crate::Result;
13
14#[derive(Debug, Clone)]
15pub struct QueueInsertRow {
16	pub row_number: RowNumber,
17	pub partition: u16,
18	pub key_hash: Option<u64>,
19	pub not_before: Option<DateTime>,
20	pub encoded: EncodedBytes,
21}
22
23pub trait QueueOperations {
24	fn insert_queue(&mut self, queue: &Queue, rows: &[QueueInsertRow]) -> Result<()>;
25}
26
27fn row_changes(queue: &Queue, rows: &[QueueInsertRow]) -> Vec<RowChange> {
28	rows.iter()
29		.map(|row| {
30			RowChange::QueueInsert(QueueRowInsertion {
31				queue_id: queue.id,
32				partition: row.partition,
33				key_hash: row.key_hash,
34				row_number: row.row_number,
35				not_before: row.not_before,
36				encoded: row.encoded.clone(),
37			})
38		})
39		.collect()
40}
41
42fn insert_rows<T: Write>(txn: &mut T, queue: &Queue, rows: &[QueueInsertRow]) -> Result<()> {
43	if rows.is_empty() {
44		return Ok(());
45	}
46
47	txn.set_with_pre(
48		rows.iter()
49			.map(|row| (RowKey::new(queue.id, row.row_number).into(), row.encoded.clone(), SetPre::Absent))
50			.collect(),
51	)?;
52
53	txn.track_row_change(&row_changes(queue, rows));
54
55	Ok(())
56}
57
58impl QueueOperations for CommandTransaction {
59	fn insert_queue(&mut self, queue: &Queue, rows: &[QueueInsertRow]) -> Result<()> {
60		insert_rows(self, queue, rows)
61	}
62}
63
64impl QueueOperations for AdminTransaction {
65	fn insert_queue(&mut self, queue: &Queue, rows: &[QueueInsertRow]) -> Result<()> {
66		insert_rows(self, queue, rows)
67	}
68}
69
70impl QueueOperations for Transaction<'_> {
71	fn insert_queue(&mut self, queue: &Queue, rows: &[QueueInsertRow]) -> Result<()> {
72		match self {
73			Transaction::Command(txn) => txn.insert_queue(queue, rows),
74			Transaction::Admin(txn) => txn.insert_queue(queue, rows),
75			Transaction::Test(t) => t.inner.insert_queue(queue, rows),
76			Transaction::Query(_) => panic!("Write operations not supported on Query transaction"),
77		}
78	}
79}
reifydb_engine/transaction/operation/queue.rs

reifydb_engine/transaction/operation/
queue.rs
