reifydb_transaction/transaction/catalog/
ringbuffer.rs1use reifydb_core::interface::catalog::{
5 change::CatalogTrackRingBufferChangeOperations,
6 id::{NamespaceId, RingBufferId},
7 ringbuffer::RingBuffer,
8};
9use reifydb_value::Result;
10
11use crate::{
12 change::{
13 Change,
14 OperationType::{Create, Delete},
15 TransactionalRingBufferChanges,
16 },
17 transaction::admin::AdminTransaction,
18};
19
20impl CatalogTrackRingBufferChangeOperations for AdminTransaction {
21 fn track_ringbuffer_created(&mut self, ringbuffer: RingBuffer) -> Result<()> {
22 let change = Change {
23 pre: None,
24 post: Some(ringbuffer),
25 op: Create,
26 };
27 self.changes.add_ringbuffer_change(change);
28 Ok(())
29 }
30
31 fn track_ringbuffer_deleted(&mut self, ringbuffer: RingBuffer) -> Result<()> {
32 let change = Change {
33 pre: Some(ringbuffer),
34 post: None,
35 op: Delete,
36 };
37 self.changes.add_ringbuffer_change(change);
38 Ok(())
39 }
40}
41
42impl TransactionalRingBufferChanges for AdminTransaction {
43 fn find_ringbuffer(&self, id: RingBufferId) -> Option<&RingBuffer> {
44 for change in self.changes.ringbuffer.iter().rev() {
45 if let Some(ringbuffer) = &change.post
46 && ringbuffer.id == id
47 {
48 return Some(ringbuffer);
49 }
50 if let Some(ringbuffer) = &change.pre
51 && ringbuffer.id == id && change.op == Delete
52 {
53 return None;
54 }
55 }
56 None
57 }
58
59 fn find_ringbuffer_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&RingBuffer> {
60 for change in self.changes.ringbuffer.iter().rev() {
61 if let Some(ringbuffer) = &change.post
62 && ringbuffer.namespace == namespace
63 && ringbuffer.name == name
64 {
65 return Some(ringbuffer);
66 }
67 if let Some(ringbuffer) = &change.pre
68 && ringbuffer.namespace == namespace
69 && ringbuffer.name == name && change.op == Delete
70 {
71 return None;
72 }
73 }
74 None
75 }
76
77 fn is_ringbuffer_deleted(&self, id: RingBufferId) -> bool {
78 self.changes
79 .ringbuffer
80 .iter()
81 .any(|change| change.op == Delete && change.pre.as_ref().map(|rb| rb.id == id).unwrap_or(false))
82 }
83
84 fn is_ringbuffer_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool {
85 self.changes.ringbuffer.iter().any(|change| {
86 change.op == Delete
87 && change
88 .pre
89 .as_ref()
90 .map(|rb| rb.namespace == namespace && rb.name == name)
91 .unwrap_or(false)
92 })
93 }
94}