linkrs_server/http/handlers/
function.rs

1//! Custom function HTTP handler
2
3use axum::{
4    extract::{Extension, Json, Path, State},
5    response::Json as JsonResponse,
6};
7use parking_lot::RwLock;
8use serde::Deserialize;
9use serde_json;
10
11use crate::http::{error::HttpError, state::AppState};
12use crate::query::executor::expression::functions::{FunctionRegistry, UdfError};
13use crate::storage::{
14    StorageClient, StorageOperationContextOps, StorageSchemaContextOps, StorageSyncContextOps,
15};
16
17/// Transport-neutral function registry mutation failure.
18#[derive(Debug)]
19pub(crate) enum FunctionOpError {
20    NotFound(String),
21    Conflict(String),
22    Invalid(String),
23    Failed(String),
24}
25
26pub(crate) fn udf_error_to_op(error: UdfError) -> FunctionOpError {
27    let message = error.to_string();
28    match error {
29        UdfError::InvalidPath(_, _) | UdfError::InvalidName(_) => FunctionOpError::Invalid(message),
30        UdfError::AlreadyLoaded(_, _) | UdfError::BuiltinConflict(_) => {
31            FunctionOpError::Conflict(message)
32        }
33        UdfError::NotLoaded(_) => FunctionOpError::NotFound(message),
34        _ => FunctionOpError::Failed(message),
35    }
36}
37
38/// Register the UDF library at `implementation` (a local dynamic-library
39/// path) under `name`. Returns the registered (library-provided) name.
40///
41/// Only library-backed implementations can execute, so registration goes
42/// through the dynamic UDF loader: metadata-only registrations would
43/// produce functions that can never run.
44pub(crate) fn register_udf_from_source(
45    registry: &RwLock<FunctionRegistry>,
46    name: &str,
47    implementation: &str,
48) -> Result<String, FunctionOpError> {
49    if name.trim().is_empty() {
50        return Err(FunctionOpError::Invalid(
51            "function name must not be empty".to_string(),
52        ));
53    }
54    if registry.read().contains(name) {
55        return Err(FunctionOpError::Conflict(format!(
56            "function '{name}' already exists"
57        )));
58    }
59    if implementation.trim().is_empty() {
60        return Err(FunctionOpError::Invalid(
61            "implementation must be a local UDF library path; remote sources and embedded code are not supported".to_string(),
62        ));
63    }
64    let mut guard = registry.write();
65    match guard.install_dynamic_udf(implementation) {
66        Ok(loaded_name) => {
67            if loaded_name.to_uppercase() != name.to_uppercase() {
68                let _ = guard.unload_dynamic_udf(&loaded_name);
69                Err(FunctionOpError::Invalid(format!(
70                    "library exports '{loaded_name}', which does not match requested '{name}'"
71                )))
72            } else {
73                Ok(loaded_name)
74            }
75        }
76        Err(e) => Err(udf_error_to_op(e)),
77    }
78}
79
80/// Unload a dynamic UDF or remove a custom function by name.
81pub(crate) fn unregister_udf_by_name(
82    registry: &RwLock<FunctionRegistry>,
83    name: &str,
84) -> Result<(), FunctionOpError> {
85    if !registry.read().contains(name) {
86        return Err(FunctionOpError::NotFound(format!(
87            "function '{name}' does not exist"
88        )));
89    }
90    if registry.read().get_builtin(name).is_some() {
91        return Err(FunctionOpError::Invalid(format!(
92            "built-in function '{name}' cannot be unregistered"
93        )));
94    }
95    let mut guard = registry.write();
96    if guard.is_dynamic(name) {
97        guard.unload_dynamic_udf(name).map_err(udf_error_to_op)
98    } else if guard.unregister_custom(name) {
99        Ok(())
100    } else {
101        Err(FunctionOpError::NotFound(format!(
102            "function '{name}' does not exist"
103        )))
104    }
105}
106
107pub(crate) fn op_error_to_http(error: FunctionOpError) -> HttpError {
108    match error {
109        FunctionOpError::NotFound(message) => HttpError::not_found(message),
110        FunctionOpError::Conflict(message) | FunctionOpError::Invalid(message) => {
111            HttpError::bad_request(message)
112        }
113        FunctionOpError::Failed(message) => HttpError::internal(message),
114    }
115}
116
117#[utoipa::path(
118    post,
119    path = "/v1/functions",
120    tag = "Function",
121    request_body = RegisterFunctionRequest,
122    responses(
123        (status = 200, body = serde_json::Value, description = "Function registered"),
124        (status = 500, description = "Internal error")
125    )
126)]
127/// Register a custom function
128pub async fn register<
129    S: StorageClient
130        + StorageSchemaContextOps
131        + StorageSyncContextOps
132        + StorageOperationContextOps
133        + Clone
134        + Send
135        + Sync
136        + 'static,
137>(
138    State(state): State<AppState<S>>,
139    Extension(session_id): Extension<i64>,
140    Json(request): Json<RegisterFunctionRequest>,
141) -> Result<JsonResponse<serde_json::Value>, HttpError> {
142    super::authz::require_admin_session(&state, session_id)?;
143    let implementation = match &request.implementation {
144        Some(serde_json::Value::String(path)) => path.clone(),
145        Some(other) => {
146            return Err(HttpError::bad_request(format!(
147                "implementation must be a local UDF library path string, got {other}"
148            )));
149        }
150        None => String::new(),
151    };
152    let registry = state.server.get_function_registry();
153    let registered_name = register_udf_from_source(&registry, &request.name, &implementation)
154        .map_err(op_error_to_http)?;
155    let registry_guard = registry.read();
156    let return_type = registry_guard
157        .get_return_type(&registered_name)
158        .map(|t| format!("{t:?}"))
159        .unwrap_or_else(|| "unknown".to_string());
160
161    Ok(JsonResponse(serde_json::json!({
162        "function_id": registered_name,
163        "name": registered_name,
164        "function_type": "custom",
165        "return_type": return_type,
166        "status": "registered",
167        "message": "Function registered successfully",
168    })))
169}
170
171#[utoipa::path(
172    get,
173    path = "/v1/functions",
174    tag = "Function",
175    responses(
176        (status = 200, body = serde_json::Value, description = "Function list"),
177        (status = 500, description = "Internal error")
178    )
179)]
180/// List all functions
181pub async fn list<
182    S: StorageClient
183        + StorageSchemaContextOps
184        + StorageSyncContextOps
185        + StorageOperationContextOps
186        + Clone
187        + Send
188        + Sync
189        + 'static,
190>(
191    State(state): State<AppState<S>>,
192) -> Result<JsonResponse<serde_json::Value>, HttpError> {
193    let registry = state.server.get_function_registry();
194    let registry_guard = registry.read();
195
196    let function_names = registry_guard.function_names();
197    let functions: Vec<serde_json::Value> = function_names
198        .iter()
199        .map(|name| {
200            serde_json::json!({
201                "name": name,
202            })
203        })
204        .collect();
205
206    Ok(JsonResponse(serde_json::json!({
207        "functions": functions,
208        "total": functions.len(),
209    })))
210}
211
212#[utoipa::path(
213    get,
214    path = "/v1/functions/{name}",
215    tag = "Function",
216    params(("name" = String, Path, description = "Function name")),
217    responses(
218        (status = 200, body = serde_json::Value, description = "Function details"),
219        (status = 404, description = "Not found"),
220        (status = 500, description = "Internal error")
221    )
222)]
223/// Obtain function details
224pub async fn info<
225    S: StorageClient
226        + StorageSchemaContextOps
227        + StorageSyncContextOps
228        + StorageOperationContextOps
229        + Clone
230        + Send
231        + Sync
232        + 'static,
233>(
234    State(state): State<AppState<S>>,
235    Path(name): Path<String>,
236) -> Result<JsonResponse<serde_json::Value>, HttpError> {
237    let registry = state.server.get_function_registry();
238    let registry_guard = registry.read();
239
240    if !registry_guard.contains(&name) {
241        return Err(HttpError::not_found(format!(
242            "Function '{}' does not exist",
243            name
244        )));
245    }
246
247    let is_builtin = registry_guard.get_builtin(&name).is_some();
248    let is_custom = registry_guard.get_custom(&name).is_some();
249
250    let function_type = if is_builtin {
251        "builtin"
252    } else if is_custom {
253        "custom"
254    } else {
255        "unknown"
256    };
257    let return_type = registry_guard
258        .get_return_type(&name)
259        .map(|t| format!("{t:?}"))
260        .unwrap_or_else(|| "unknown".to_string());
261    // Dynamic UDFs record their library load time; builtins and in-process
262    // customs have no registration event, so the field stays null.
263    let registered_at = registry_guard
264        .dynamic_udf_mtime(&name)
265        .map(|mtime| chrono::DateTime::<chrono::Utc>::from(mtime).to_rfc3339());
266
267    Ok(JsonResponse(serde_json::json!({
268        "name": name,
269        "type": function_type,
270        "is_builtin": is_builtin,
271        "is_custom": is_custom,
272        "parameters": [],
273        "return_type": return_type,
274        "registered_at": registered_at,
275    })))
276}
277
278#[utoipa::path(
279    delete,
280    path = "/v1/functions/{name}",
281    tag = "Function",
282    params(("name" = String, Path, description = "Function name")),
283    responses(
284        (status = 200, body = serde_json::Value, description = "Function unregistered"),
285        (status = 404, description = "Not found"),
286        (status = 500, description = "Internal error")
287    )
288)]
289/// Logout function
290pub async fn unregister<
291    S: StorageClient
292        + StorageSchemaContextOps
293        + StorageSyncContextOps
294        + StorageOperationContextOps
295        + Clone
296        + Send
297        + Sync
298        + 'static,
299>(
300    State(state): State<AppState<S>>,
301    Extension(session_id): Extension<i64>,
302    Path(name): Path<String>,
303) -> Result<JsonResponse<serde_json::Value>, HttpError> {
304    super::authz::require_admin_session(&state, session_id)?;
305    let registry = state.server.get_function_registry();
306    unregister_udf_by_name(&registry, &name).map_err(op_error_to_http)?;
307
308    Ok(JsonResponse(serde_json::json!({
309        "message": "Function unregistered",
310        "name": name,
311    })))
312}
313
314/// Registration function request.
315///
316/// Only `name` and `implementation` (local dynamic-library path) decide
317/// registration. The remaining metadata fields are accepted for client
318/// compatibility but ignored; the loaded library is the source of truth.
319#[derive(Debug, Deserialize, utoipa::ToSchema)]
320pub struct RegisterFunctionRequest {
321    pub name: String,
322    #[serde(rename = "type", default)]
323    pub function_type: String,
324    #[serde(default)]
325    pub parameters: Vec<String>,
326    #[serde(rename = "return_type", default)]
327    pub return_type: String,
328    pub description: Option<String>,
329    pub implementation: Option<serde_json::Value>,
330}
linkrs_server/http/handlers/function.rs

linkrs_server/http/handlers/
function.rs
