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	metadata: RequestMetadata,
50) -> Result<CreateSubscriptionResult, CreateSubscriptionError> {
51	debug!("Subscription rql: {}", rql);
52
53	let ctx = RequestContext {
54		identity,
55		operation: Operation::Subscribe,
56		rql: rql.to_string(),
57		params: Params::None,
58		metadata,
59	};
60	let (frames, _metrics) = dispatch_subscribe(state, ctx).await?;
61	let frame = frames.first().ok_or(CreateSubscriptionError::ExtractionFailed)?;
62
63	if let Some(remote) = extract_remote_result(frame)? {
64		return Ok(remote);
65	}
66	extract_local_result(frame)
67}
68
69#[cfg(not(reifydb_single_threaded))]
70fn extract_remote_result(frame: &Frame) -> Result<Option<CreateSubscriptionResult>, CreateSubscriptionError> {
71	let Some(addr_col) = frame.columns.iter().find(|c| c.name == "remote_address") else {
72		return Ok(None);
73	};
74	let address = match first_utf8_value(addr_col) {
75		Some(s) => s,
76		None => return Err(CreateSubscriptionError::ExtractionFailed),
77	};
78	let body = frame
79		.columns
80		.iter()
81		.find(|c| c.name == "remote_body")
82		.and_then(first_utf8_value)
83		.ok_or(CreateSubscriptionError::ExtractionFailed)?;
84	let token = frame.columns.iter().find(|c| c.name == "remote_token").and_then(first_utf8_value);
85	let enabled = first_bool_value(frame, "remote_hydration_enabled").unwrap_or(true);
86	let max_rows = first_uint8_value(frame, "remote_hydration_max_rows");
87	let throttle = first_uint8_value(frame, "remote_throttle_ms")
88		.map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
89	let linger =
90		first_uint8_value(frame, "remote_linger_ms").map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
91	Ok(Some(CreateSubscriptionResult::Remote {
92		address,
93		body,
94		token,
95		hydration: HydrationConfig {
96			enabled,
97			max_rows,
98		},
99		throttle,
100		linger,
101	}))
102}
103
104#[cfg(not(reifydb_single_threaded))]
105fn extract_local_result(frame: &Frame) -> Result<CreateSubscriptionResult, CreateSubscriptionError> {
106	let id = frame
107		.columns
108		.iter()
109		.find(|c| c.name == "subscription_id")
110		.and_then(|col| {
111			if col.data.is_empty() {
112				None
113			} else {
114				Some(col.data.get_value(0))
115			}
116		})
117		.and_then(|value| match value {
118			Value::Uint8(id) => Some(SubscriptionId(id)),
119			other => {
120				error!("subscription_id column has wrong type: {:?}", other);
121				None
122			}
123		})
124		.ok_or(CreateSubscriptionError::ExtractionFailed)?;
125
126	let enabled = first_bool_value(frame, "hydration_enabled").unwrap_or(true);
127	let max_rows = first_uint8_value(frame, "hydration_max_rows");
128	let throttle =
129		first_uint8_value(frame, "throttle_ms").map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
130	let linger = first_uint8_value(frame, "linger_ms").map(|ms| Duration::from_milliseconds(ms as i64).unwrap());
131
132	Ok(CreateSubscriptionResult::Local {
133		id,
134		hydration: HydrationConfig {
135			enabled,
136			max_rows,
137		},
138		throttle,
139		linger,
140	})
141}
142
143#[cfg(not(reifydb_single_threaded))]
144#[inline]
145fn first_utf8_value(col: &FrameColumn) -> Option<String> {
146	if col.data.is_empty() {
147		return None;
148	}
149	match col.data.get_value(0) {
150		Value::Utf8(s) => Some(s),
151		_ => None,
152	}
153}
154
155#[cfg(not(reifydb_single_threaded))]
156#[inline]
157fn first_bool_value(frame: &Frame, name: &str) -> Option<bool> {
158	let col = frame.columns.iter().find(|c| c.name == name)?;
159	if col.data.is_empty() {
160		return None;
161	}
162	match col.data.get_value(0) {
163		Value::Boolean(b) => Some(b),
164		_ => None,
165	}
166}
167
168#[cfg(not(reifydb_single_threaded))]
169#[inline]
170fn first_uint8_value(frame: &Frame, name: &str) -> Option<u64> {
171	let col = frame.columns.iter().find(|c| c.name == name)?;
172	if col.data.is_empty() {
173		return None;
174	}
175	match col.data.get_value(0) {
176		Value::Uint8(n) => Some(n),
177		_ => None,
178	}
179}
reifydb_sub_server/subscription/create.rs

reifydb_sub_server/subscription/
create.rs
