reifydb_sub_server/subscription/
create.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_core::interface::catalog::{id::SubscriptionId, subscription::HydrationConfig};
5use reifydb_value::value::duration::Duration;
6#[cfg(not(reifydb_single_threaded))]
7use reifydb_value::value::frame::{column::FrameColumn, frame::Frame};
8#[cfg(not(reifydb_single_threaded))]
9use reifydb_value::{
10	params::Params,
11	value::{Value, identity::IdentityId},
12};
13#[cfg(not(reifydb_single_threaded))]
14use tracing::{debug, error};
15
16pub enum CreateSubscriptionResult {
17	Local {
18		id: SubscriptionId,
19		hydration: HydrationConfig,
20		throttle: Option<Duration>,
21		linger: Option<Duration>,
22	},
23	Remote {
24		address: String,
25		body: String,
26		token: Option<String>,
27		hydration: HydrationConfig,
28		throttle: Option<Duration>,
29		linger: Option<Duration>,
30	},
31}
32
33#[cfg(not(reifydb_single_threaded))]
34use reifydb_core::actors::server::Operation;
35
36#[cfg(not(reifydb_single_threaded))]
37use crate::{
38	dispatch::dispatch_subscribe,
39	interceptor::{RequestContext, RequestMetadata},
40	state::AppState,
41	subscription::errors::CreateSubscriptionError,
42};
43
44#[cfg(not(reifydb_single_threaded))]
45pub async fn create_subscription(
46	state: &AppState,
47	identity: IdentityId,
48	rql: &str,
49	params: Params,
50	metadata: RequestMetadata,
51) -> Result<CreateSubscriptionResult, CreateSubscriptionError> {
52	debug!("Subscription rql: {}", rql);
53
54	let ctx = RequestContext {
55		identity,
56		operation: Operation::Subscribe,
57		rql: rql.to_string(),
58		params,
59		metadata,
60	};
61	let (frames, _metrics) = dispatch_subscribe(state, ctx).await?;
62	let frame = frames.first().ok_or(CreateSubscriptionError::ExtractionFailed)?;
63
64	if let Some(remote) = extract_remote_result(frame)? {
65		return Ok(remote);
66	}
67	extract_local_result(frame)
68}
69
70#[cfg(not(reifydb_single_threaded))]
71fn extract_remote_result(frame: &Frame) -> Result<Option<CreateSubscriptionResult>, CreateSubscriptionError> {
72	let Some(addr_col) = frame.columns.iter().find(|c| c.name == "remote_address") else {
73		return Ok(None);
74	};
75	let address = match first_utf8_value(addr_col) {
76		Some(s) => s,
77		None => return Err(CreateSubscriptionError::ExtractionFailed),
78	};
79	let body = frame
80		.columns
81		.iter()
82		.find(|c| c.name == "remote_body")
83		.and_then(first_utf8_value)
84		.ok_or(CreateSubscriptionError::ExtractionFailed)?;
85	let token = frame.columns.iter().find(|c| c.name == "remote_token").and_then(first_utf8_value);
86	let enabled = first_bool_value(frame, "remote_hydration_enabled").unwrap_or(true);
87	let max_rows = first_uint8_value(frame, "remote_hydration_max_rows");
88	let throttle = first_uint8_value(frame, "remote_throttle_ms")
89		.map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
90	let linger =
91		first_uint8_value(frame, "remote_linger_ms").map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
92	Ok(Some(CreateSubscriptionResult::Remote {
93		address,
94		body,
95		token,
96		hydration: HydrationConfig {
97			enabled,
98			max_rows,
99		},
100		throttle,
101		linger,
102	}))
103}
104
105#[cfg(not(reifydb_single_threaded))]
106fn extract_local_result(frame: &Frame) -> Result<CreateSubscriptionResult, CreateSubscriptionError> {
107	let id = frame
108		.columns
109		.iter()
110		.find(|c| c.name == "subscription_id")
111		.and_then(|col| {
112			if col.data.is_empty() {
113				None
114			} else {
115				Some(col.data.get_value(0))
116			}
117		})
118		.and_then(|value| match value {
119			Value::Uint8(id) => Some(SubscriptionId(id)),
120			other => {
121				error!("subscription_id column has wrong type: {:?}", other);
122				None
123			}
124		})
125		.ok_or(CreateSubscriptionError::ExtractionFailed)?;
126
127	let enabled = first_bool_value(frame, "hydration_enabled").unwrap_or(true);
128	let max_rows = first_uint8_value(frame, "hydration_max_rows");
129	let throttle =
130		first_uint8_value(frame, "throttle_ms").map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
131	let linger = first_uint8_value(frame, "linger_ms").map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
132
133	Ok(CreateSubscriptionResult::Local {
134		id,
135		hydration: HydrationConfig {
136			enabled,
137			max_rows,
138		},
139		throttle,
140		linger,
141	})
142}
143
144#[cfg(not(reifydb_single_threaded))]
145#[inline]
146fn first_utf8_value(col: &FrameColumn) -> Option<String> {
147	if col.data.is_empty() {
148		return None;
149	}
150	match col.data.get_value(0) {
151		Value::Utf8(s) => Some(s),
152		_ => None,
153	}
154}
155
156#[cfg(not(reifydb_single_threaded))]
157#[inline]
158fn first_bool_value(frame: &Frame, name: &str) -> Option<bool> {
159	let col = frame.columns.iter().find(|c| c.name == name)?;
160	if col.data.is_empty() {
161		return None;
162	}
163	match col.data.get_value(0) {
164		Value::Boolean(b) => Some(b),
165		_ => None,
166	}
167}
168
169#[cfg(not(reifydb_single_threaded))]
170#[inline]
171fn first_uint8_value(frame: &Frame, name: &str) -> Option<u64> {
172	let col = frame.columns.iter().find(|c| c.name == name)?;
173	if col.data.is_empty() {
174		return None;
175	}
176	match col.data.get_value(0) {
177		Value::Uint8(n) => Some(n),
178		_ => None,
179	}
180}
reifydb_sub_server/subscription/create.rs

reifydb_sub_server/subscription/
create.rs
