reifydb_transaction/transaction/catalog/
series.rs1use reifydb_core::interface::catalog::{
5 change::CatalogTrackSeriesChangeOperations,
6 id::{NamespaceId, SeriesId},
7 series::SeriesDef,
8};
9use reifydb_type::Result;
10
11use crate::{
12 change::{
13 Change,
14 OperationType::{Create, Delete, Update},
15 TransactionalSeriesChanges,
16 },
17 transaction::{admin::AdminTransaction, subscription::SubscriptionTransaction},
18};
19
20impl CatalogTrackSeriesChangeOperations for AdminTransaction {
21 fn track_series_def_created(&mut self, series: SeriesDef) -> Result<()> {
22 let change = Change {
23 pre: None,
24 post: Some(series),
25 op: Create,
26 };
27 self.changes.add_series_def_change(change);
28 Ok(())
29 }
30
31 fn track_series_def_updated(&mut self, pre: SeriesDef, post: SeriesDef) -> Result<()> {
32 let change = Change {
33 pre: Some(pre),
34 post: Some(post),
35 op: Update,
36 };
37 self.changes.add_series_def_change(change);
38 Ok(())
39 }
40
41 fn track_series_def_deleted(&mut self, series: SeriesDef) -> Result<()> {
42 let change = Change {
43 pre: Some(series),
44 post: None,
45 op: Delete,
46 };
47 self.changes.add_series_def_change(change);
48 Ok(())
49 }
50}
51
52impl TransactionalSeriesChanges for AdminTransaction {
53 fn find_series(&self, id: SeriesId) -> Option<&SeriesDef> {
54 for change in self.changes.series_def.iter().rev() {
55 if let Some(series) = &change.post {
56 if series.id == id {
57 return Some(series);
58 }
59 }
60 if let Some(series) = &change.pre {
61 if series.id == id && change.op == Delete {
62 return None;
63 }
64 }
65 }
66 None
67 }
68
69 fn find_series_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&SeriesDef> {
70 for change in self.changes.series_def.iter().rev() {
71 if let Some(series) = &change.post {
72 if series.namespace == namespace && series.name == name {
73 return Some(series);
74 }
75 }
76 if let Some(series) = &change.pre {
77 if series.namespace == namespace && series.name == name && change.op == Delete {
78 return None;
79 }
80 }
81 }
82 None
83 }
84
85 fn is_series_deleted(&self, id: SeriesId) -> bool {
86 self.changes
87 .series_def
88 .iter()
89 .any(|change| change.op == Delete && change.pre.as_ref().map(|s| s.id == id).unwrap_or(false))
90 }
91
92 fn is_series_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool {
93 self.changes.series_def.iter().any(|change| {
94 change.op == Delete
95 && change
96 .pre
97 .as_ref()
98 .map(|s| s.namespace == namespace && s.name == name)
99 .unwrap_or(false)
100 })
101 }
102}
103
104impl CatalogTrackSeriesChangeOperations for SubscriptionTransaction {
105 fn track_series_def_created(&mut self, series: SeriesDef) -> Result<()> {
106 self.inner.track_series_def_created(series)
107 }
108
109 fn track_series_def_updated(&mut self, pre: SeriesDef, post: SeriesDef) -> Result<()> {
110 self.inner.track_series_def_updated(pre, post)
111 }
112
113 fn track_series_def_deleted(&mut self, series: SeriesDef) -> Result<()> {
114 self.inner.track_series_def_deleted(series)
115 }
116}
117
118impl TransactionalSeriesChanges for SubscriptionTransaction {
119 fn find_series(&self, id: SeriesId) -> Option<&SeriesDef> {
120 self.inner.find_series(id)
121 }
122
123 fn find_series_by_name(&self, namespace: NamespaceId, name: &str) -> Option<&SeriesDef> {
124 self.inner.find_series_by_name(namespace, name)
125 }
126
127 fn is_series_deleted(&self, id: SeriesId) -> bool {
128 self.inner.is_series_deleted(id)
129 }
130
131 fn is_series_deleted_by_name(&self, namespace: NamespaceId, name: &str) -> bool {
132 self.inner.is_series_deleted_by_name(namespace, name)
133 }
134}