eidetica/sync/transports/
http.rs

1//! HTTP transport implementation for sync communication.
2//!
3//! This module provides HTTP-based sync communication using a single
4//! JSON endpoint (/api/v0) with axum for the server and reqwest for the client.
5
6use std::{net::SocketAddr, sync::Arc};
7
8use async_trait::async_trait;
9use axum::{
10    Router,
11    extract::{ConnectInfo, Json as ExtractJson, State},
12    response::Json,
13    routing::post,
14};
15use serde::{Deserialize, Serialize};
16use tokio::sync::oneshot;
17
18use super::{SyncTransport, TransportBuilder, TransportConfig, shared::*};
19use crate::{
20    Result,
21    crdt::Doc,
22    store::Registered,
23    sync::{
24        error::SyncError,
25        handler::SyncHandler,
26        peer_types::Address,
27        protocol::{RequestContext, SyncRequest, SyncResponse},
28    },
29};
30
31/// Persistable configuration for the HTTP transport.
32///
33/// Stores HTTP-specific configuration such as the bind address for listening.
34#[derive(Debug, Clone, Serialize, Deserialize, Default)]
35pub struct HttpTransportConfig {
36    /// Bind address for HTTP server (e.g., "127.0.0.1:8080").
37    /// If None, HTTP server won't be started by accept_connections().
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub bind_address: Option<String>,
40}
41
42impl Registered for HttpTransportConfig {
43    fn type_id() -> &'static str {
44        "http:v0"
45    }
46}
47
48impl TransportConfig for HttpTransportConfig {}
49
50/// Builder for configuring HTTP transport.
51///
52/// # Example
53///
54/// ```ignore
55/// use eidetica::sync::transports::http::HttpTransport;
56///
57/// // Create transport with specific bind address
58/// let builder = HttpTransport::builder()
59///     .bind("127.0.0.1:8080");
60///
61/// // Register with sync system
62/// sync.register_transport("http-local", builder).await?;
63/// ```
64#[derive(Debug, Clone, Default)]
65pub struct HttpTransportBuilder {
66    bind_address: Option<String>,
67}
68
69impl HttpTransportBuilder {
70    /// Create a new builder with default settings.
71    pub fn new() -> Self {
72        Self::default()
73    }
74
75    /// Set the bind address for the HTTP server.
76    ///
77    /// # Arguments
78    /// * `addr` - The address to bind to (e.g., "127.0.0.1:8080" or "0.0.0.0:80")
79    ///
80    /// # Example
81    ///
82    /// ```ignore
83    /// let builder = HttpTransport::builder()
84    ///     .bind("0.0.0.0:8080");
85    /// ```
86    pub fn bind(mut self, addr: impl Into<String>) -> Self {
87        self.bind_address = Some(addr.into());
88        self
89    }
90
91    /// Build the transport synchronously (for backwards compatibility).
92    ///
93    /// Note: The bind address is stored but the server is not started.
94    /// Call `start_server()` on the transport to actually bind.
95    pub fn build_sync(self) -> Result<HttpTransport> {
96        Ok(HttpTransport {
97            server_state: ServerState::new(),
98            bind_address: self.bind_address,
99            client: reqwest::Client::new(),
100        })
101    }
102}
103
104#[async_trait]
105impl TransportBuilder for HttpTransportBuilder {
106    type Transport = HttpTransport;
107
108    /// Build the HTTP transport.
109    ///
110    /// HTTP transport doesn't require persisted state for identity.
111    async fn build(self, _persisted: Doc) -> Result<(Self::Transport, Option<Doc>)> {
112        let transport = HttpTransport {
113            server_state: ServerState::new(),
114            bind_address: self.bind_address,
115            client: reqwest::Client::new(),
116        };
117        Ok((transport, None))
118    }
119}
120
121/// HTTP transport implementation using axum and reqwest.
122pub struct HttpTransport {
123    /// Shared server state management.
124    server_state: ServerState,
125    /// Configured bind address (used when start_server is called with empty addr)
126    bind_address: Option<String>,
127    /// Outbound HTTP client, shared across requests so its connection pool
128    /// survives between sends rather than being rebuilt per request.
129    client: reqwest::Client,
130}
131
132impl HttpTransport {
133    /// Transport type identifier for HTTP
134    pub const TRANSPORT_TYPE: &'static str = "http";
135
136    /// Create a new HTTP transport instance.
137    pub fn new() -> Result<Self> {
138        Ok(Self {
139            server_state: ServerState::new(),
140            bind_address: None,
141            client: reqwest::Client::new(),
142        })
143    }
144
145    /// Create a builder for configuring the transport.
146    pub fn builder() -> HttpTransportBuilder {
147        HttpTransportBuilder::new()
148    }
149
150    /// Create the axum router with single JSON endpoint and handler state.
151    fn create_router(handler: Arc<dyn SyncHandler>) -> Router {
152        Router::new()
153            .route("/api/v0", post(handle_sync_request))
154            .with_state(handler)
155    }
156}
157
158#[async_trait]
159impl SyncTransport for HttpTransport {
160    fn transport_type(&self) -> &'static str {
161        Self::TRANSPORT_TYPE
162    }
163
164    fn can_handle_address(&self, address: &Address) -> bool {
165        address.transport_type == Self::TRANSPORT_TYPE
166    }
167
168    async fn start_server(&mut self, handler: Arc<dyn SyncHandler>) -> Result<()> {
169        // No bind address configured = client-only transport, nothing to start
170        let Some(effective_addr) = self.bind_address.as_deref() else {
171            return Ok(());
172        };
173
174        // Check if server is already running
175        if self.server_state.is_running() {
176            return Err(SyncError::ServerAlreadyRunning {
177                address: effective_addr.to_string(),
178            }
179            .into());
180        }
181
182        let socket_addr: SocketAddr =
183            effective_addr.parse().map_err(|e| SyncError::ServerBind {
184                address: effective_addr.to_string(),
185                reason: format!("Invalid address: {e}"),
186            })?;
187
188        let router = Self::create_router(handler);
189
190        // Create server coordination channels
191        let (ready_tx, ready_rx) = oneshot::channel();
192        let (shutdown_tx, shutdown_rx) = oneshot::channel();
193
194        // Create a channel to get the actual bound address back
195        let (addr_tx, addr_rx) = oneshot::channel::<SocketAddr>();
196
197        // Spawn server task
198        tokio::spawn(async move {
199            let listener = tokio::net::TcpListener::bind(socket_addr)
200                .await
201                .expect("Failed to bind address");
202
203            // Get the actual bound address (important for port 0)
204            let actual_addr = listener.local_addr().expect("Failed to get local address");
205
206            // Send the actual address back
207            let _ = addr_tx.send(actual_addr);
208
209            // Signal that server is ready
210            let _ = ready_tx.send(());
211
212            // Run server with graceful shutdown
213            // Convert router to service with ConnectInfo support
214            axum::serve(
215                listener,
216                router.into_make_service_with_connect_info::<SocketAddr>(),
217            )
218            .with_graceful_shutdown(async move {
219                let _ = shutdown_rx.await;
220            })
221            .await
222            .expect("Server failed");
223        });
224
225        // Get the actual bound address
226        let actual_addr = addr_rx.await.map_err(|_| SyncError::ServerBind {
227            address: effective_addr.to_string(),
228            reason: "Failed to get actual server address".to_string(),
229        })?;
230
231        // Wait for server to be ready
232        wait_for_ready(ready_rx, effective_addr).await?;
233
234        // Start server state with address and shutdown sender
235        self.server_state
236            .server_started(actual_addr.to_string(), shutdown_tx);
237
238        Ok(())
239    }
240
241    async fn stop_server(&mut self) -> Result<()> {
242        if !self.server_state.is_running() {
243            return Err(SyncError::ServerNotRunning.into());
244        }
245
246        // Stop server using combined method
247        self.server_state.stop_server();
248
249        Ok(())
250    }
251
252    async fn send_request(&self, address: &Address, request: &SyncRequest) -> Result<SyncResponse> {
253        if !self.can_handle_address(address) {
254            return Err(SyncError::UnsupportedTransport {
255                transport_type: address.transport_type.clone(),
256            }
257            .into());
258        }
259
260        let url = format!("http://{}/api/v0", address.address);
261
262        let response = self
263            .client
264            .post(&url)
265            .json(&request) // Send SyncRequest as JSON body
266            .send()
267            .await
268            .map_err(|e| SyncError::ConnectionFailed {
269                address: address.address.clone(),
270                reason: e.to_string(),
271            })?;
272
273        if !response.status().is_success() {
274            return Err(SyncError::Network(format!(
275                "Server returned error: {}",
276                response.status()
277            ))
278            .into());
279        }
280
281        let sync_response: SyncResponse = response
282            .json()
283            .await
284            .map_err(|e| SyncError::Network(format!("Failed to parse response: {e}")))?;
285
286        Ok(sync_response)
287    }
288
289    fn is_server_running(&self) -> bool {
290        self.server_state.is_running()
291    }
292
293    fn get_server_address(&self) -> Result<String> {
294        self.server_state.get_address().map_err(|e| e.into())
295    }
296}
297
298/// Handler for the /api/v0 endpoint - accepts JSON SyncRequest and returns JSON SyncResponse.
299async fn handle_sync_request(
300    State(handler): State<Arc<dyn SyncHandler>>,
301    ConnectInfo(addr): ConnectInfo<SocketAddr>,
302    ExtractJson(request): ExtractJson<SyncRequest>,
303) -> Json<SyncResponse> {
304    // Extract peer_pubkey from SyncTreeRequest if present
305    let peer_pubkey = match &request {
306        SyncRequest::SyncTree(sync_tree_request) => sync_tree_request.peer_pubkey.clone(),
307        _ => None,
308    };
309
310    // Create request context with remote address and peer pubkey
311    let context = RequestContext {
312        remote_address: Some(Address {
313            transport_type: HttpTransport::TRANSPORT_TYPE.to_string(),
314            address: addr.to_string(),
315        }),
316        peer_pubkey,
317    };
318
319    // Call handler directly (Transaction is now Send since it uses Arc<Mutex>)
320    let response = handler.handle_request(&request, &context).await;
321
322    Json(response)
323}
eidetica/sync/transports/http.rs

eidetica/sync/transports/
http.rs
