reifydb_sub_server/subscription/
create.rs

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

reifydb_sub_server/subscription/
create.rs
