linkrs_server/
http_server.rs

1//! HTTP and gRPC server bootstrap functions
2
3use std::sync::Arc;
4
5use log::info;
6
7use crate::config::Config;
8use crate::storage::UndoTarget;
9use crate::storage::{
10    StorageClient, StorageOperationContextOps, StorageSchemaContextOps, StorageSnapshotOps,
11    StorageSyncContextOps,
12};
13use crate::HttpServer;
14use linkrs_core::error::DBResult;
15
16use super::shutdown::async_shutdown_signal;
17
18/// Start an HTTP server using an asynchronous runtime.
19pub async fn start_http_server<
20    S: StorageClient
21        + StorageSchemaContextOps
22        + StorageSnapshotOps
23        + StorageSyncContextOps
24        + StorageOperationContextOps
25        + crate::storage::AutoCommitBatchOps
26        + crate::storage::AutoCommitGroupOps
27        + UndoTarget
28        + Clone
29        + Send
30        + Sync
31        + 'static,
32>(
33    server: Arc<HttpServer<S>>,
34    config: &Config,
35) -> DBResult<()> {
36    use axum::serve;
37    use tokio::net::TcpListener;
38
39    if config.server.http.https_enabled {
40        return Err(linkrs_core::DBError::validation(
41            "http.https_enabled is not served by this build; terminate TLS upstream and leave it disabled",
42        ));
43    }
44    if !config.server.http.enabled {
45        return Err(linkrs_core::DBError::validation(
46            "http.enabled is false; nothing to serve in http-only mode",
47        ));
48    }
49
50    let state = crate::http::AppState::new(server.clone());
51
52    // Create WebState for web management APIs
53    let storage_path = format!("{}/metadata.db", config.storage_path());
54    let web_router = match crate::web::WebState::new(&storage_path, state.clone()).await {
55        Ok(web_state) => Some(crate::web::create_router(web_state)),
56        Err(e) => {
57            log::warn!(
58                "Failed to initialize web management: {}, continuing without it",
59                e
60            );
61            None
62        }
63    };
64
65    let app = crate::http::router::create_router(state, web_router);
66
67    let addr = format!("{}:{}", config.http_bind_address(), config.http_port());
68    let listener = TcpListener::bind(&addr).await?;
69
70    info!("HTTP server listening on {}", addr);
71
72    serve(listener, app)
73        .with_graceful_shutdown(async_shutdown_signal())
74        .await?;
75
76    Ok(())
77}
78
79/// Start both HTTP and gRPC servers concurrently.
80#[cfg(feature = "grpc")]
81pub async fn start_http_and_grpc_servers<
82    S: StorageClient
83        + StorageSchemaContextOps
84        + StorageSnapshotOps
85        + StorageSyncContextOps
86        + StorageOperationContextOps
87        + crate::storage::AutoCommitBatchOps
88        + crate::storage::AutoCommitGroupOps
89        + Clone
90        + Send
91        + Sync
92        + 'static,
93>(
94    http_server: Arc<HttpServer<S>>,
95    config: &Config,
96) -> DBResult<()> {
97    use axum::serve;
98    use tokio::net::TcpListener;
99
100    if config.server.http.https_enabled {
101        return Err(linkrs_core::DBError::validation(
102            "http.https_enabled is not served by this build; terminate TLS upstream and leave it disabled",
103        ));
104    }
105
106    let http_state = crate::http::AppState::new(http_server.clone());
107
108    // Create WebState for web management APIs
109    let storage_path = format!("{}/metadata.db", config.storage_path());
110    let web_router = match crate::web::WebState::new(&storage_path, http_state.clone()).await {
111        Ok(web_state) => Some(crate::web::create_router(web_state)),
112        Err(e) => {
113            log::warn!(
114                "Failed to initialize web management: {}, continuing without it",
115                e
116            );
117            None
118        }
119    };
120
121    let http_enabled = config.server.http.enabled;
122    let grpc_enabled = config.server.grpc.enabled;
123    if !http_enabled && !grpc_enabled {
124        return Err(linkrs_core::DBError::validation(
125            "at least one of http.enabled and grpc.enabled must be true",
126        ));
127    }
128
129    let http_app = crate::http::router::create_router(http_state.clone(), web_router);
130
131    // Clone state for gRPC server
132    let grpc_state = http_state.clone();
133    let grpc_config = config.clone();
134    let http_bind = config.http_bind_address().to_string();
135    let http_port = config.http_port();
136
137    // Start HTTP server when enabled, otherwise wait for shutdown only.
138    let http_future = async move {
139        if !http_enabled {
140            async_shutdown_signal().await;
141            return Ok::<(), linkrs_core::error::DBError>(());
142        }
143        let http_addr = format!("{}:{}", http_bind, http_port);
144        let http_listener = TcpListener::bind(&http_addr).await?;
145        info!("HTTP server listening on {}", http_addr);
146        serve(http_listener, http_app)
147            .with_graceful_shutdown(async_shutdown_signal())
148            .await?;
149        Ok::<(), linkrs_core::error::DBError>(())
150    };
151
152    // Start gRPC server when enabled, otherwise wait for shutdown only.
153    let grpc_future = async move {
154        if !grpc_enabled {
155            async_shutdown_signal().await;
156            return Ok::<(), linkrs_core::error::DBError>(());
157        }
158        let grpc_addr = format!(
159            "{}:{}",
160            grpc_state.server.get_config().grpc_bind_address(),
161            grpc_state.server.get_config().grpc_port()
162        )
163        .parse::<std::net::SocketAddr>()
164        .map_err(|e| linkrs_core::error::DBError::internal(e.to_string()))?;
165        info!("gRPC server listening on {}", grpc_addr);
166        crate::grpc::run_server(grpc_state, grpc_config, grpc_addr)
167            .await
168            .map_err(|e| linkrs_core::error::DBError::internal(e.to_string()))?;
169        Ok::<(), linkrs_core::error::DBError>(())
170    };
171
172    // Run both servers concurrently
173    tokio::select! {
174        result = http_future => result?,
175        result = grpc_future => result?,
176    }
177
178    Ok(())
179}
linkrs_server/http_server.rs

linkrs_server/
http_server.rs
