hashtree_cli/server/
mod.rs

1mod auth;
2mod blob_read;
3pub mod blossom;
4mod handlers;
5mod ingest_filter;
6mod mime;
7mod nostr_query;
8mod peer_status;
9mod request_paths;
10mod status_metrics;
11#[cfg(feature = "p2p")]
12pub mod stun;
13mod ui;
14pub mod ws_relay;
15
16use crate::nostr_relay::NostrRelay;
17use crate::socialgraph;
18use crate::storage::HashtreeStore;
19use crate::webrtc::WebRTCState;
20use anyhow::Result;
21use axum::{
22    body::Body,
23    extract::DefaultBodyLimit,
24    http::Request,
25    middleware,
26    response::Response,
27    routing::{get, post, put},
28    Router,
29};
30use futures::{future::poll_fn, pin_mut, FutureExt};
31use hashtree_core::Cid;
32use hyper::body::Incoming;
33use hyper_util::{
34    rt::{TokioExecutor, TokioIo, TokioTimer},
35    server::conn::auto::Builder as HyperBuilder,
36    service::TowerToHyperService,
37};
38use socket2::{SockRef, TcpKeepalive};
39use std::collections::{HashMap, HashSet};
40use std::convert::Infallible;
41use std::future;
42use std::io;
43use std::net::SocketAddr;
44use std::sync::{Arc, OnceLock, RwLock};
45use std::time::{Duration, Instant};
46use tokio::sync::watch;
47use tower::{Service, ServiceExt as _};
48use tower_http::cors::CorsLayer;
49use tracing::{debug, error, trace};
50
51pub use auth::{
52    new_lookup_cache, new_upstream_http_client, AppState, AuthCredentials, CachedTreeRootEntry,
53};
54
55static VIRTUAL_TREE_HOSTS: OnceLock<RwLock<HashMap<String, String>>> = OnceLock::new();
56const DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES: usize = 512 * 1024 * 1024;
57const DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES: usize = 512 * 1024 * 1024;
58
59#[cfg(not(test))]
60const HTTP1_HEADER_READ_TIMEOUT: Duration = Duration::from_secs(30);
61#[cfg(test)]
62const HTTP1_HEADER_READ_TIMEOUT: Duration = Duration::from_millis(200);
63const HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
64const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);
65const TCP_KEEPALIVE_TIME: Duration = Duration::from_secs(60);
66const TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15);
67
68fn virtual_tree_hosts() -> &'static RwLock<HashMap<String, String>> {
69    VIRTUAL_TREE_HOSTS.get_or_init(|| RwLock::new(HashMap::new()))
70}
71
72fn normalize_virtual_tree_host(host: &str) -> Option<String> {
73    let trimmed = host.trim().trim_end_matches('.').to_ascii_lowercase();
74    if trimmed.is_empty() {
75        return None;
76    }
77
78    if let Some(stripped) = trimmed
79        .strip_prefix('[')
80        .and_then(|value| value.split_once(']'))
81    {
82        let host_only = stripped.0.trim();
83        if host_only.is_empty() {
84            return None;
85        }
86        return Some(host_only.to_string());
87    }
88
89    if let Some((host_only, port)) = trimmed.rsplit_once(':') {
90        if !host_only.is_empty() && !port.is_empty() && port.chars().all(|ch| ch.is_ascii_digit()) {
91            return Some(host_only.to_string());
92        }
93    }
94
95    Some(trimmed)
96}
97
98pub fn register_virtual_tree_host(host: &str, internal_root: &str) {
99    let Some(normalized_host) = normalize_virtual_tree_host(host) else {
100        return;
101    };
102
103    let normalized_root = internal_root.trim().trim_end_matches('/');
104    if normalized_root.is_empty() {
105        return;
106    }
107
108    if let Ok(mut hosts) = virtual_tree_hosts().write() {
109        hosts.insert(normalized_host, normalized_root.to_string());
110    }
111}
112
113pub fn resolve_virtual_tree_host(host: &str) -> Option<String> {
114    let normalized_host = normalize_virtual_tree_host(host)?;
115    virtual_tree_hosts()
116        .read()
117        .ok()
118        .and_then(|hosts| hosts.get(&normalized_host).cloned())
119}
120
121#[cfg(test)]
122pub fn clear_virtual_tree_hosts_for_test() {
123    if let Ok(mut hosts) = virtual_tree_hosts().write() {
124        hosts.clear();
125    }
126}
127
128pub struct HashtreeServer {
129    state: AppState,
130    addr: String,
131    extra_routes: Option<Router<AppState>>,
132    cors: Option<CorsLayer>,
133}
134
135impl HashtreeServer {
136    pub fn new(store: Arc<HashtreeStore>, addr: String) -> Self {
137        Self {
138            state: AppState {
139                store,
140                auth: None,
141                daemon_started_at: current_unix_secs(),
142                peer_mode: crate::config::ServerMode::Normal,
143                hash_get_enabled: true,
144                http_webrtc_fetch: true,
145                webrtc_peers: None,
146                fips_transport: None,
147                fetch_from_fips_peers: true,
148                ws_relay: Arc::new(auth::WsRelayState::new()),
149                max_upload_bytes: 5 * 1024 * 1024, // 5 MB default
150                public_writes: true,               // Allow anyone with valid Nostr auth by default
151                public_plaintext_reads: true,
152                require_random_untrusted_ingest: true,
153                optimistic_blossom_uploads: false,
154                optimistic_upload_queue_bytes: DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES,
155                optimistic_upload_queue: Arc::new(tokio::sync::Semaphore::new(
156                    DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES,
157                )),
158                allowed_pubkeys: HashSet::new(), // No pubkeys allowed by default (use public_writes)
159                upstream_blossom: Vec::new(),
160                upstream_http_client: new_upstream_http_client(),
161                upstream_blossom_miss_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
162                upstream_blossom_fetch_metrics: Arc::new(
163                    auth::UpstreamBlossomFetchMetrics::default(),
164                ),
165                blossom_upload_replicas: Vec::new(),
166                blossom_upload_replica_queue_bytes: DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES,
167                blossom_upload_replica_queue: Arc::new(tokio::sync::Semaphore::new(
168                    DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES,
169                )),
170                blossom_upload_replica_keys: None,
171                blossom_upload_replica_scheduler: Arc::new(
172                    blossom::BlossomUploadReplicaScheduler::new(),
173                ),
174                social_graph: None,
175                social_graph_store: None,
176                social_graph_root: None,
177                socialgraph_snapshot_public: false,
178                nostr_relay: None,
179                nostr_relay_urls: Vec::new(),
180                tree_root_cache: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
181                inflight_blob_fetches: Arc::new(tokio::sync::Mutex::new(
182                    std::collections::HashMap::new(),
183                )),
184                inflight_blob_reads: Arc::new(tokio::sync::Mutex::new(
185                    std::collections::HashMap::new(),
186                )),
187                blob_cache: Arc::new(crate::blob_cache::BlobCache::from_env()),
188                directory_listing_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
189                resolved_path_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
190                thumbnail_path_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
191                cid_size_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
192            },
193            addr,
194            extra_routes: None,
195            cors: None,
196        }
197    }
198
199    /// Set maximum upload size for Blossom uploads
200    pub fn with_max_upload_bytes(mut self, bytes: usize) -> Self {
201        self.state.max_upload_bytes = bytes;
202        self
203    }
204
205    /// Set whether to allow public writes (anyone with valid Nostr auth)
206    /// When false, only social graph members can write
207    pub fn with_public_writes(mut self, public: bool) -> Self {
208        self.state.public_writes = public;
209        self
210    }
211
212    /// Set whether mutable npub routes serve plaintext for unapproved pubkeys
213    pub fn with_public_plaintext_reads(mut self, public: bool) -> Self {
214        self.state.public_plaintext_reads = public;
215        self
216    }
217
218    pub fn with_require_random_untrusted_ingest(mut self, require: bool) -> Self {
219        self.state.require_random_untrusted_ingest = require;
220        self
221    }
222
223    pub fn with_optimistic_blossom_uploads(mut self, enabled: bool) -> Self {
224        self.state.optimistic_blossom_uploads = enabled;
225        self
226    }
227
228    pub fn with_server_mode(mut self, mode: crate::config::ServerMode) -> Self {
229        self.state.peer_mode = mode;
230        self
231    }
232
233    pub fn with_hash_get_enabled(mut self, enabled: bool) -> Self {
234        self.state.hash_get_enabled = enabled;
235        self
236    }
237
238    pub fn with_http_webrtc_fetch(mut self, enabled: bool) -> Self {
239        self.state.http_webrtc_fetch = enabled;
240        self
241    }
242
243    pub fn with_fetch_from_fips_peers(mut self, enabled: bool) -> Self {
244        self.state.fetch_from_fips_peers = enabled;
245        self
246    }
247
248    pub fn with_fips_transport(
249        mut self,
250        transport: Arc<crate::fips_transport::DaemonFipsTransport>,
251    ) -> Self {
252        self.state.fips_transport = Some(transport);
253        self
254    }
255
256    /// Set WebRTC state for P2P peer queries
257    pub fn with_webrtc_peers(mut self, webrtc_state: Arc<WebRTCState>) -> Self {
258        self.state.webrtc_peers = Some(webrtc_state);
259        self
260    }
261
262    pub fn with_auth(mut self, username: String, password: String) -> Self {
263        self.state.auth = Some(AuthCredentials { username, password });
264        self
265    }
266
267    /// Set allowed pubkeys for blossom write access (hex format)
268    pub fn with_allowed_pubkeys(mut self, pubkeys: HashSet<String>) -> Self {
269        self.state.allowed_pubkeys = pubkeys;
270        self
271    }
272
273    /// Set upstream Blossom servers for cascade fetching
274    pub fn with_upstream_blossom(mut self, servers: Vec<String>) -> Self {
275        self.state.upstream_blossom = servers;
276        self
277    }
278
279    /// Set write-behind Blossom servers for blobs accepted by this server.
280    pub fn with_blossom_upload_replicas(
281        mut self,
282        servers: Vec<String>,
283        queue_bytes: usize,
284        keys: nostr::Keys,
285    ) -> Self {
286        let queue_bytes = queue_bytes.max(1);
287        let mut servers: Vec<String> = servers
288            .into_iter()
289            .map(|server| server.trim().trim_end_matches('/').to_string())
290            .filter(|server| !server.is_empty())
291            .collect();
292        servers.sort();
293        servers.dedup();
294        let replica_keys = (!servers.is_empty()).then(|| Arc::new(keys));
295        self.state.blossom_upload_replicas = servers;
296        self.state.blossom_upload_replica_queue_bytes = queue_bytes;
297        self.state.blossom_upload_replica_queue =
298            Arc::new(tokio::sync::Semaphore::new(queue_bytes));
299        self.state.blossom_upload_replica_keys = replica_keys;
300        self
301    }
302
303    /// Set social graph access control
304    pub fn with_social_graph(mut self, sg: Arc<socialgraph::SocialGraphAccessControl>) -> Self {
305        self.state.social_graph = Some(sg);
306        self
307    }
308
309    /// Configure social graph snapshot export (store handle + root)
310    pub fn with_socialgraph_snapshot(
311        mut self,
312        store: Arc<dyn socialgraph::SocialGraphBackend>,
313        root: [u8; 32],
314        public: bool,
315    ) -> Self {
316        self.state.social_graph_store = Some(store);
317        self.state.social_graph_root = Some(root);
318        self.state.socialgraph_snapshot_public = public;
319        self
320    }
321
322    /// Set Nostr relay state (shared for /ws and WebRTC)
323    pub fn with_nostr_relay(mut self, relay: Arc<NostrRelay>) -> Self {
324        self.state.nostr_relay = Some(relay);
325        self
326    }
327
328    /// Set active upstream Nostr relays for HTTP resolver operations.
329    pub fn with_nostr_relay_urls(mut self, relays: Vec<String>) -> Self {
330        self.state.nostr_relay_urls = relays;
331        self
332    }
333
334    /// Seed mutable root cache entries before the server starts.
335    pub fn with_cached_tree_roots(self, roots: Vec<(String, Cid)>) -> Self {
336        if let Ok(mut cache) = self.state.tree_root_cache.lock() {
337            let now = Instant::now();
338            for (key, cid) in roots {
339                cache.insert(
340                    key,
341                    CachedTreeRootEntry {
342                        cid,
343                        source: "embedded-bootstrap",
344                        root_event: None,
345                        cached_at: now,
346                    },
347                );
348            }
349        }
350        self
351    }
352
353    /// Merge extra routes into the daemon router (e.g. Tauri embeds /nip07).
354    pub fn with_extra_routes(mut self, routes: Router<AppState>) -> Self {
355        self.extra_routes = Some(routes);
356        self
357    }
358
359    /// Apply a CORS layer to all routes (used by embedded clients like Tauri).
360    pub fn with_cors(mut self, cors: CorsLayer) -> Self {
361        self.cors = Some(cors);
362        self
363    }
364
365    pub async fn run(self) -> Result<()> {
366        let listener = tokio::net::TcpListener::bind(&self.addr).await?;
367        let _ = self.run_with_listener(listener).await?;
368        Ok(())
369    }
370
371    pub async fn run_with_listener(self, listener: tokio::net::TcpListener) -> Result<u16> {
372        self.run_with_listener_until(listener, future::pending::<()>())
373            .await
374    }
375
376    pub async fn run_with_listener_until<F>(
377        self,
378        listener: tokio::net::TcpListener,
379        shutdown: F,
380    ) -> Result<u16>
381    where
382        F: std::future::Future<Output = ()> + Send + 'static,
383    {
384        let local_addr = listener.local_addr()?;
385
386        // Public endpoints (no auth required)
387        // Note: /:id serves raw SHA256 blobs only. Logical tree/file assembly
388        // stays on explicitly configured mutable routes such as approved npub.
389        let state = self.state.clone();
390        let public_routes = Router::new()
391            .route("/", get(handlers::serve_root_or_virtual_host))
392            .route("/ws", get(ws_relay::ws_data))
393            .route("/ws/", get(ws_relay::ws_data))
394            .route(
395                "/__iris/store/:hash",
396                get(handlers::iris_store_get)
397                    .head(handlers::iris_store_head)
398                    .put(handlers::iris_store_put)
399                    .delete(handlers::iris_store_delete),
400            )
401            .route(
402                "/htree/test",
403                get(handlers::htree_test).head(handlers::htree_test),
404            )
405            // /htree/nhash1...[/path] - content-addressed (immutable)
406            .route("/htree/nhash1:nhash", get(handlers::htree_nhash))
407            .route("/htree/nhash1:nhash/", get(handlers::htree_nhash))
408            .route("/htree/nhash1:nhash/*path", get(handlers::htree_nhash_path))
409            // /htree/npub1.../tree[/path] - mutable (resolver-backed)
410            .route("/htree/npub1:npub/:treename", get(handlers::htree_npub))
411            .route("/htree/npub1:npub/:treename/", get(handlers::htree_npub))
412            .route(
413                "/htree/npub1:npub/:treename/*path",
414                get(handlers::htree_npub_path),
415            )
416            // Nostr resolver endpoints - resolve npub/treename to content
417            .route("/n/:pubkey/:treename", get(handlers::resolve_and_serve))
418            // Direct npub route (clients should parse nhash and request by hex hash)
419            .route("/npub1:rest", get(handlers::serve_npub))
420            .route("/npub1:rest/*path", get(handlers::serve_npub))
421            // Blossom endpoints (BUD-01, BUD-02)
422            .route(
423                "/:id",
424                get(handlers::serve_content_or_blob)
425                    .head(blossom::head_blob)
426                    .delete(blossom::delete_blob)
427                    .options(blossom::cors_preflight),
428            )
429            .route(
430                "/upload",
431                put(blossom::upload_blob)
432                    .head(blossom::head_upload)
433                    .options(blossom::cors_preflight),
434            )
435            .route(
436                "/upload/batch",
437                post(blossom::upload_blob_batch).options(blossom::cors_preflight),
438            )
439            .route(
440                "/upload/batch-binary",
441                post(blossom::upload_blob_batch_binary).options(blossom::cors_preflight),
442            )
443            .route(
444                "/upload/check",
445                post(blossom::upload_check).options(blossom::cors_preflight),
446            )
447            .route(
448                "/blob/batch",
449                post(handlers::download_blob_batch).options(blossom::cors_preflight),
450            )
451            .route(
452                "/list/:pubkey",
453                get(blossom::list_blobs).options(blossom::cors_preflight),
454            )
455            // Hashtree API endpoints
456            .route("/health", get(handlers::health_check))
457            .route("/api/pins", get(handlers::list_pins))
458            .route("/api/stats", get(handlers::storage_stats))
459            .route("/api/peers", get(handlers::webrtc_peers))
460            .route("/api/status", get(handlers::daemon_status))
461            .route("/api/p2p/signal", post(handlers::p2p_signal))
462            .route("/api/socialgraph", get(handlers::socialgraph_stats))
463            .route(
464                "/api/socialgraph/snapshot",
465                get(handlers::socialgraph_snapshot),
466            )
467            .route(
468                "/api/socialgraph/distance/:pubkey",
469                get(handlers::follow_distance),
470            )
471            // Resolver API endpoints
472            .route(
473                "/api/resolve/:pubkey/:treename",
474                get(handlers::resolve_to_hash),
475            )
476            .route(
477                "/api/nostr/resolve/:pubkey/:treename",
478                get(handlers::resolve_to_hash),
479            )
480            .route("/api/nostr/profile/:pubkey", get(handlers::nostr_profile))
481            .route("/api/cache-tree-root", post(handlers::cache_tree_root))
482            .route(
483                "/api/clear-tree-root-cache",
484                post(handlers::clear_tree_root_cache),
485            )
486            .route("/api/trees/:pubkey", get(handlers::list_trees))
487            .fallback(get(handlers::serve_virtual_host_fallback))
488            .with_state(state.clone());
489
490        // Protected endpoints (require auth if enabled)
491        let protected_routes = Router::new()
492            .route("/upload", post(handlers::upload_file))
493            .route("/api/pin/:cid", post(handlers::pin_cid))
494            .route("/api/unpin/:cid", post(handlers::unpin_cid))
495            .route("/api/gc", post(handlers::garbage_collect))
496            .layer(middleware::from_fn_with_state(
497                state.clone(),
498                auth::auth_middleware,
499            ))
500            .with_state(state.clone());
501
502        let mut app = public_routes
503            .merge(protected_routes)
504            .layer(DefaultBodyLimit::max(10 * 1024 * 1024 * 1024)) // 10GB limit
505            .layer(middleware::from_fn(status_metrics::record_http_status));
506
507        if let Some(extra) = self.extra_routes {
508            app = app.merge(extra.with_state(state));
509        }
510
511        if let Some(cors) = self.cors {
512            app = app.layer(cors);
513        }
514
515        let make_service = app.into_make_service_with_connect_info::<std::net::SocketAddr>();
516        serve_with_connection_limits(listener, make_service, shutdown).await?;
517
518        Ok(local_addr.port())
519    }
520
521    pub fn addr(&self) -> &str {
522        &self.addr
523    }
524}
525
526async fn serve_with_connection_limits<M, S, F>(
527    listener: tokio::net::TcpListener,
528    mut make_service: M,
529    shutdown: F,
530) -> io::Result<()>
531where
532    M: Service<SocketAddr, Error = Infallible, Response = S> + Send + 'static,
533    M::Future: Send,
534    S: Service<Request<Body>, Response = Response, Error = Infallible> + Clone + Send + 'static,
535    S::Future: Send,
536    F: std::future::Future<Output = ()> + Send + 'static,
537{
538    let (signal_tx, signal_rx) = watch::channel(());
539    let signal_tx = Arc::new(signal_tx);
540    tokio::spawn(async move {
541        shutdown.await;
542        trace!("received graceful shutdown signal; stopping daemon listener");
543        drop(signal_rx);
544    });
545
546    let (close_tx, close_rx) = watch::channel(());
547
548    loop {
549        let (tcp_stream, remote_addr) = tokio::select! {
550            accepted = accept_tcp(&listener) => {
551                match accepted? {
552                    Some(connection) => connection,
553                    None => continue,
554                }
555            }
556            _ = signal_tx.closed() => {
557                trace!("shutdown signal received; no longer accepting daemon connections");
558                break;
559            }
560        };
561
562        configure_tcp_stream(&tcp_stream);
563        let tcp_stream = TokioIo::new(tcp_stream);
564
565        poll_fn(|cx| make_service.poll_ready(cx))
566            .await
567            .unwrap_or_else(|err| match err {});
568
569        let tower_service = make_service
570            .call(remote_addr)
571            .await
572            .unwrap_or_else(|err| match err {})
573            .map_request(|req: Request<Incoming>| req.map(Body::new));
574        let hyper_service = TowerToHyperService::new(tower_service);
575
576        let signal_tx = Arc::clone(&signal_tx);
577        let close_rx = close_rx.clone();
578
579        tokio::spawn(async move {
580            let mut builder = HyperBuilder::new(TokioExecutor::new());
581            builder
582                .http1()
583                .timer(TokioTimer::new())
584                .header_read_timeout(HTTP1_HEADER_READ_TIMEOUT);
585            builder
586                .http2()
587                .timer(TokioTimer::new())
588                .keep_alive_interval(Some(HTTP2_KEEPALIVE_INTERVAL))
589                .keep_alive_timeout(HTTP2_KEEPALIVE_TIMEOUT);
590
591            let conn = builder.serve_connection_with_upgrades(tcp_stream, hyper_service);
592            pin_mut!(conn);
593
594            let signal_closed = signal_tx.closed().fuse();
595            pin_mut!(signal_closed);
596
597            loop {
598                tokio::select! {
599                    result = conn.as_mut() => {
600                        if let Err(err) = result {
601                            trace!("daemon connection closed with error: {err:#}");
602                        }
603                        break;
604                    }
605                    _ = &mut signal_closed => {
606                        trace!("shutdown signal received by connection task");
607                        conn.as_mut().graceful_shutdown();
608                    }
609                }
610            }
611
612            drop(close_rx);
613        });
614    }
615
616    drop(close_rx);
617    drop(listener);
618    close_tx.closed().await;
619
620    Ok(())
621}
622
623fn configure_tcp_stream(tcp_stream: &tokio::net::TcpStream) {
624    if let Err(err) = tcp_stream.set_nodelay(true) {
625        debug!("failed to set TCP_NODELAY on daemon connection: {err:#}");
626    }
627
628    let socket = SockRef::from(tcp_stream);
629    if let Err(err) = socket.set_tcp_keepalive(
630        &TcpKeepalive::new()
631            .with_time(TCP_KEEPALIVE_TIME)
632            .with_interval(TCP_KEEPALIVE_INTERVAL),
633    ) {
634        debug!("failed to set TCP keepalive on daemon connection: {err:#}");
635    }
636}
637
638async fn accept_tcp(
639    listener: &tokio::net::TcpListener,
640) -> io::Result<Option<(tokio::net::TcpStream, SocketAddr)>> {
641    match listener.accept().await {
642        Ok(connection) => Ok(Some(connection)),
643        Err(err) => {
644            if is_connection_error(&err) {
645                return Ok(None);
646            }
647            if is_resource_exhaustion_error(&err) {
648                error!(
649                    "daemon accept failed due to file descriptor exhaustion; exiting for supervisor restart: {err}"
650                );
651                return Err(err);
652            }
653            error!("daemon accept error: {err}");
654            tokio::time::sleep(Duration::from_secs(1)).await;
655            Ok(None)
656        }
657    }
658}
659
660fn is_resource_exhaustion_error(err: &io::Error) -> bool {
661    matches!(
662        err.raw_os_error(),
663        Some(code) if code == libc::EMFILE || code == libc::ENFILE
664    )
665}
666
667fn is_connection_error(err: &io::Error) -> bool {
668    matches!(
669        err.kind(),
670        io::ErrorKind::ConnectionRefused
671            | io::ErrorKind::ConnectionAborted
672            | io::ErrorKind::ConnectionReset
673    )
674}
675
676fn current_unix_secs() -> u64 {
677    std::time::SystemTime::now()
678        .duration_since(std::time::UNIX_EPOCH)
679        .unwrap_or(std::time::Duration::ZERO)
680        .as_secs()
681}
682
683#[cfg(test)]
684mod tests {
685    use super::*;
686    use crate::nostr_relay::{NostrRelay, NostrRelayConfig};
687    use crate::storage::HashtreeStore;
688    use hashtree_core::{from_hex, nhash_encode, DirEntry, HashTree, HashTreeConfig, LinkType};
689    use nostr::{EventBuilder, Keys, Kind, Timestamp};
690    use serde_json::json;
691    use tempfile::TempDir;
692
693    #[test]
694    fn resource_exhaustion_errors_are_fatal_accept_errors() {
695        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
696            libc::EMFILE
697        )));
698        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
699            libc::ENFILE
700        )));
701        assert!(!is_resource_exhaustion_error(
702            &io::Error::from_raw_os_error(libc::ECONNRESET)
703        ));
704    }
705
706    #[test]
707    fn server_builder_seeds_initial_tree_roots() -> Result<()> {
708        let temp_dir = TempDir::new()?;
709        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
710        let hash = from_hex("1111111111111111111111111111111111111111111111111111111111111111")?;
711        let key = from_hex("2222222222222222222222222222222222222222222222222222222222222222")?;
712        let cid = Cid {
713            hash,
714            key: Some(key),
715        };
716
717        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
718            .with_cached_tree_roots(vec![("npub1example/sites".to_string(), cid.clone())]);
719        let cached = server
720            .state
721            .tree_root_cache
722            .lock()
723            .unwrap()
724            .get("npub1example/sites")
725            .cloned()
726            .expect("seeded root");
727
728        assert_eq!(cached.cid, cid);
729        assert_eq!(cached.source, "embedded-bootstrap");
730        assert!(cached.root_event.is_none());
731        Ok(())
732    }
733
734    #[tokio::test]
735    async fn test_server_serve_file() -> Result<()> {
736        let temp_dir = TempDir::new()?;
737        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
738
739        // Create and upload a test file
740        let test_file = temp_dir.path().join("test.txt");
741        std::fs::write(&test_file, b"Hello, Hashtree!")?;
742
743        let cid = store.upload_file(&test_file)?;
744        let hash = from_hex(&cid)?;
745
746        // Verify we can get it
747        let content = store.get_file(&hash)?;
748        assert!(content.is_some());
749        assert_eq!(content.unwrap(), b"Hello, Hashtree!");
750
751        Ok(())
752    }
753
754    #[tokio::test]
755    async fn test_server_list_pins() -> Result<()> {
756        let temp_dir = TempDir::new()?;
757        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
758
759        let test_file = temp_dir.path().join("test.txt");
760        std::fs::write(&test_file, b"Test")?;
761
762        let cid = store.upload_file(&test_file)?;
763        let hash = from_hex(&cid)?;
764
765        let pins = store.list_pins_raw()?;
766        assert_eq!(pins.len(), 1);
767        assert_eq!(pins[0], hash);
768
769        Ok(())
770    }
771
772    async fn spawn_test_server(
773        store: Arc<HashtreeStore>,
774    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
775        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
776        let port = listener.local_addr()?.port();
777        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string());
778        let handle =
779            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
780        Ok((port, handle))
781    }
782
783    async fn spawn_test_server_with_nostr_relay(
784        store: Arc<HashtreeStore>,
785        relay: Arc<NostrRelay>,
786    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
787        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
788        let port = listener.local_addr()?.port();
789        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string()).with_nostr_relay(relay);
790        let handle =
791            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
792        Ok((port, handle))
793    }
794
795    #[tokio::test]
796    async fn virtual_tree_hosts_serve_root_assets_and_spa_fallbacks() -> Result<()> {
797        clear_virtual_tree_hosts_for_test();
798
799        let temp_dir = TempDir::new()?;
800        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
801        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
802
803        let (index_cid, _) = tree
804            .put(b"<!doctype html><title>Virtual host ok</title>")
805            .await?;
806        let (favicon_cid, _) = tree.put(b"ico").await?;
807        let (main_js_cid, _) = tree.put(b"console.log('ok');").await?;
808        let assets_dir = tree
809            .put_directory(vec![
810                DirEntry::from_cid("main.js", &main_js_cid).with_link_type(LinkType::File)
811            ])
812            .await?;
813        let root_cid = tree
814            .put_directory(vec![
815                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File),
816                DirEntry::from_cid("favicon.ico", &favicon_cid).with_link_type(LinkType::File),
817                DirEntry::from_cid("assets", &assets_dir).with_link_type(LinkType::Dir),
818            ])
819            .await?;
820        let nhash = nhash_encode(&root_cid.hash)?;
821        let host = "tree-test.htree.localhost";
822        register_virtual_tree_host(host, &format!("/htree/{nhash}"));
823
824        let (port, handle) = spawn_test_server(store).await?;
825        let base_url = format!("http://127.0.0.1:{port}");
826        let host_header = format!("{host}:{port}");
827        let client = reqwest::Client::new();
828
829        let root_response = client
830            .get(format!("{base_url}/"))
831            .header("Host", &host_header)
832            .header("Accept", "text/html")
833            .send()
834            .await?;
835        assert_eq!(root_response.status(), reqwest::StatusCode::OK);
836        assert_eq!(
837            root_response.bytes().await?.as_ref(),
838            b"<!doctype html><title>Virtual host ok</title>"
839        );
840
841        let favicon_response = client
842            .get(format!("{base_url}/favicon.ico"))
843            .header("Host", &host_header)
844            .send()
845            .await?;
846        assert_eq!(favicon_response.status(), reqwest::StatusCode::OK);
847        assert_eq!(favicon_response.bytes().await?.as_ref(), b"ico");
848
849        let js_response = client
850            .get(format!("{base_url}/assets/main.js"))
851            .header("Host", &host_header)
852            .send()
853            .await?;
854        assert_eq!(js_response.status(), reqwest::StatusCode::OK);
855        assert_eq!(js_response.bytes().await?.as_ref(), b"console.log('ok');");
856
857        let profile_response = client
858            .get(format!("{base_url}/users/npub1example"))
859            .header("Host", &host_header)
860            .header("Accept", "text/html")
861            .send()
862            .await?;
863        assert_eq!(profile_response.status(), reqwest::StatusCode::OK);
864        assert_eq!(
865            profile_response.bytes().await?.as_ref(),
866            b"<!doctype html><title>Virtual host ok</title>"
867        );
868
869        handle.abort();
870        clear_virtual_tree_hosts_for_test();
871
872        Ok(())
873    }
874
875    #[tokio::test]
876    async fn nostr_profile_route_returns_latest_metadata_event() -> Result<()> {
877        let temp_dir = TempDir::new()?;
878        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
879        let graph_store = {
880            let _guard = crate::socialgraph::test_lock();
881            crate::socialgraph::open_test_social_graph_store_with_mapsize(
882                &temp_dir.path().join("relay-db"),
883                Some(128 * 1024 * 1024),
884            )?
885        };
886        let backend: Arc<dyn crate::socialgraph::SocialGraphBackend> = graph_store;
887        let relay = Arc::new(NostrRelay::new(
888            backend,
889            temp_dir.path().to_path_buf(),
890            HashSet::new(),
891            None,
892            NostrRelayConfig {
893                spambox_db_max_bytes: 0,
894                ..Default::default()
895            },
896        )?);
897
898        let author = Keys::generate();
899        let older = EventBuilder::new(
900            Kind::Metadata,
901            json!({ "name": "older", "about": "before" }).to_string(),
902        )
903        .custom_created_at(Timestamp::from_secs(10))
904        .sign_with_keys(&author)?;
905        let newer = EventBuilder::new(
906            Kind::Metadata,
907            json!({ "name": "newer", "about": "after" }).to_string(),
908        )
909        .custom_created_at(Timestamp::from_secs(20))
910        .sign_with_keys(&author)?;
911
912        relay.ingest_trusted_event(older).await?;
913        relay.ingest_trusted_event(newer.clone()).await?;
914
915        let (port, handle) = spawn_test_server_with_nostr_relay(store, relay).await?;
916        let response = reqwest::get(format!(
917            "http://127.0.0.1:{port}/api/nostr/profile/{}",
918            author.public_key().to_hex()
919        ))
920        .await?;
921
922        assert_eq!(response.status(), reqwest::StatusCode::OK);
923        let payload: serde_json::Value = response.json().await?;
924        assert_eq!(payload["profile"]["name"].as_str(), Some("newer"),);
925        assert_eq!(payload["profile"]["about"].as_str(), Some("after"));
926        assert_eq!(payload["created_at"].as_u64(), Some(20));
927        let expected_event_id = newer.id.to_hex();
928        assert_eq!(
929            payload["event_id"].as_str(),
930            Some(expected_event_id.as_str())
931        );
932
933        handle.abort();
934        Ok(())
935    }
936}
hashtree_cli/server/mod.rs

hashtree_cli/server/
mod.rs
