reifydb_transaction/interceptor/
transaction.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_codec::{key::encoded::EncodedKey, row::bytes::EncodedBytes};
5use reifydb_core::{
6	actors::pending::PendingWrite,
7	common::CommitVersion,
8	interface::{
9		catalog::object::ObjectId,
10		change::{Change, Diff},
11	},
12	key::any::TaggedKey,
13};
14use reifydb_value::Result;
15
16use crate::{
17	TransactionId,
18	change::{RowChange, TransactionalCatalogChanges},
19	interceptor::chain::InterceptorChain,
20};
21
22pub struct PreCommitContext {
23	pub flow_changes: Vec<Change>,
24
25	pub pending_writes: Vec<(TaggedKey, PendingWrite)>,
26
27	pub transaction_writes: Vec<(EncodedKey, Option<EncodedBytes>)>,
28
29	pub view_entries: Vec<(ObjectId, Diff)>,
30}
31
32impl PreCommitContext {
33	pub fn new() -> Self {
34		Self {
35			flow_changes: Vec::new(),
36			pending_writes: Vec::new(),
37			transaction_writes: Vec::new(),
38			view_entries: Vec::new(),
39		}
40	}
41}
42
43impl Default for PreCommitContext {
44	fn default() -> Self {
45		Self::new()
46	}
47}
48
49pub trait PreCommitInterceptor: Send + Sync {
50	fn intercept(&self, ctx: &mut PreCommitContext) -> Result<()>;
51}
52
53impl InterceptorChain<dyn PreCommitInterceptor + Send + Sync> {
54	pub fn execute(&self, ctx: &mut PreCommitContext) -> Result<()> {
55		for interceptor in &self.interceptors {
56			interceptor.intercept(ctx)?;
57		}
58		Ok(())
59	}
60}
61
62pub struct ClosurePreCommitInterceptor<F>
63where
64	F: Fn(&mut PreCommitContext) -> Result<()> + Send + Sync,
65{
66	closure: F,
67}
68
69impl<F> ClosurePreCommitInterceptor<F>
70where
71	F: Fn(&mut PreCommitContext) -> Result<()> + Send + Sync,
72{
73	pub fn new(closure: F) -> Self {
74		Self {
75			closure,
76		}
77	}
78}
79
80impl<F> Clone for ClosurePreCommitInterceptor<F>
81where
82	F: Fn(&mut PreCommitContext) -> Result<()> + Send + Sync + Clone,
83{
84	fn clone(&self) -> Self {
85		Self {
86			closure: self.closure.clone(),
87		}
88	}
89}
90
91impl<F> PreCommitInterceptor for ClosurePreCommitInterceptor<F>
92where
93	F: Fn(&mut PreCommitContext) -> Result<()> + Send + Sync,
94{
95	fn intercept(&self, ctx: &mut PreCommitContext) -> Result<()> {
96		(self.closure)(ctx)
97	}
98}
99
100pub fn pre_commit<F>(f: F) -> ClosurePreCommitInterceptor<F>
101where
102	F: Fn(&mut PreCommitContext) -> Result<()> + Send + Sync + Clone + 'static,
103{
104	ClosurePreCommitInterceptor::new(f)
105}
106
107pub struct PostCommitContext {
108	pub id: TransactionId,
109	pub version: CommitVersion,
110	pub changes: TransactionalCatalogChanges,
111	pub row_changes: Vec<RowChange>,
112}
113
114impl PostCommitContext {
115	pub fn new(
116		id: TransactionId,
117		version: CommitVersion,
118		changes: TransactionalCatalogChanges,
119		row_changes: Vec<RowChange>,
120	) -> Self {
121		Self {
122			id,
123			version,
124			changes,
125			row_changes,
126		}
127	}
128}
129
130pub trait PostCommitInterceptor: Send + Sync {
131	fn intercept(&self, ctx: &mut PostCommitContext) -> Result<()>;
132}
133
134impl InterceptorChain<dyn PostCommitInterceptor + Send + Sync> {
135	pub fn execute(&self, mut ctx: PostCommitContext) -> Result<()> {
136		for interceptor in &self.interceptors {
137			interceptor.intercept(&mut ctx)?;
138		}
139		Ok(())
140	}
141}
142
143pub struct ClosurePostCommitInterceptor<F>
144where
145	F: Fn(&mut PostCommitContext) -> Result<()> + Send + Sync,
146{
147	closure: F,
148}
149
150impl<F> ClosurePostCommitInterceptor<F>
151where
152	F: Fn(&mut PostCommitContext) -> Result<()> + Send + Sync,
153{
154	pub fn new(closure: F) -> Self {
155		Self {
156			closure,
157		}
158	}
159}
160
161impl<F> Clone for ClosurePostCommitInterceptor<F>
162where
163	F: Fn(&mut PostCommitContext) -> Result<()> + Send + Sync + Clone,
164{
165	fn clone(&self) -> Self {
166		Self {
167			closure: self.closure.clone(),
168		}
169	}
170}
171
172impl<F> PostCommitInterceptor for ClosurePostCommitInterceptor<F>
173where
174	F: Fn(&mut PostCommitContext) -> Result<()> + Send + Sync,
175{
176	fn intercept(&self, ctx: &mut PostCommitContext) -> Result<()> {
177		(self.closure)(ctx)
178	}
179}
180
181pub fn post_commit<F>(f: F) -> ClosurePostCommitInterceptor<F>
182where
183	F: Fn(&mut PostCommitContext) -> Result<()> + Send + Sync + Clone + 'static,
184{
185	ClosurePostCommitInterceptor::new(f)
186}
reifydb_transaction/interceptor/transaction.rs

reifydb_transaction/interceptor/
transaction.rs
