1use std::{ops::Bound, sync::Arc};
5
6use reifydb_codec::{key::encoded::EncodedKey, row::bytes::EncodedBytes};
7use reifydb_core::{
8 actors::pending::PendingWrite,
9 common::CommitVersion,
10 delta::{Delta, RemoveVisibility},
11 execution::ExecutionResult,
12 interface::{
13 catalog::{object::ObjectId, policy::SessionOp, storage::StorageId},
14 change::{Change, ChangeOrigin, Diff},
15 store::{MultiVersionBatch, MultiVersionRow},
16 },
17 key::{
18 any::TaggedKey,
19 bound::TaggedKeyBoundRange,
20 row::{StoragePartitionedRowKey, StorageRowKey},
21 },
22 testing::{CapturedEvent, CapturedInvocation},
23 value::column::columns::Columns,
24};
25use reifydb_value::{Result, error::Diagnostic, params::Params, value::identity::IdentityId};
26
27use crate::{
28 TransactionId,
29 change::{CatalogChangesSavepoint, RowChange},
30 interceptor::{
31 WithInterceptors,
32 authentication::{AuthenticationPostCreateInterceptor, AuthenticationPreDeleteInterceptor},
33 chain::InterceptorChain as Chain,
34 dictionary::{
35 DictionaryPostCreateInterceptor, DictionaryPostUpdateInterceptor,
36 DictionaryPreDeleteInterceptor, DictionaryPreUpdateInterceptor,
37 },
38 dictionary_row::{
39 DictionaryRowPostDeleteInterceptor, DictionaryRowPostInsertInterceptor,
40 DictionaryRowPostUpdateInterceptor, DictionaryRowPreDeleteInterceptor,
41 DictionaryRowPreInsertInterceptor, DictionaryRowPreUpdateInterceptor,
42 },
43 granted_role::{GrantedRolePostCreateInterceptor, GrantedRolePreDeleteInterceptor},
44 identity::{IdentityPostCreateInterceptor, IdentityPreDeleteInterceptor},
45 identity_attribute::{IdentityAttributePostCreateInterceptor, IdentityAttributePreDeleteInterceptor},
46 identity_attribute_value::{
47 IdentityAttributeValuePostCreateInterceptor, IdentityAttributeValuePreDeleteInterceptor,
48 },
49 namespace::{
50 NamespacePostCreateInterceptor, NamespacePostUpdateInterceptor, NamespacePreDeleteInterceptor,
51 NamespacePreUpdateInterceptor,
52 },
53 ringbuffer::{
54 RingBufferPostCreateInterceptor, RingBufferPostUpdateInterceptor,
55 RingBufferPreDeleteInterceptor, RingBufferPreUpdateInterceptor,
56 },
57 ringbuffer_row::{
58 RingBufferRowPostDeleteInterceptor, RingBufferRowPostInsertInterceptor,
59 RingBufferRowPostUpdateInterceptor, RingBufferRowPreDeleteInterceptor,
60 RingBufferRowPreInsertInterceptor, RingBufferRowPreUpdateInterceptor,
61 },
62 role::{RolePostCreateInterceptor, RolePreDeleteInterceptor},
63 series::{
64 SeriesPostCreateInterceptor, SeriesPostUpdateInterceptor, SeriesPreDeleteInterceptor,
65 SeriesPreUpdateInterceptor,
66 },
67 series_row::{
68 SeriesRowPostDeleteInterceptor, SeriesRowPostInsertInterceptor, SeriesRowPostUpdateInterceptor,
69 SeriesRowPreDeleteInterceptor, SeriesRowPreInsertInterceptor, SeriesRowPreUpdateInterceptor,
70 },
71 table::{
72 TablePostCreateInterceptor, TablePostUpdateInterceptor, TablePreDeleteInterceptor,
73 TablePreUpdateInterceptor,
74 },
75 table_row::{
76 TableRowPostDeleteInterceptor, TableRowPostInsertInterceptor, TableRowPostUpdateInterceptor,
77 TableRowPreDeleteInterceptor, TableRowPreInsertInterceptor, TableRowPreUpdateInterceptor,
78 },
79 transaction::{PostCommitInterceptor, PreCommitContext, PreCommitInterceptor},
80 view::{
81 ViewPostCreateInterceptor, ViewPostUpdateInterceptor, ViewPreDeleteInterceptor,
82 ViewPreUpdateInterceptor,
83 },
84 },
85 multi::{RangeScope, transaction::write::WriteSavepoint},
86 single::{SingleTransaction, read::SingleReadTransaction, write::SingleWriteTransaction},
87 transaction::{admin::AdminTransaction, command::CommandTransaction, query::QueryTransaction, write::Write},
88};
89
90pub trait RqlExecutor: Send + Sync {
91 fn rql(&self, tx: &mut Transaction<'_>, rql: &str, params: Params) -> ExecutionResult;
92}
93
94pub mod admin;
95pub mod catalog;
96pub mod command;
97pub mod query;
98pub mod write;
99
100use crate::multi::{pending::PendingWrites, transaction::write::MultiWriteTransaction};
101
102#[inline]
103pub(super) fn collect_transaction_writes(pending: &PendingWrites) -> Vec<(EncodedKey, Option<EncodedBytes>)> {
104 pending.iter()
105 .map(|(key, p)| match &p.delta {
106 Delta::Set {
107 bytes,
108 ..
109 } => (key.encode(), Some(bytes.clone())),
110 _ => (key.encode(), None),
111 })
112 .collect()
113}
114
115#[inline]
116pub(super) fn apply_pre_commit_writes(
117 multi: &mut MultiWriteTransaction,
118 pending_writes: &[(TaggedKey, PendingWrite)],
119) -> Result<()> {
120 for (key, write) in pending_writes {
121 match write {
122 PendingWrite::Set(v) => multi.set(key, v.clone())?,
123 PendingWrite::Remove {
124 announce: RemoveVisibility::Announced,
125 } => multi.remove(key)?,
126 PendingWrite::Remove {
127 announce: RemoveVisibility::Unobserved,
128 } => multi.remove_unobserved(key)?,
129 PendingWrite::Remove {
130 announce: RemoveVisibility::Silent,
131 } => multi.remove_silent(key)?,
132 }
133 }
134 Ok(())
135}
136
137pub struct Savepoint {
138 write: WriteSavepoint,
139 row_changes_len: usize,
140 accumulator_len: usize,
141 changes: CatalogChangesSavepoint,
142}
143
144pub struct TestTransaction<'a> {
145 pub inner: &'a mut AdminTransaction,
146 pub baseline: usize,
147 pub events: &'a mut Vec<CapturedEvent>,
148 pub invocations: &'a mut Vec<CapturedInvocation>,
149 pub event_seq: &'a mut u64,
150 pub handler_seq: &'a mut u64,
151 pub savepoint: Option<Savepoint>,
152 pub session_type: SessionOp,
153 pub session_default_deny: bool,
154}
155
156impl<'a> TestTransaction<'a> {
157 pub fn new(
158 inner: &'a mut AdminTransaction,
159 events: &'a mut Vec<CapturedEvent>,
160 invocations: &'a mut Vec<CapturedInvocation>,
161 event_seq: &'a mut u64,
162 handler_seq: &'a mut u64,
163 session_type: SessionOp,
164 session_default_deny: bool,
165 ) -> Self {
166 let baseline = inner.accumulator.len();
167 let savepoint = Savepoint {
168 write: inner.cmd.as_ref().unwrap().savepoint(),
169 row_changes_len: inner.row_changes.len(),
170 accumulator_len: inner.accumulator.len(),
171 changes: inner.changes.savepoint(),
172 };
173 Self {
174 inner,
175 baseline,
176 events,
177 invocations,
178 event_seq,
179 handler_seq,
180 savepoint: Some(savepoint),
181 session_type,
182 session_default_deny,
183 }
184 }
185
186 pub fn restore(&mut self) {
187 if let Some(sp) = self.savepoint.take() {
188 self.inner.cmd.as_mut().unwrap().restore_savepoint(sp.write);
189 self.inner.row_changes.truncate(sp.row_changes_len);
190 self.inner.accumulator.truncate(sp.accumulator_len);
191 self.inner.changes.restore_savepoint(sp.changes);
192 self.inner.unpoison();
193 }
194 }
195
196 pub fn reborrow(&mut self) -> TestTransaction<'_> {
197 TestTransaction {
198 inner: &mut *self.inner,
199 baseline: self.baseline,
200 events: &mut *self.events,
201 invocations: &mut *self.invocations,
202 event_seq: &mut *self.event_seq,
203 handler_seq: &mut *self.handler_seq,
204 savepoint: None,
205 session_type: self.session_type,
206 session_default_deny: self.session_default_deny,
207 }
208 }
209
210 pub fn accumulator_entries_from(&self) -> &[(ObjectId, Diff)] {
211 self.inner.accumulator.entries_from(self.baseline)
212 }
213
214 pub fn capture_testing_pre_commit(&mut self) -> Result<()> {
215 let has_source_changes = self
216 .inner
217 .accumulator
218 .entries_from(self.baseline)
219 .iter()
220 .any(|(id, _)| !matches!(id, ObjectId::View(_)));
221
222 if !has_source_changes {
223 return Ok(());
224 }
225
226 let offset = self.baseline;
227 let transaction_writes: Vec<(EncodedKey, Option<EncodedBytes>)> = self
228 .inner
229 .pending_writes()
230 .iter()
231 .map(|(key, pending)| match &pending.delta {
232 Delta::Set {
233 bytes,
234 ..
235 } => (key.encode(), Some(bytes.clone())),
236 _ => (key.encode(), None),
237 })
238 .collect();
239
240 let (carried, flow_changes): (Vec<Change>, Vec<Change>) = self
241 .inner
242 .accumulator
243 .take_changes_from(offset, CommitVersion(0), self.inner.clock.now())?
244 .into_iter()
245 .partition(|change| matches!(change.origin, ChangeOrigin::Object(ObjectId::View(_))));
246
247 let mut ctx = PreCommitContext {
248 flow_changes,
249 pending_writes: Vec::new(),
250 transaction_writes,
251 view_entries: Vec::new(),
252 };
253
254 self.inner.interceptors.pre_commit.execute(&mut ctx)?;
255
256 for (key, write) in &ctx.pending_writes {
257 match write {
258 PendingWrite::Set(v) => self.inner.cmd.as_mut().unwrap().set(key, v.clone())?,
259 PendingWrite::Remove {
260 announce: RemoveVisibility::Announced,
261 } => self.inner.cmd.as_mut().unwrap().remove(key)?,
262 PendingWrite::Remove {
263 announce: RemoveVisibility::Unobserved,
264 } => self.inner.cmd.as_mut().unwrap().remove_unobserved(key)?,
265 PendingWrite::Remove {
266 announce: RemoveVisibility::Silent,
267 } => self.inner.cmd.as_mut().unwrap().remove_silent(key)?,
268 }
269 }
270
271 for change in carried {
272 if let ChangeOrigin::Object(id) = change.origin {
273 for diff in change.diffs {
274 self.inner.accumulator.track(id, diff);
275 }
276 }
277 }
278 for (id, diff) in ctx.view_entries {
279 self.inner.accumulator.track(id, diff);
280 }
281
282 Ok(())
283 }
284}
285
286pub enum Transaction<'a> {
287 Command(&'a mut CommandTransaction),
288 Admin(&'a mut AdminTransaction),
289 Query(&'a mut QueryTransaction),
290 Test(Box<TestTransaction<'a>>),
291}
292
293impl<'a> Transaction<'a> {
294 pub fn version(&self) -> CommitVersion {
295 match self {
296 Self::Command(txn) => txn.version(),
297 Self::Admin(txn) => txn.version(),
298 Self::Query(txn) => txn.version(),
299 Self::Test(t) => t.inner.version(),
300 }
301 }
302
303 pub fn id(&self) -> TransactionId {
304 match self {
305 Self::Command(txn) => txn.id(),
306 Self::Admin(txn) => txn.id(),
307 Self::Query(txn) => txn.id(),
308 Self::Test(t) => t.inner.id(),
309 }
310 }
311
312 pub fn has_unprocessed_flow_changes(&self) -> bool {
313 match self {
314 Self::Command(txn) => !txn.accumulator.is_empty(),
315 Self::Admin(txn) => !txn.accumulator.is_empty(),
316 Self::Query(_) | Self::Test(_) => false,
317 }
318 }
319
320 pub fn unprocessed_flow_change_objects(&self) -> Vec<ObjectId> {
321 match self {
322 Self::Command(txn) => txn.accumulator.pending_objects(),
323 Self::Admin(txn) => txn.accumulator.pending_objects(),
324 Self::Query(_) | Self::Test(_) => Vec::new(),
325 }
326 }
327
328 pub fn get<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<Option<MultiVersionRow<TaggedKey>>> {
329 match self {
330 Self::Command(txn) => txn.get(key),
331 Self::Admin(txn) => txn.get(key),
332 Self::Query(txn) => txn.get(key),
333 Self::Test(t) => t.inner.get(key),
334 }
335 }
336
337 pub fn get_committed<K: Into<TaggedKey> + Clone>(
338 &mut self,
339 key: &K,
340 ) -> Result<Option<MultiVersionRow<TaggedKey>>> {
341 match self {
342 Self::Command(txn) => txn.get_committed(key),
343 Self::Admin(txn) => txn.get_committed(key),
344 Self::Query(txn) => txn.get(key),
345 Self::Test(t) => t.inner.get_committed(key),
346 }
347 }
348
349 pub fn contains<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<bool> {
350 match self {
351 Self::Command(txn) => txn.contains(key),
352 Self::Admin(txn) => txn.contains(key),
353 Self::Query(txn) => txn.contains(key),
354 Self::Test(t) => t.inner.contains(key),
355 }
356 }
357
358 pub fn prefix(&mut self, prefix: &EncodedKey) -> Result<MultiVersionBatch<TaggedKey>> {
359 match self {
360 Self::Command(txn) => txn.prefix(prefix),
361 Self::Admin(txn) => txn.prefix(prefix),
362 Self::Query(txn) => txn.prefix(prefix),
363 Self::Test(t) => t.inner.prefix(prefix),
364 }
365 }
366
367 pub fn prefix_rev(&mut self, prefix: &EncodedKey) -> Result<MultiVersionBatch<TaggedKey>> {
368 match self {
369 Self::Command(txn) => txn.prefix_rev(prefix),
370 Self::Admin(txn) => txn.prefix_rev(prefix),
371 Self::Query(txn) => txn.prefix_rev(prefix),
372 Self::Test(t) => t.inner.prefix_rev(prefix),
373 }
374 }
375
376 pub fn read_as_of_version_exclusive(&mut self, version: CommitVersion) -> Result<()> {
377 match self {
378 Transaction::Command(txn) => txn.read_as_of_version_exclusive(version),
379 Transaction::Admin(txn) => txn.read_as_of_version_exclusive(version),
380 Transaction::Query(txn) => txn.read_as_of_version_exclusive(version),
381 Transaction::Test(t) => t.inner.read_as_of_version_exclusive(version),
382 }
383 }
384
385 pub fn range(
386 &mut self,
387 range: TaggedKeyBoundRange,
388 scope: RangeScope,
389 batch_size: usize,
390 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>> {
391 match self {
392 Transaction::Command(txn) => txn.range(range, scope, batch_size),
393 Transaction::Admin(txn) => txn.range(range, scope, batch_size),
394 Transaction::Query(txn) => Ok(txn.range(range, scope, batch_size)),
395 Transaction::Test(t) => t.inner.range(range, scope, batch_size),
396 }
397 }
398
399 pub fn range_row(
400 &mut self,
401 storage: StorageId,
402 start: Bound<StorageRowKey>,
403 end: Bound<StorageRowKey>,
404 scope: RangeScope,
405 batch_size: usize,
406 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<StorageRowKey>>> + Send + '_>> {
407 match self {
408 Transaction::Command(txn) => txn.range_row(storage, start, end, scope, batch_size),
409 Transaction::Admin(txn) => txn.range_row(storage, start, end, scope, batch_size),
410 Transaction::Query(txn) => Ok(txn.range_row(storage, start, end, scope, batch_size)),
411 Transaction::Test(t) => t.inner.range_row(storage, start, end, scope, batch_size),
412 }
413 }
414
415 #[inline]
416 pub fn range_partitioned_row(
417 &mut self,
418 storage: StorageId,
419 start: Bound<StoragePartitionedRowKey>,
420 end: Bound<StoragePartitionedRowKey>,
421 scope: RangeScope,
422 batch_size: usize,
423 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<StoragePartitionedRowKey>>> + Send + '_>> {
424 match self {
425 Transaction::Command(txn) => txn.range_partitioned_row(storage, start, end, scope, batch_size),
426 Transaction::Admin(txn) => txn.range_partitioned_row(storage, start, end, scope, batch_size),
427 Transaction::Query(txn) => {
428 Ok(txn.range_partitioned_row(storage, start, end, scope, batch_size))
429 }
430 Transaction::Test(t) => t.inner.range_partitioned_row(storage, start, end, scope, batch_size),
431 }
432 }
433
434 pub fn range_rev(
435 &mut self,
436 range: TaggedKeyBoundRange,
437 scope: RangeScope,
438 batch_size: usize,
439 ) -> Result<Box<dyn Iterator<Item = Result<MultiVersionRow<TaggedKey>>> + Send + '_>> {
440 match self {
441 Transaction::Command(txn) => txn.range_rev(range, scope, batch_size),
442 Transaction::Admin(txn) => txn.range_rev(range, scope, batch_size),
443 Transaction::Query(txn) => Ok(txn.range_rev(range, scope, batch_size)),
444 Transaction::Test(t) => t.inner.range_rev(range, scope, batch_size),
445 }
446 }
447}
448
449impl<'a> From<&'a mut CommandTransaction> for Transaction<'a> {
450 fn from(txn: &'a mut CommandTransaction) -> Self {
451 Self::Command(txn)
452 }
453}
454
455impl<'a> From<&'a mut AdminTransaction> for Transaction<'a> {
456 fn from(txn: &'a mut AdminTransaction) -> Self {
457 Self::Admin(txn)
458 }
459}
460
461impl<'a> From<&'a mut QueryTransaction> for Transaction<'a> {
462 fn from(txn: &'a mut QueryTransaction) -> Self {
463 Self::Query(txn)
464 }
465}
466
467impl<'a> Transaction<'a> {
468 pub fn identity(&self) -> IdentityId {
469 match self {
470 Self::Command(txn) => txn.identity,
471 Self::Admin(txn) => txn.identity,
472 Self::Query(txn) => txn.identity,
473 Self::Test(t) => t.inner.identity,
474 }
475 }
476
477 pub fn set_identity(&mut self, identity: IdentityId) {
478 match self {
479 Self::Command(txn) => txn.identity = identity,
480 Self::Admin(txn) => txn.identity = identity,
481 Self::Query(txn) => txn.identity = identity,
482 Self::Test(t) => t.inner.identity = identity,
483 }
484 }
485
486 fn executor_clone(&self) -> Option<Arc<dyn RqlExecutor>> {
487 match self {
488 Self::Command(txn) => txn.executor.clone(),
489 Self::Admin(txn) => txn.executor.clone(),
490 Self::Query(txn) => txn.executor.clone(),
491 Self::Test(t) => t.inner.executor.clone(),
492 }
493 }
494
495 pub fn rql(&mut self, rql: &str, params: Params) -> ExecutionResult {
496 let executor = self.executor_clone().expect("RqlExecutor not set");
497 let mut tx = self.reborrow();
498 let result = executor.rql(&mut tx, rql, params);
499 if let Some(ref e) = result.error {
500 self.poison(*e.0.clone());
501 }
502 result
503 }
504
505 fn poison(&mut self, cause: Diagnostic) {
506 match self {
507 Transaction::Command(txn) => txn.poison(cause),
508 Transaction::Admin(txn) => txn.poison(cause),
509 Transaction::Query(_) => {}
510 Transaction::Test(t) => t.inner.poison(cause),
511 }
512 }
513
514 pub fn reborrow(&mut self) -> Transaction<'_> {
515 match self {
516 Transaction::Command(cmd) => Transaction::Command(cmd),
517 Transaction::Admin(admin) => Transaction::Admin(admin),
518 Transaction::Query(qry) => Transaction::Query(qry),
519 Transaction::Test(t) => Transaction::Test(Box::new(TestTransaction {
520 inner: t.inner,
521 baseline: t.baseline,
522 events: t.events,
523 invocations: t.invocations,
524 event_seq: t.event_seq,
525 handler_seq: t.handler_seq,
526 savepoint: None,
527 session_type: t.session_type,
528 session_default_deny: t.session_default_deny,
529 })),
530 }
531 }
532
533 pub fn command(self) -> &'a mut CommandTransaction {
534 match self {
535 Self::Command(txn) => txn,
536 _ => panic!("Expected Command transaction"),
537 }
538 }
539
540 pub fn admin(self) -> &'a mut AdminTransaction {
541 match self {
542 Self::Admin(txn) => txn,
543 Self::Test(t) => t.inner,
544 _ => panic!("Expected Admin transaction"),
545 }
546 }
547
548 pub fn query(self) -> &'a mut QueryTransaction {
549 match self {
550 Self::Query(txn) => txn,
551 _ => panic!("Expected Query transaction"),
552 }
553 }
554
555 pub fn admin_mut(&mut self) -> &mut AdminTransaction {
556 match self {
557 Self::Admin(txn) => txn,
558 Self::Test(t) => t.inner,
559 _ => panic!("Expected Admin transaction"),
560 }
561 }
562
563 pub fn begin_single_query<'b, I>(&self, keys: I) -> Result<SingleReadTransaction<'_>>
564 where
565 I: IntoIterator<Item = &'b EncodedKey>,
566 {
567 match self {
568 Transaction::Command(txn) => txn.begin_single_query(keys),
569 Transaction::Admin(txn) => txn.begin_single_query(keys),
570 Transaction::Query(txn) => txn.begin_single_query(keys),
571 Transaction::Test(t) => t.inner.begin_single_query(keys),
572 }
573 }
574
575 pub fn begin_single_command<'b, I>(&self, keys: I) -> Result<SingleWriteTransaction<'_>>
576 where
577 I: IntoIterator<Item = &'b EncodedKey>,
578 {
579 match self {
580 Transaction::Command(txn) => txn.begin_single_command(keys),
581 Transaction::Admin(txn) => txn.begin_single_command(keys),
582 Transaction::Query(_) => panic!("Write operations not supported on Query transaction"),
583 Transaction::Test(t) => t.inner.begin_single_command(keys),
584 }
585 }
586
587 pub fn single(&self) -> Option<&SingleTransaction> {
588 match self {
589 Transaction::Command(txn) => Some(&txn.single),
590 Transaction::Admin(txn) => Some(&txn.single),
591 Transaction::Query(txn) => txn.single.as_ref(),
592 Transaction::Test(t) => Some(&t.inner.single),
593 }
594 }
595
596 fn write_ops(&mut self) -> &mut dyn Write {
597 match self {
598 Transaction::Command(txn) => &mut **txn,
599 Transaction::Admin(txn) => &mut **txn,
600 Transaction::Query(_) => panic!("Write operations not supported on Query transaction"),
601 Transaction::Test(t) => &mut *t.inner,
602 }
603 }
604
605 pub fn set<K: Into<TaggedKey> + Clone>(&mut self, key: &K, bytes: impl Into<EncodedBytes>) -> Result<()> {
606 Write::set(self.write_ops(), &key.clone().into(), bytes.into())
607 }
608
609 pub fn remove_with_pre<K: Into<TaggedKey> + Clone>(&mut self, key: &K, pre: EncodedBytes) -> Result<()> {
610 Write::remove_with_pre(self.write_ops(), &key.clone().into(), pre)
611 }
612
613 pub fn remove<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<()> {
614 Write::remove(self.write_ops(), &key.clone().into())
615 }
616
617 pub fn mark_preexisting<K: Into<TaggedKey> + Clone>(&mut self, key: &K) -> Result<()> {
618 Write::mark_preexisting(self.write_ops(), &key.clone().into())
619 }
620
621 pub fn track_row_change(&mut self, changes: &[RowChange]) {
622 Write::track_row_change(self.write_ops(), changes)
623 }
624
625 pub fn track_flow_change(&mut self, change: Change) {
626 Write::track_flow_change(self.write_ops(), change)
627 }
628
629 pub fn record_test_event(
630 &mut self,
631 namespace: String,
632 event: String,
633 variant: String,
634 depth: u8,
635 columns: Columns,
636 ) {
637 if let Transaction::Test(t) = self {
638 *t.event_seq += 1;
639 t.events.push(CapturedEvent {
640 sequence: *t.event_seq,
641 namespace,
642 event,
643 variant,
644 depth,
645 columns,
646 });
647 }
648 }
649
650 pub fn record_test_handler(&mut self, mut invocation: CapturedInvocation) {
651 if let Transaction::Test(t) = self {
652 *t.handler_seq += 1;
653 invocation.sequence = *t.handler_seq;
654 t.invocations.push(invocation);
655 }
656 }
657}
658
659macro_rules! delegate_interceptor {
660 ($method:ident, $ret:ty) => {
661 fn $method(&mut self) -> $ret {
662 match self {
663 Transaction::Command(txn) => txn.$method(),
664 Transaction::Admin(txn) => txn.$method(),
665 Transaction::Query(_) => panic!("Interceptors not supported on Query transaction"),
666 Transaction::Test(t) => t.inner.$method(),
667 }
668 }
669 };
670}
671
672impl WithInterceptors for Transaction<'_> {
673 delegate_interceptor!(
674 table_row_pre_insert_interceptors,
675 &mut Chain<dyn TableRowPreInsertInterceptor + Send + Sync>
676 );
677 delegate_interceptor!(
678 table_row_post_insert_interceptors,
679 &mut Chain<dyn TableRowPostInsertInterceptor + Send + Sync>
680 );
681 delegate_interceptor!(
682 table_row_pre_update_interceptors,
683 &mut Chain<dyn TableRowPreUpdateInterceptor + Send + Sync>
684 );
685 delegate_interceptor!(
686 table_row_post_update_interceptors,
687 &mut Chain<dyn TableRowPostUpdateInterceptor + Send + Sync>
688 );
689 delegate_interceptor!(
690 table_row_pre_delete_interceptors,
691 &mut Chain<dyn TableRowPreDeleteInterceptor + Send + Sync>
692 );
693 delegate_interceptor!(
694 table_row_post_delete_interceptors,
695 &mut Chain<dyn TableRowPostDeleteInterceptor + Send + Sync>
696 );
697 delegate_interceptor!(
698 ringbuffer_row_pre_insert_interceptors,
699 &mut Chain<dyn RingBufferRowPreInsertInterceptor + Send + Sync>
700 );
701 delegate_interceptor!(
702 ringbuffer_row_post_insert_interceptors,
703 &mut Chain<dyn RingBufferRowPostInsertInterceptor + Send + Sync>
704 );
705 delegate_interceptor!(
706 ringbuffer_row_pre_update_interceptors,
707 &mut Chain<dyn RingBufferRowPreUpdateInterceptor + Send + Sync>
708 );
709 delegate_interceptor!(
710 ringbuffer_row_post_update_interceptors,
711 &mut Chain<dyn RingBufferRowPostUpdateInterceptor + Send + Sync>
712 );
713 delegate_interceptor!(
714 ringbuffer_row_pre_delete_interceptors,
715 &mut Chain<dyn RingBufferRowPreDeleteInterceptor + Send + Sync>
716 );
717 delegate_interceptor!(
718 ringbuffer_row_post_delete_interceptors,
719 &mut Chain<dyn RingBufferRowPostDeleteInterceptor + Send + Sync>
720 );
721 delegate_interceptor!(pre_commit_interceptors, &mut Chain<dyn PreCommitInterceptor + Send + Sync>);
722 delegate_interceptor!(post_commit_interceptors, &mut Chain<dyn PostCommitInterceptor + Send + Sync>);
723 delegate_interceptor!(
724 namespace_post_create_interceptors,
725 &mut Chain<dyn NamespacePostCreateInterceptor + Send + Sync>
726 );
727 delegate_interceptor!(
728 namespace_pre_update_interceptors,
729 &mut Chain<dyn NamespacePreUpdateInterceptor + Send + Sync>
730 );
731 delegate_interceptor!(
732 namespace_post_update_interceptors,
733 &mut Chain<dyn NamespacePostUpdateInterceptor + Send + Sync>
734 );
735 delegate_interceptor!(
736 namespace_pre_delete_interceptors,
737 &mut Chain<dyn NamespacePreDeleteInterceptor + Send + Sync>
738 );
739 delegate_interceptor!(table_post_create_interceptors, &mut Chain<dyn TablePostCreateInterceptor + Send + Sync>);
740 delegate_interceptor!(table_pre_update_interceptors, &mut Chain<dyn TablePreUpdateInterceptor + Send + Sync>);
741 delegate_interceptor!(table_post_update_interceptors, &mut Chain<dyn TablePostUpdateInterceptor + Send + Sync>);
742 delegate_interceptor!(table_pre_delete_interceptors, &mut Chain<dyn TablePreDeleteInterceptor + Send + Sync>);
743 delegate_interceptor!(view_post_create_interceptors, &mut Chain<dyn ViewPostCreateInterceptor + Send + Sync>);
744 delegate_interceptor!(view_pre_update_interceptors, &mut Chain<dyn ViewPreUpdateInterceptor + Send + Sync>);
745 delegate_interceptor!(view_post_update_interceptors, &mut Chain<dyn ViewPostUpdateInterceptor + Send + Sync>);
746 delegate_interceptor!(view_pre_delete_interceptors, &mut Chain<dyn ViewPreDeleteInterceptor + Send + Sync>);
747 delegate_interceptor!(
748 ringbuffer_post_create_interceptors,
749 &mut Chain<dyn RingBufferPostCreateInterceptor + Send + Sync>
750 );
751 delegate_interceptor!(
752 ringbuffer_pre_update_interceptors,
753 &mut Chain<dyn RingBufferPreUpdateInterceptor + Send + Sync>
754 );
755 delegate_interceptor!(
756 ringbuffer_post_update_interceptors,
757 &mut Chain<dyn RingBufferPostUpdateInterceptor + Send + Sync>
758 );
759 delegate_interceptor!(
760 ringbuffer_pre_delete_interceptors,
761 &mut Chain<dyn RingBufferPreDeleteInterceptor + Send + Sync>
762 );
763 delegate_interceptor!(
764 dictionary_row_pre_insert_interceptors,
765 &mut Chain<dyn DictionaryRowPreInsertInterceptor + Send + Sync>
766 );
767 delegate_interceptor!(
768 dictionary_row_post_insert_interceptors,
769 &mut Chain<dyn DictionaryRowPostInsertInterceptor + Send + Sync>
770 );
771 delegate_interceptor!(
772 dictionary_row_pre_update_interceptors,
773 &mut Chain<dyn DictionaryRowPreUpdateInterceptor + Send + Sync>
774 );
775 delegate_interceptor!(
776 dictionary_row_post_update_interceptors,
777 &mut Chain<dyn DictionaryRowPostUpdateInterceptor + Send + Sync>
778 );
779 delegate_interceptor!(
780 dictionary_row_pre_delete_interceptors,
781 &mut Chain<dyn DictionaryRowPreDeleteInterceptor + Send + Sync>
782 );
783 delegate_interceptor!(
784 dictionary_row_post_delete_interceptors,
785 &mut Chain<dyn DictionaryRowPostDeleteInterceptor + Send + Sync>
786 );
787 delegate_interceptor!(
788 dictionary_post_create_interceptors,
789 &mut Chain<dyn DictionaryPostCreateInterceptor + Send + Sync>
790 );
791 delegate_interceptor!(
792 dictionary_pre_update_interceptors,
793 &mut Chain<dyn DictionaryPreUpdateInterceptor + Send + Sync>
794 );
795 delegate_interceptor!(
796 dictionary_post_update_interceptors,
797 &mut Chain<dyn DictionaryPostUpdateInterceptor + Send + Sync>
798 );
799 delegate_interceptor!(
800 dictionary_pre_delete_interceptors,
801 &mut Chain<dyn DictionaryPreDeleteInterceptor + Send + Sync>
802 );
803 delegate_interceptor!(
804 series_row_pre_insert_interceptors,
805 &mut Chain<dyn SeriesRowPreInsertInterceptor + Send + Sync>
806 );
807 delegate_interceptor!(
808 series_row_post_insert_interceptors,
809 &mut Chain<dyn SeriesRowPostInsertInterceptor + Send + Sync>
810 );
811 delegate_interceptor!(
812 series_row_pre_update_interceptors,
813 &mut Chain<dyn SeriesRowPreUpdateInterceptor + Send + Sync>
814 );
815 delegate_interceptor!(
816 series_row_post_update_interceptors,
817 &mut Chain<dyn SeriesRowPostUpdateInterceptor + Send + Sync>
818 );
819 delegate_interceptor!(
820 series_row_pre_delete_interceptors,
821 &mut Chain<dyn SeriesRowPreDeleteInterceptor + Send + Sync>
822 );
823 delegate_interceptor!(
824 series_row_post_delete_interceptors,
825 &mut Chain<dyn SeriesRowPostDeleteInterceptor + Send + Sync>
826 );
827 delegate_interceptor!(
828 series_post_create_interceptors,
829 &mut Chain<dyn SeriesPostCreateInterceptor + Send + Sync>
830 );
831 delegate_interceptor!(series_pre_update_interceptors, &mut Chain<dyn SeriesPreUpdateInterceptor + Send + Sync>);
832 delegate_interceptor!(
833 series_post_update_interceptors,
834 &mut Chain<dyn SeriesPostUpdateInterceptor + Send + Sync>
835 );
836 delegate_interceptor!(series_pre_delete_interceptors, &mut Chain<dyn SeriesPreDeleteInterceptor + Send + Sync>);
837 delegate_interceptor!(
838 identity_post_create_interceptors,
839 &mut Chain<dyn IdentityPostCreateInterceptor + Send + Sync>
840 );
841 delegate_interceptor!(
842 identity_pre_delete_interceptors,
843 &mut Chain<dyn IdentityPreDeleteInterceptor + Send + Sync>
844 );
845 delegate_interceptor!(role_post_create_interceptors, &mut Chain<dyn RolePostCreateInterceptor + Send + Sync>);
846 delegate_interceptor!(role_pre_delete_interceptors, &mut Chain<dyn RolePreDeleteInterceptor + Send + Sync>);
847 delegate_interceptor!(
848 granted_role_post_create_interceptors,
849 &mut Chain<dyn GrantedRolePostCreateInterceptor + Send + Sync>
850 );
851 delegate_interceptor!(
852 granted_role_pre_delete_interceptors,
853 &mut Chain<dyn GrantedRolePreDeleteInterceptor + Send + Sync>
854 );
855 delegate_interceptor!(
856 identity_attribute_post_create_interceptors,
857 &mut Chain<dyn IdentityAttributePostCreateInterceptor + Send + Sync>
858 );
859 delegate_interceptor!(
860 identity_attribute_pre_delete_interceptors,
861 &mut Chain<dyn IdentityAttributePreDeleteInterceptor + Send + Sync>
862 );
863 delegate_interceptor!(
864 identity_attribute_value_post_create_interceptors,
865 &mut Chain<dyn IdentityAttributeValuePostCreateInterceptor + Send + Sync>
866 );
867 delegate_interceptor!(
868 identity_attribute_value_pre_delete_interceptors,
869 &mut Chain<dyn IdentityAttributeValuePreDeleteInterceptor + Send + Sync>
870 );
871 delegate_interceptor!(
872 authentication_post_create_interceptors,
873 &mut Chain<dyn AuthenticationPostCreateInterceptor + Send + Sync>
874 );
875 delegate_interceptor!(
876 authentication_pre_delete_interceptors,
877 &mut Chain<dyn AuthenticationPreDeleteInterceptor + Send + Sync>
878 );
879}