reifydb_sub_server/
interceptor.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc};
5
6use reifydb_core::{actors::server::Operation, metrics::execution::ExecutionMetrics};
7use reifydb_value::{params::Params, value::identity::IdentityId};
8
9use crate::execute::ExecuteError;
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
12pub enum Protocol {
13	#[default]
14	Http,
15	WebSocket,
16	Grpc,
17}
18
19#[derive(Debug, Clone, Default)]
20pub struct RequestMetadata {
21	headers: HashMap<String, String>,
22	protocol: Protocol,
23}
24
25impl RequestMetadata {
26	pub fn new(protocol: Protocol) -> Self {
27		Self {
28			headers: HashMap::new(),
29			protocol,
30		}
31	}
32
33	pub fn insert(&mut self, key: impl Into<String>, value: impl Into<String>) {
34		self.headers.insert(key.into().to_ascii_lowercase(), value.into());
35	}
36
37	pub fn get(&self, key: &str) -> Option<&str> {
38		self.headers.get(&key.to_ascii_lowercase()).map(|s| s.as_str())
39	}
40
41	pub fn protocol(&self) -> Protocol {
42		self.protocol
43	}
44
45	pub fn headers(&self) -> &HashMap<String, String> {
46		&self.headers
47	}
48}
49
50pub struct RequestContext {
51	pub identity: IdentityId,
52
53	pub operation: Operation,
54
55	pub rql: String,
56
57	pub params: Params,
58
59	pub metadata: RequestMetadata,
60}
61
62pub struct ResponseContext {
63	pub identity: IdentityId,
64
65	pub operation: Operation,
66
67	pub rql: String,
68
69	pub metrics: ExecutionMetrics,
70
71	pub params: Params,
72
73	pub metadata: RequestMetadata,
74
75	pub result: Result<usize, String>,
76}
77
78pub trait RequestInterceptor: Send + Sync + 'static {
79	fn pre_execute<'a>(
80		&'a self,
81		ctx: &'a mut RequestContext,
82	) -> Pin<Box<dyn Future<Output = Result<(), ExecuteError>> + Send + 'a>>;
83
84	fn post_execute<'a>(&'a self, ctx: &'a ResponseContext) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>>;
85}
86
87#[derive(Clone)]
88pub struct RequestInterceptorChain {
89	interceptors: Arc<Vec<Arc<dyn RequestInterceptor>>>,
90}
91
92impl RequestInterceptorChain {
93	pub fn new(interceptors: Vec<Arc<dyn RequestInterceptor>>) -> Self {
94		Self {
95			interceptors: Arc::new(interceptors),
96		}
97	}
98
99	pub fn empty() -> Self {
100		Self {
101			interceptors: Arc::new(Vec::new()),
102		}
103	}
104
105	pub fn is_empty(&self) -> bool {
106		self.interceptors.is_empty()
107	}
108
109	pub async fn pre_execute(&self, ctx: &mut RequestContext) -> Result<(), ExecuteError> {
110		for interceptor in self.interceptors.iter() {
111			interceptor.pre_execute(ctx).await?;
112		}
113		Ok(())
114	}
115
116	pub async fn post_execute(&self, ctx: &ResponseContext) {
117		for interceptor in self.interceptors.iter().rev() {
118			interceptor.post_execute(ctx).await;
119		}
120	}
121}
122
123impl Default for RequestInterceptorChain {
124	fn default() -> Self {
125		Self::empty()
126	}
127}
reifydb_sub_server/interceptor.rs

reifydb_sub_server/
interceptor.rs
