reifydb_sub_server/subscription/
create.rs

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

reifydb_sub_server/subscription/
create.rs
