reifydb_transaction/transaction/catalog/
migration.rs1use reifydb_core::interface::catalog::{
5 change::{CatalogTrackMigrationChangeOperations, CatalogTrackMigrationEventChangeOperations},
6 id::MigrationId,
7 migration::{MigrationDef, MigrationEvent},
8};
9use reifydb_type::Result;
10
11use crate::{
12 change::{
13 Change,
14 OperationType::{Create, Delete},
15 TransactionalMigrationChanges,
16 },
17 transaction::{admin::AdminTransaction, subscription::SubscriptionTransaction},
18};
19
20impl CatalogTrackMigrationChangeOperations for AdminTransaction {
21 fn track_migration_def_created(&mut self, migration: MigrationDef) -> Result<()> {
22 let change = Change {
23 pre: None,
24 post: Some(migration),
25 op: Create,
26 };
27 self.changes.add_migration_def_change(change);
28 Ok(())
29 }
30
31 fn track_migration_def_deleted(&mut self, migration: MigrationDef) -> Result<()> {
32 let change = Change {
33 pre: Some(migration),
34 post: None,
35 op: Delete,
36 };
37 self.changes.add_migration_def_change(change);
38 Ok(())
39 }
40}
41
42impl CatalogTrackMigrationEventChangeOperations for AdminTransaction {
43 fn track_migration_event_created(&mut self, event: MigrationEvent) -> Result<()> {
44 let change = Change {
45 pre: None,
46 post: Some(event),
47 op: Create,
48 };
49 self.changes.add_migration_event_change(change);
50 Ok(())
51 }
52}
53
54impl TransactionalMigrationChanges for AdminTransaction {
55 fn find_migration(&self, id: MigrationId) -> Option<&MigrationDef> {
56 for change in self.changes.migration_def.iter().rev() {
57 if let Some(migration) = &change.post {
58 if migration.id == id {
59 return Some(migration);
60 }
61 } else if let Some(migration) = &change.pre {
62 if migration.id == id && change.op == Delete {
63 return None;
64 }
65 }
66 }
67 None
68 }
69
70 fn find_migration_by_name(&self, name: &str) -> Option<&MigrationDef> {
71 self.changes
72 .migration_def
73 .iter()
74 .rev()
75 .find_map(|change| change.post.as_ref().filter(|m| m.name == name))
76 }
77
78 fn is_migration_deleted(&self, id: MigrationId) -> bool {
79 self.changes
80 .migration_def
81 .iter()
82 .rev()
83 .any(|change| change.op == Delete && change.pre.as_ref().map(|m| m.id == id).unwrap_or(false))
84 }
85
86 fn is_migration_deleted_by_name(&self, name: &str) -> bool {
87 self.changes.migration_def.iter().rev().any(|change| {
88 change.op == Delete && change.pre.as_ref().map(|m| m.name == name).unwrap_or(false)
89 })
90 }
91}
92
93impl CatalogTrackMigrationChangeOperations for SubscriptionTransaction {
94 fn track_migration_def_created(&mut self, migration: MigrationDef) -> Result<()> {
95 self.inner.track_migration_def_created(migration)
96 }
97
98 fn track_migration_def_deleted(&mut self, migration: MigrationDef) -> Result<()> {
99 self.inner.track_migration_def_deleted(migration)
100 }
101}
102
103impl CatalogTrackMigrationEventChangeOperations for SubscriptionTransaction {
104 fn track_migration_event_created(&mut self, event: MigrationEvent) -> Result<()> {
105 self.inner.track_migration_event_created(event)
106 }
107}
108
109impl TransactionalMigrationChanges for SubscriptionTransaction {
110 fn find_migration(&self, id: MigrationId) -> Option<&MigrationDef> {
111 self.inner.find_migration(id)
112 }
113
114 fn find_migration_by_name(&self, name: &str) -> Option<&MigrationDef> {
115 self.inner.find_migration_by_name(name)
116 }
117
118 fn is_migration_deleted(&self, id: MigrationId) -> bool {
119 self.inner.is_migration_deleted(id)
120 }
121
122 fn is_migration_deleted_by_name(&self, name: &str) -> bool {
123 self.inner.is_migration_deleted_by_name(name)
124 }
125}