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 = 256 * 1024 * 1024;
57const DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES: usize = 256 * 1024 * 1024;
58const INTERNAL_JSON_BODY_LIMIT_BYTES: usize = 64 * 1024;
59
60#[cfg(not(test))]
61const HTTP1_HEADER_READ_TIMEOUT: Duration = Duration::from_secs(30);
62#[cfg(test)]
63const HTTP1_HEADER_READ_TIMEOUT: Duration = Duration::from_millis(200);
64const HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
65const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);
66const TCP_KEEPALIVE_TIME: Duration = Duration::from_secs(60);
67const TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15);
68
69pub fn bounded_upload_queue_bytes(bytes: u64) -> usize {
70    usize::try_from(bytes)
71        .unwrap_or(usize::MAX)
72        .clamp(1, tokio::sync::Semaphore::MAX_PERMITS)
73}
74
75fn virtual_tree_hosts() -> &'static RwLock<HashMap<String, String>> {
76    VIRTUAL_TREE_HOSTS.get_or_init(|| RwLock::new(HashMap::new()))
77}
78
79fn normalize_virtual_tree_host(host: &str) -> Option<String> {
80    let trimmed = host.trim().trim_end_matches('.').to_ascii_lowercase();
81    if trimmed.is_empty() {
82        return None;
83    }
84
85    if let Some(stripped) = trimmed
86        .strip_prefix('[')
87        .and_then(|value| value.split_once(']'))
88    {
89        let host_only = stripped.0.trim();
90        if host_only.is_empty() {
91            return None;
92        }
93        return Some(host_only.to_string());
94    }
95
96    if let Some((host_only, port)) = trimmed.rsplit_once(':') {
97        if !host_only.is_empty() && !port.is_empty() && port.chars().all(|ch| ch.is_ascii_digit()) {
98            return Some(host_only.to_string());
99        }
100    }
101
102    Some(trimmed)
103}
104
105pub fn register_virtual_tree_host(host: &str, internal_root: &str) {
106    let Some(normalized_host) = normalize_virtual_tree_host(host) else {
107        return;
108    };
109
110    let normalized_root = internal_root.trim().trim_end_matches('/');
111    if normalized_root.is_empty() {
112        return;
113    }
114
115    if let Ok(mut hosts) = virtual_tree_hosts().write() {
116        hosts.insert(normalized_host, normalized_root.to_string());
117    }
118}
119
120pub fn resolve_virtual_tree_host(host: &str) -> Option<String> {
121    let normalized_host = normalize_virtual_tree_host(host)?;
122    let configured = virtual_tree_hosts()
123        .read()
124        .ok()
125        .and_then(|hosts| hosts.get(&normalized_host).cloned());
126    configured.or_else(|| resolve_iris_localhost_tree_root(&normalized_host))
127}
128
129fn resolve_iris_localhost_tree_root(host: &str) -> Option<String> {
130    let labels: Vec<&str> = host.strip_suffix(".iris.localhost")?.split('.').collect();
131    match labels.as_slice() {
132        [nhash] if nhash.starts_with("nhash1") => Some(format!("/htree/{nhash}")),
133        [site, npub] if !site.is_empty() && npub.starts_with("npub1") => {
134            Some(format!("/htree/{npub}/{site}"))
135        }
136        _ => None,
137    }
138}
139
140#[cfg(test)]
141pub fn clear_virtual_tree_hosts_for_test() {
142    if let Ok(mut hosts) = virtual_tree_hosts().write() {
143        hosts.clear();
144    }
145}
146
147pub struct HashtreeServer {
148    state: AppState,
149    addr: String,
150    extra_routes: Option<Router<AppState>>,
151    cors: Option<CorsLayer>,
152}
153
154impl HashtreeServer {
155    pub fn new(store: Arc<HashtreeStore>, addr: String) -> Self {
156        Self {
157            state: AppState {
158                store,
159                auth: None,
160                daemon_started_at: current_unix_secs(),
161                peer_mode: crate::config::ServerMode::Normal,
162                hash_get_enabled: true,
163                http_webrtc_fetch: true,
164                webrtc_peers: None,
165                fips_transport: None,
166                fetch_from_fips_peers: true,
167                ws_relay: Arc::new(auth::WsRelayState::new()),
168                max_upload_bytes: 5 * 1024 * 1024, // 5 MB default
169                public_writes: true,               // Allow anyone with valid Nostr auth by default
170                public_plaintext_reads: false,
171                require_random_untrusted_ingest: true,
172                optimistic_blossom_uploads: false,
173                optimistic_upload_queue_bytes: DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES,
174                optimistic_upload_queue: Arc::new(tokio::sync::Semaphore::new(
175                    DEFAULT_OPTIMISTIC_UPLOAD_QUEUE_BYTES,
176                )),
177                allowed_pubkeys: HashSet::new(), // No pubkeys allowed by default (use public_writes)
178                upstream_blossom: Vec::new(),
179                upstream_http_client: new_upstream_http_client(),
180                upstream_blossom_miss_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
181                upstream_blossom_fetch_metrics: Arc::new(
182                    auth::UpstreamBlossomFetchMetrics::default(),
183                ),
184                blossom_upload_replicas: Vec::new(),
185                blossom_upload_replica_queue_bytes: DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES,
186                blossom_upload_replica_queue: Arc::new(tokio::sync::Semaphore::new(
187                    DEFAULT_BLOSSOM_UPLOAD_REPLICA_QUEUE_BYTES,
188                )),
189                blossom_upload_replica_keys: None,
190                blossom_upload_replica_scheduler: Arc::new(
191                    blossom::BlossomUploadReplicaScheduler::new(),
192                ),
193                social_graph: None,
194                social_graph_store: None,
195                social_graph_root: None,
196                socialgraph_snapshot_public: false,
197                nostr_relay: None,
198                nostr_provider: None,
199                nostr_relay_urls: Vec::new(),
200                tree_root_cache: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
201                inflight_blob_fetches: Arc::new(tokio::sync::Mutex::new(
202                    std::collections::HashMap::new(),
203                )),
204                inflight_blob_reads: Arc::new(tokio::sync::Mutex::new(
205                    std::collections::HashMap::new(),
206                )),
207                blob_cache: Arc::new(crate::blob_cache::BlobCache::from_env()),
208                directory_listing_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
209                resolved_path_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
210                thumbnail_path_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
211                cid_size_cache: Arc::new(std::sync::Mutex::new(new_lookup_cache())),
212            },
213            addr,
214            extra_routes: None,
215            cors: None,
216        }
217    }
218
219    /// Set maximum upload size for Blossom uploads
220    pub fn with_max_upload_bytes(mut self, bytes: usize) -> Self {
221        self.state.max_upload_bytes = bytes;
222        self
223    }
224
225    /// Set whether to allow public writes (anyone with valid Nostr auth)
226    /// When false, only social graph members can write
227    pub fn with_public_writes(mut self, public: bool) -> Self {
228        self.state.public_writes = public;
229        self
230    }
231
232    /// Set whether mutable npub routes serve plaintext for unapproved pubkeys
233    pub fn with_public_plaintext_reads(mut self, public: bool) -> Self {
234        self.state.public_plaintext_reads = public;
235        self
236    }
237
238    pub fn with_require_random_untrusted_ingest(mut self, require: bool) -> Self {
239        self.state.require_random_untrusted_ingest = require;
240        self
241    }
242
243    pub fn with_optimistic_blossom_uploads(mut self, enabled: bool) -> Self {
244        self.state.optimistic_blossom_uploads = enabled;
245        self
246    }
247
248    pub fn with_server_mode(mut self, mode: crate::config::ServerMode) -> Self {
249        self.state.peer_mode = mode;
250        self
251    }
252
253    pub fn with_hash_get_enabled(mut self, enabled: bool) -> Self {
254        self.state.hash_get_enabled = enabled;
255        self
256    }
257
258    pub fn with_http_webrtc_fetch(mut self, enabled: bool) -> Self {
259        self.state.http_webrtc_fetch = enabled;
260        self
261    }
262
263    pub fn with_fetch_from_fips_peers(mut self, enabled: bool) -> Self {
264        self.state.fetch_from_fips_peers = enabled;
265        self
266    }
267
268    pub fn with_fips_transport(
269        mut self,
270        transport: Arc<crate::fips_transport::DaemonFipsTransport>,
271    ) -> Self {
272        self.state.fips_transport = Some(transport);
273        self
274    }
275
276    /// Set WebRTC state for P2P peer queries
277    pub fn with_webrtc_peers(mut self, webrtc_state: Arc<WebRTCState>) -> Self {
278        self.state.webrtc_peers = Some(webrtc_state);
279        self
280    }
281
282    pub fn with_auth(mut self, username: String, password: String) -> Self {
283        self.state.auth = Some(AuthCredentials { username, password });
284        self
285    }
286
287    /// Set allowed pubkeys for blossom write access (hex format)
288    pub fn with_allowed_pubkeys(mut self, pubkeys: HashSet<String>) -> Self {
289        self.state.allowed_pubkeys = pubkeys;
290        self
291    }
292
293    /// Set upstream Blossom servers for cascade fetching
294    pub fn with_upstream_blossom(mut self, servers: Vec<String>) -> Self {
295        self.state.upstream_blossom = servers;
296        self
297    }
298
299    /// Set write-behind Blossom servers for blobs accepted by this server.
300    pub fn with_blossom_upload_replicas(
301        mut self,
302        servers: Vec<String>,
303        queue_bytes: usize,
304        keys: nostr::Keys,
305    ) -> Self {
306        let queue_bytes = queue_bytes.clamp(1, tokio::sync::Semaphore::MAX_PERMITS);
307        let mut servers: Vec<String> = servers
308            .into_iter()
309            .map(|server| server.trim().trim_end_matches('/').to_string())
310            .filter(|server| !server.is_empty())
311            .collect();
312        servers.sort();
313        servers.dedup();
314        let replica_keys = (!servers.is_empty()).then(|| Arc::new(keys));
315        self.state.blossom_upload_replicas = servers;
316        self.state.blossom_upload_replica_queue_bytes = queue_bytes;
317        self.state.blossom_upload_replica_queue =
318            Arc::new(tokio::sync::Semaphore::new(queue_bytes));
319        self.state.blossom_upload_replica_keys = replica_keys;
320        self
321    }
322
323    /// Set social graph access control
324    pub fn with_social_graph(mut self, sg: Arc<socialgraph::SocialGraphAccessControl>) -> Self {
325        self.state.social_graph = Some(sg);
326        self
327    }
328
329    /// Configure social graph snapshot export (store handle + root)
330    pub fn with_socialgraph_snapshot(
331        mut self,
332        store: Arc<dyn socialgraph::SocialGraphBackend>,
333        root: [u8; 32],
334        public: bool,
335    ) -> Self {
336        self.state.social_graph_store = Some(store);
337        self.state.social_graph_root = Some(root);
338        self.state.socialgraph_snapshot_public = public;
339        self
340    }
341
342    /// Set Nostr relay state (shared for /ws and WebRTC)
343    pub fn with_nostr_relay(mut self, relay: Arc<NostrRelay>) -> Self {
344        self.state.nostr_relay = Some(relay);
345        self
346    }
347
348    pub fn with_nostr_provider(mut self, provider: Arc<dyn nostr_pubsub::PubsubProvider>) -> Self {
349        self.state.nostr_provider = Some(provider);
350        self
351    }
352
353    /// Set active upstream Nostr relays for HTTP resolver operations.
354    pub fn with_nostr_relay_urls(mut self, relays: Vec<String>) -> Self {
355        self.state.nostr_relay_urls = relays;
356        self
357    }
358
359    /// Seed mutable root cache entries before the server starts.
360    pub fn with_cached_tree_roots(self, roots: Vec<(String, Cid)>) -> Self {
361        if let Ok(mut cache) = self.state.tree_root_cache.lock() {
362            let now = Instant::now();
363            for (key, cid) in roots {
364                cache.insert(
365                    key,
366                    CachedTreeRootEntry {
367                        cid,
368                        source: "embedded-bootstrap",
369                        root_event: None,
370                        event: None,
371                        cached_at: now,
372                    },
373                );
374            }
375        }
376        self
377    }
378
379    /// Merge extra routes into the daemon router (e.g. Tauri embeds /nip07).
380    pub fn with_extra_routes(mut self, routes: Router<AppState>) -> Self {
381        self.extra_routes = Some(routes);
382        self
383    }
384
385    /// Apply a CORS layer to all routes (used by embedded clients like Tauri).
386    pub fn with_cors(mut self, cors: CorsLayer) -> Self {
387        self.cors = Some(cors);
388        self
389    }
390
391    pub async fn run(self) -> Result<()> {
392        let listener = tokio::net::TcpListener::bind(&self.addr).await?;
393        let _ = self.run_with_listener(listener).await?;
394        Ok(())
395    }
396
397    pub async fn run_with_listener(self, listener: tokio::net::TcpListener) -> Result<u16> {
398        self.run_with_listener_until(listener, future::pending::<()>())
399            .await
400    }
401
402    pub async fn run_with_listener_until<F>(
403        self,
404        listener: tokio::net::TcpListener,
405        shutdown: F,
406    ) -> Result<u16>
407    where
408        F: std::future::Future<Output = ()> + Send + 'static,
409    {
410        let local_addr = listener.local_addr()?;
411
412        // Public endpoints (no auth required)
413        // Note: /:id serves raw SHA256 blobs only. Logical tree/file assembly
414        // stays on explicitly configured mutable routes such as approved npub.
415        let state = self.state.clone();
416        let public_routes = Router::new()
417            .route("/", get(handlers::serve_root_or_virtual_host))
418            .route("/ws", get(ws_relay::ws_data))
419            .route("/ws/", get(ws_relay::ws_data))
420            .route(
421                "/__iris/store/:hash",
422                get(handlers::iris_store_get).head(handlers::iris_store_head),
423            )
424            .route(
425                "/htree/test",
426                get(handlers::htree_test).head(handlers::htree_test),
427            )
428            // /htree/nhash1...[/path] - content-addressed (immutable)
429            .route("/htree/nhash1:nhash", get(handlers::htree_nhash))
430            .route("/htree/nhash1:nhash/", get(handlers::htree_nhash))
431            .route("/htree/nhash1:nhash/*path", get(handlers::htree_nhash_path))
432            // /htree/npub1.../tree[/path] - mutable (resolver-backed)
433            .route("/htree/npub1:npub/:treename", get(handlers::htree_npub))
434            .route("/htree/npub1:npub/:treename/", get(handlers::htree_npub))
435            .route(
436                "/htree/npub1:npub/:treename/*path",
437                get(handlers::htree_npub_path),
438            )
439            // Nostr resolver endpoints - resolve npub/treename to content
440            .route("/n/:pubkey/:treename", get(handlers::resolve_and_serve))
441            // Direct npub route (clients should parse nhash and request by hex hash)
442            .route("/npub1:rest", get(handlers::serve_npub))
443            .route("/npub1:rest/*path", get(handlers::serve_npub))
444            // Blossom endpoints (BUD-01, BUD-02)
445            .route(
446                "/:id",
447                get(handlers::serve_content_or_blob)
448                    .head(blossom::head_blob)
449                    .delete(blossom::delete_blob)
450                    .options(blossom::cors_preflight),
451            )
452            .route(
453                "/upload",
454                put(blossom::upload_blob)
455                    .layer::<_, std::convert::Infallible>(middleware::from_fn(
456                        blossom::require_upload_auth_middleware,
457                    ))
458                    .layer(DefaultBodyLimit::max(blossom::MAX_SINGLE_UPLOAD_BODY_BYTES))
459                    .head(blossom::head_upload)
460                    .options(blossom::cors_preflight),
461            )
462            .route(
463                "/upload/batch",
464                post(blossom::upload_blob_batch)
465                    .layer::<_, std::convert::Infallible>(middleware::from_fn(
466                        blossom::require_upload_auth_middleware,
467                    ))
468                    .options(blossom::cors_preflight)
469                    .layer(DefaultBodyLimit::max(
470                        blossom::MAX_BATCH_UPLOAD_JSON_BODY_BYTES,
471                    )),
472            )
473            .route(
474                "/upload/batch-binary",
475                post(blossom::upload_blob_batch_binary)
476                    .layer::<_, std::convert::Infallible>(middleware::from_fn(
477                        blossom::require_upload_auth_middleware,
478                    ))
479                    .options(blossom::cors_preflight)
480                    .layer(DefaultBodyLimit::max(
481                        blossom::MAX_BATCH_UPLOAD_BINARY_BODY_BYTES,
482                    )),
483            )
484            .route(
485                "/upload/check",
486                post(blossom::upload_check).options(blossom::cors_preflight),
487            )
488            .route(
489                "/blob/batch",
490                post(handlers::download_blob_batch).options(blossom::cors_preflight),
491            )
492            .route(
493                "/list/:pubkey",
494                get(blossom::list_blobs).options(blossom::cors_preflight),
495            )
496            // Hashtree API endpoints
497            .route("/health", get(handlers::health_check))
498            .route("/api/pins", get(handlers::list_pins))
499            .route("/api/stats", get(handlers::storage_stats))
500            .route("/api/peers", get(handlers::webrtc_peers))
501            .route("/api/status", get(handlers::daemon_status))
502            .route("/api/p2p/signal", post(handlers::p2p_signal))
503            .route("/api/socialgraph", get(handlers::socialgraph_stats))
504            .route(
505                "/api/socialgraph/snapshot",
506                get(handlers::socialgraph_snapshot),
507            )
508            .route(
509                "/api/socialgraph/distance/:pubkey",
510                get(handlers::follow_distance),
511            )
512            // Resolver API endpoints
513            .route(
514                "/api/resolve/:pubkey/:treename",
515                get(handlers::resolve_to_hash),
516            )
517            .route(
518                "/api/nostr/resolve/:pubkey/:treename",
519                get(handlers::resolve_to_hash),
520            )
521            .route("/api/nostr/profile/:pubkey", get(handlers::nostr_profile))
522            .route(
523                "/api/nostr/events",
524                post(handlers::publish_nostr_event)
525                    .layer(DefaultBodyLimit::max(INTERNAL_JSON_BODY_LIMIT_BYTES)),
526            )
527            .route("/api/trees/:pubkey", get(handlers::list_trees))
528            .fallback(get(handlers::serve_virtual_host_fallback))
529            .with_state(state.clone());
530
531        // Protected endpoints (require auth if enabled)
532        let protected_routes = Router::new()
533            .route("/upload", post(handlers::upload_file))
534            .route("/api/pin/:cid", post(handlers::pin_cid))
535            .route("/api/unpin/:cid", post(handlers::unpin_cid))
536            .route("/api/gc", post(handlers::garbage_collect))
537            .layer(middleware::from_fn_with_state(
538                state.clone(),
539                auth::auth_middleware,
540            ))
541            .with_state(state.clone());
542
543        // Internal mutating endpoints require configured Basic auth. These
544        // routes stay closed even when optional API auth is disabled.
545        let internal_routes = Router::new()
546            .route(
547                "/__iris/store/:hash",
548                put(handlers::iris_store_put)
549                    .delete(handlers::iris_store_delete)
550                    .layer(DefaultBodyLimit::max(blossom::MAX_BATCH_UPLOAD_BYTES)),
551            )
552            .route(
553                "/api/cache-tree-root",
554                post(handlers::cache_tree_root)
555                    .layer(DefaultBodyLimit::max(INTERNAL_JSON_BODY_LIMIT_BYTES)),
556            )
557            .route(
558                "/api/clear-tree-root-cache",
559                post(handlers::clear_tree_root_cache)
560                    .layer(DefaultBodyLimit::max(INTERNAL_JSON_BODY_LIMIT_BYTES)),
561            )
562            .layer(middleware::from_fn_with_state(
563                state.clone(),
564                auth::require_auth_middleware,
565            ))
566            .with_state(state.clone());
567
568        let mut app = public_routes
569            .merge(protected_routes)
570            .merge(internal_routes)
571            .layer(DefaultBodyLimit::max(10 * 1024 * 1024 * 1024)) // 10GB limit
572            .layer(middleware::from_fn(status_metrics::record_http_status));
573
574        if let Some(extra) = self.extra_routes {
575            app = app.merge(extra.with_state(state));
576        }
577
578        if let Some(cors) = self.cors {
579            app = app.layer(cors);
580        }
581
582        let make_service = app.into_make_service_with_connect_info::<std::net::SocketAddr>();
583        serve_with_connection_limits(listener, make_service, shutdown).await?;
584
585        Ok(local_addr.port())
586    }
587
588    pub fn addr(&self) -> &str {
589        &self.addr
590    }
591}
592
593async fn serve_with_connection_limits<M, S, F>(
594    listener: tokio::net::TcpListener,
595    mut make_service: M,
596    shutdown: F,
597) -> io::Result<()>
598where
599    M: Service<SocketAddr, Error = Infallible, Response = S> + Send + 'static,
600    M::Future: Send,
601    S: Service<Request<Body>, Response = Response, Error = Infallible> + Clone + Send + 'static,
602    S::Future: Send,
603    F: std::future::Future<Output = ()> + Send + 'static,
604{
605    let (signal_tx, signal_rx) = watch::channel(());
606    let signal_tx = Arc::new(signal_tx);
607    tokio::spawn(async move {
608        shutdown.await;
609        trace!("received graceful shutdown signal; stopping daemon listener");
610        drop(signal_rx);
611    });
612
613    let (close_tx, close_rx) = watch::channel(());
614
615    loop {
616        let (tcp_stream, remote_addr) = tokio::select! {
617            accepted = accept_tcp(&listener) => {
618                match accepted? {
619                    Some(connection) => connection,
620                    None => continue,
621                }
622            }
623            _ = signal_tx.closed() => {
624                trace!("shutdown signal received; no longer accepting daemon connections");
625                break;
626            }
627        };
628
629        configure_tcp_stream(&tcp_stream);
630        let tcp_stream = TokioIo::new(tcp_stream);
631
632        poll_fn(|cx| make_service.poll_ready(cx))
633            .await
634            .unwrap_or_else(|err| match err {});
635
636        let tower_service = make_service
637            .call(remote_addr)
638            .await
639            .unwrap_or_else(|err| match err {})
640            .map_request(|req: Request<Incoming>| req.map(Body::new));
641        let hyper_service = TowerToHyperService::new(tower_service);
642
643        let signal_tx = Arc::clone(&signal_tx);
644        let close_rx = close_rx.clone();
645
646        tokio::spawn(async move {
647            let mut builder = HyperBuilder::new(TokioExecutor::new());
648            builder
649                .http1()
650                .timer(TokioTimer::new())
651                .header_read_timeout(HTTP1_HEADER_READ_TIMEOUT);
652            builder
653                .http2()
654                .timer(TokioTimer::new())
655                .keep_alive_interval(Some(HTTP2_KEEPALIVE_INTERVAL))
656                .keep_alive_timeout(HTTP2_KEEPALIVE_TIMEOUT);
657
658            let conn = builder.serve_connection_with_upgrades(tcp_stream, hyper_service);
659            pin_mut!(conn);
660
661            let signal_closed = signal_tx.closed().fuse();
662            pin_mut!(signal_closed);
663
664            loop {
665                tokio::select! {
666                    result = conn.as_mut() => {
667                        if let Err(err) = result {
668                            trace!("daemon connection closed with error: {err:#}");
669                        }
670                        break;
671                    }
672                    _ = &mut signal_closed => {
673                        trace!("shutdown signal received by connection task");
674                        conn.as_mut().graceful_shutdown();
675                    }
676                }
677            }
678
679            drop(close_rx);
680        });
681    }
682
683    drop(close_rx);
684    drop(listener);
685    close_tx.closed().await;
686
687    Ok(())
688}
689
690fn configure_tcp_stream(tcp_stream: &tokio::net::TcpStream) {
691    if let Err(err) = tcp_stream.set_nodelay(true) {
692        debug!("failed to set TCP_NODELAY on daemon connection: {err:#}");
693    }
694
695    let socket = SockRef::from(tcp_stream);
696    if let Err(err) = socket.set_tcp_keepalive(
697        &TcpKeepalive::new()
698            .with_time(TCP_KEEPALIVE_TIME)
699            .with_interval(TCP_KEEPALIVE_INTERVAL),
700    ) {
701        debug!("failed to set TCP keepalive on daemon connection: {err:#}");
702    }
703}
704
705async fn accept_tcp(
706    listener: &tokio::net::TcpListener,
707) -> io::Result<Option<(tokio::net::TcpStream, SocketAddr)>> {
708    match listener.accept().await {
709        Ok(connection) => Ok(Some(connection)),
710        Err(err) => {
711            if is_connection_error(&err) {
712                return Ok(None);
713            }
714            if is_resource_exhaustion_error(&err) {
715                error!(
716                    "daemon accept failed due to file descriptor exhaustion; exiting for supervisor restart: {err}"
717                );
718                return Err(err);
719            }
720            error!("daemon accept error: {err}");
721            tokio::time::sleep(Duration::from_secs(1)).await;
722            Ok(None)
723        }
724    }
725}
726
727fn is_resource_exhaustion_error(err: &io::Error) -> bool {
728    matches!(
729        err.raw_os_error(),
730        Some(code) if code == libc::EMFILE || code == libc::ENFILE
731    )
732}
733
734fn is_connection_error(err: &io::Error) -> bool {
735    matches!(
736        err.kind(),
737        io::ErrorKind::ConnectionRefused
738            | io::ErrorKind::ConnectionAborted
739            | io::ErrorKind::ConnectionReset
740    )
741}
742
743fn current_unix_secs() -> u64 {
744    std::time::SystemTime::now()
745        .duration_since(std::time::UNIX_EPOCH)
746        .unwrap_or(std::time::Duration::ZERO)
747        .as_secs()
748}
749
750#[cfg(test)]
751mod tests {
752    use super::*;
753    use crate::nostr_relay::{NostrRelay, NostrRelayConfig};
754    use crate::storage::HashtreeStore;
755    use async_trait::async_trait;
756    use hashtree_core::{
757        from_hex, nhash_encode, sha256, to_hex, DirEntry, HashTree, HashTreeConfig, LinkType,
758    };
759    use nostr::{nips::nip19::ToBech32, EventBuilder, Keys, Kind, Timestamp};
760    use nostr_pubsub::{
761        EventBus, EventSource, PublishReport, PubsubProvider, PubsubProviderMode, QueryEvent,
762        QueryOptions, QueryReport, VerifiedEvent,
763    };
764    use serde_json::json;
765    use std::sync::atomic::{AtomicUsize, Ordering};
766    use tempfile::TempDir;
767
768    struct StaticProvider {
769        event: Option<VerifiedEvent>,
770        queries: AtomicUsize,
771        publishes: AtomicUsize,
772        mode: PubsubProviderMode,
773    }
774
775    #[async_trait]
776    impl EventBus for StaticProvider {
777        async fn publish(
778            &self,
779            _event: VerifiedEvent,
780            _source: EventSource,
781        ) -> nostr_pubsub::Result<PublishReport> {
782            self.publishes.fetch_add(1, Ordering::Relaxed);
783            Ok(PublishReport {
784                accepted: true,
785                priority: 0,
786                reason: None,
787            })
788        }
789
790        async fn query(
791            &self,
792            _filters: Vec<nostr_pubsub::Filter>,
793            _options: QueryOptions,
794        ) -> nostr_pubsub::Result<QueryReport> {
795            self.queries.fetch_add(1, Ordering::Relaxed);
796            Ok(QueryReport {
797                events: self
798                    .event
799                    .clone()
800                    .map(|event| QueryEvent {
801                        event,
802                        source: EventSource::fips_endpoint("browser-router"),
803                        priority: 0,
804                    })
805                    .into_iter()
806                    .collect(),
807            })
808        }
809    }
810
811    impl PubsubProvider for StaticProvider {
812        fn mode(&self) -> PubsubProviderMode {
813            self.mode
814        }
815    }
816
817    #[test]
818    fn resource_exhaustion_errors_are_fatal_accept_errors() {
819        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
820            libc::EMFILE
821        )));
822        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
823            libc::ENFILE
824        )));
825        assert!(!is_resource_exhaustion_error(
826            &io::Error::from_raw_os_error(libc::ECONNRESET)
827        ));
828    }
829
830    #[test]
831    fn upload_queue_semaphores_fit_all_tokio_targets() -> Result<()> {
832        let temp_dir = TempDir::new()?;
833        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
834        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string());
835
836        assert_eq!(
837            server.state.optimistic_upload_queue.available_permits(),
838            256 * 1024 * 1024
839        );
840        assert_eq!(
841            server
842                .state
843                .blossom_upload_replica_queue
844                .available_permits(),
845            256 * 1024 * 1024
846        );
847        assert_eq!(
848            bounded_upload_queue_bytes(u64::MAX),
849            tokio::sync::Semaphore::MAX_PERMITS
850        );
851        Ok(())
852    }
853
854    #[test]
855    fn server_builder_seeds_initial_tree_roots() -> Result<()> {
856        let temp_dir = TempDir::new()?;
857        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
858        let hash = from_hex("1111111111111111111111111111111111111111111111111111111111111111")?;
859        let key = from_hex("2222222222222222222222222222222222222222222222222222222222222222")?;
860        let cid = Cid {
861            hash,
862            key: Some(key),
863        };
864
865        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
866            .with_cached_tree_roots(vec![("npub1example/sites".to_string(), cid.clone())]);
867        let cached = server
868            .state
869            .tree_root_cache
870            .lock()
871            .unwrap()
872            .get("npub1example/sites")
873            .cloned()
874            .expect("seeded root");
875
876        assert_eq!(cached.cid, cid);
877        assert_eq!(cached.source, "embedded-bootstrap");
878        assert!(cached.root_event.is_none());
879        Ok(())
880    }
881
882    #[test]
883    fn iris_localhost_hosts_map_to_existing_tree_routes() {
884        assert_eq!(
885            resolve_virtual_tree_host("NHASH1EXAMPLE.iris.localhost:8080"),
886            Some("/htree/nhash1example".to_string())
887        );
888        assert_eq!(
889            resolve_virtual_tree_host("audio.NPUB1EXAMPLE.iris.localhost:8080"),
890            Some("/htree/npub1example/audio".to_string())
891        );
892        assert_eq!(
893            resolve_virtual_tree_host("audio.extra.npub1example.iris.localhost"),
894            None
895        );
896        assert_eq!(
897            resolve_virtual_tree_host("nhash1example.htree.localhost"),
898            None
899        );
900    }
901
902    #[tokio::test]
903    async fn test_server_serve_file() -> Result<()> {
904        let temp_dir = TempDir::new()?;
905        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
906
907        // Create and upload a test file
908        let test_file = temp_dir.path().join("test.txt");
909        std::fs::write(&test_file, b"Hello, Hashtree!")?;
910
911        let cid = store.upload_file(&test_file)?;
912        let hash = from_hex(&cid)?;
913
914        // Verify we can get it
915        let content = store.get_file(&hash)?;
916        assert!(content.is_some());
917        assert_eq!(content.unwrap(), b"Hello, Hashtree!");
918
919        Ok(())
920    }
921
922    #[tokio::test]
923    async fn test_server_list_pins() -> Result<()> {
924        let temp_dir = TempDir::new()?;
925        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
926
927        let test_file = temp_dir.path().join("test.txt");
928        std::fs::write(&test_file, b"Test")?;
929
930        let cid = store.upload_file(&test_file)?;
931        let hash = from_hex(&cid)?;
932
933        let pins = store.list_pins_raw()?;
934        assert_eq!(pins.len(), 1);
935        assert_eq!(pins[0], hash);
936
937        Ok(())
938    }
939
940    async fn spawn_test_server(
941        store: Arc<HashtreeStore>,
942    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
943        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
944        let port = listener.local_addr()?.port();
945        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string());
946        let handle =
947            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
948        Ok((port, handle))
949    }
950
951    async fn spawn_test_server_with_nostr_relay(
952        store: Arc<HashtreeStore>,
953        relay: Arc<NostrRelay>,
954    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
955        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
956        let port = listener.local_addr()?.port();
957        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string()).with_nostr_relay(relay);
958        let handle =
959            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
960        Ok((port, handle))
961    }
962
963    #[tokio::test]
964    async fn unauthenticated_native_store_mutation_is_rejected() -> Result<()> {
965        let temp_dir = TempDir::new()?;
966        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
967        let body = b"unauthorized native store write";
968        let hash = sha256(body);
969        let hash_hex = to_hex(&hash);
970        let (port, handle) = spawn_test_server(Arc::clone(&store)).await?;
971
972        let response = reqwest::Client::new()
973            .put(format!("http://127.0.0.1:{port}/__iris/store/{hash_hex}"))
974            .body(body.to_vec())
975            .send()
976            .await?;
977
978        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
979        assert!(store.get_blob(&hash)?.is_none());
980        handle.abort();
981        Ok(())
982    }
983
984    #[tokio::test]
985    async fn unauthenticated_cache_tree_root_is_rejected() -> Result<()> {
986        let temp_dir = TempDir::new()?;
987        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
988        let (port, handle) = spawn_test_server(store).await?;
989
990        let response = reqwest::Client::new()
991            .post(format!("http://127.0.0.1:{port}/api/cache-tree-root"))
992            .json(&json!({
993                "npub": "npub1example",
994                "treeName": "video",
995                "hash": "988db3f24dc222715f1c1e1fa5876690d3147122243d72d85fd44283867cd61a",
996                "visibility": "public"
997            }))
998            .send()
999            .await?;
1000
1001        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
1002        handle.abort();
1003        Ok(())
1004    }
1005
1006    #[tokio::test]
1007    async fn unauthenticated_upload_batch_is_rejected_before_json_extraction() -> Result<()> {
1008        let temp_dir = TempDir::new()?;
1009        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1010        let (port, handle) = spawn_test_server(store).await?;
1011
1012        let response = reqwest::Client::new()
1013            .post(format!("http://127.0.0.1:{port}/upload/batch"))
1014            .header(reqwest::header::CONTENT_TYPE, "application/json")
1015            .body("not valid json")
1016            .send()
1017            .await?;
1018
1019        assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
1020        handle.abort();
1021        Ok(())
1022    }
1023
1024    #[tokio::test]
1025    async fn unauthenticated_single_upload_is_rejected_before_body_limit() -> Result<()> {
1026        let temp_dir = TempDir::new()?;
1027        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1028        let (port, handle) = spawn_test_server(store).await?;
1029        let body = vec![0_u8; blossom::MAX_SINGLE_UPLOAD_BODY_BYTES + 1];
1030
1031        let response = reqwest::Client::new()
1032            .put(format!("http://127.0.0.1:{port}/upload"))
1033            .body(body)
1034            .send()
1035            .await?;
1036
1037        assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
1038        handle.abort();
1039        Ok(())
1040    }
1041
1042    #[tokio::test]
1043    async fn upload_options_preflight_does_not_require_auth() -> Result<()> {
1044        let temp_dir = TempDir::new()?;
1045        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1046        let (port, handle) = spawn_test_server(store).await?;
1047
1048        let response = reqwest::Client::new()
1049            .request(
1050                reqwest::Method::OPTIONS,
1051                format!("http://127.0.0.1:{port}/upload"),
1052            )
1053            .send()
1054            .await?;
1055
1056        assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
1057        handle.abort();
1058        Ok(())
1059    }
1060
1061    #[tokio::test]
1062    async fn virtual_tree_hosts_serve_root_assets_and_spa_fallbacks() -> Result<()> {
1063        clear_virtual_tree_hosts_for_test();
1064
1065        let temp_dir = TempDir::new()?;
1066        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1067        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1068
1069        let (index_cid, _) = tree
1070            .put(b"<!doctype html><title>Virtual host ok</title>")
1071            .await?;
1072        let (favicon_cid, _) = tree.put(b"ico").await?;
1073        let (main_js_cid, _) = tree.put(b"console.log('ok');").await?;
1074        let assets_dir = tree
1075            .put_directory(vec![
1076                DirEntry::from_cid("main.js", &main_js_cid).with_link_type(LinkType::File)
1077            ])
1078            .await?;
1079        let root_cid = tree
1080            .put_directory(vec![
1081                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File),
1082                DirEntry::from_cid("favicon.ico", &favicon_cid).with_link_type(LinkType::File),
1083                DirEntry::from_cid("assets", &assets_dir).with_link_type(LinkType::Dir),
1084            ])
1085            .await?;
1086        let nhash = nhash_encode(&root_cid.hash)?;
1087        let host = "tree-test.htree.localhost";
1088        register_virtual_tree_host(host, &format!("/htree/{nhash}"));
1089
1090        let (port, handle) = spawn_test_server(store).await?;
1091        let base_url = format!("http://127.0.0.1:{port}");
1092        let host_header = format!("{host}:{port}");
1093        let client = reqwest::Client::new();
1094
1095        let root_response = client
1096            .get(format!("{base_url}/"))
1097            .header("Host", &host_header)
1098            .header("Accept", "text/html")
1099            .send()
1100            .await?;
1101        assert_eq!(root_response.status(), reqwest::StatusCode::OK);
1102        assert_eq!(
1103            root_response.bytes().await?.as_ref(),
1104            b"<!doctype html><title>Virtual host ok</title>"
1105        );
1106
1107        let favicon_response = client
1108            .get(format!("{base_url}/favicon.ico"))
1109            .header("Host", &host_header)
1110            .send()
1111            .await?;
1112        assert_eq!(favicon_response.status(), reqwest::StatusCode::OK);
1113        assert_eq!(favicon_response.bytes().await?.as_ref(), b"ico");
1114
1115        let js_response = client
1116            .get(format!("{base_url}/assets/main.js"))
1117            .header("Host", &host_header)
1118            .send()
1119            .await?;
1120        assert_eq!(js_response.status(), reqwest::StatusCode::OK);
1121        assert_eq!(js_response.bytes().await?.as_ref(), b"console.log('ok');");
1122
1123        let profile_response = client
1124            .get(format!("{base_url}/users/npub1example"))
1125            .header("Host", &host_header)
1126            .header("Accept", "text/html")
1127            .send()
1128            .await?;
1129        assert_eq!(profile_response.status(), reqwest::StatusCode::OK);
1130        assert_eq!(
1131            profile_response.bytes().await?.as_ref(),
1132            b"<!doctype html><title>Virtual host ok</title>"
1133        );
1134
1135        handle.abort();
1136        clear_virtual_tree_hosts_for_test();
1137
1138        Ok(())
1139    }
1140
1141    #[tokio::test]
1142    async fn iris_localhost_hosts_serve_immutable_and_named_sites() -> Result<()> {
1143        let temp_dir = TempDir::new()?;
1144        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1145        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1146
1147        let (index_cid, _) = tree
1148            .put(b"<!doctype html><title>Iris localhost ok</title>")
1149            .await?;
1150        let (main_js_cid, _) = tree.put(b"console.log('iris localhost');").await?;
1151        let assets_dir = tree
1152            .put_directory(vec![
1153                DirEntry::from_cid("main.js", &main_js_cid).with_link_type(LinkType::File)
1154            ])
1155            .await?;
1156        let root_cid = tree
1157            .put_directory(vec![
1158                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File),
1159                DirEntry::from_cid("assets", &assets_dir).with_link_type(LinkType::Dir),
1160            ])
1161            .await?;
1162
1163        let nhash = nhash_encode(&root_cid.hash)?;
1164        let npub = Keys::generate().public_key().to_bech32()?;
1165        let site = "audio";
1166        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1167        let port = listener.local_addr()?.port();
1168        let server = HashtreeServer::new(Arc::clone(&store), "127.0.0.1:0".to_string())
1169            .with_public_plaintext_reads(true)
1170            .with_cached_tree_roots(vec![(format!("{npub}/{site}"), root_cid)]);
1171        let handle =
1172            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1173        let client = reqwest::Client::new();
1174        let base_url = format!("http://127.0.0.1:{port}");
1175
1176        for host in [
1177            format!("{nhash}.iris.localhost:{port}"),
1178            format!("{site}.{npub}.iris.localhost:{port}"),
1179        ] {
1180            let root_response = client
1181                .get(format!("{base_url}/"))
1182                .header("Host", &host)
1183                .header("Accept", "text/html")
1184                .send()
1185                .await?;
1186            assert_eq!(root_response.status(), reqwest::StatusCode::OK, "{host}");
1187            assert_eq!(
1188                root_response.bytes().await?.as_ref(),
1189                b"<!doctype html><title>Iris localhost ok</title>"
1190            );
1191
1192            let asset_response = client
1193                .get(format!("{base_url}/assets/main.js"))
1194                .header("Host", &host)
1195                .send()
1196                .await?;
1197            assert_eq!(asset_response.status(), reqwest::StatusCode::OK, "{host}");
1198            assert_eq!(
1199                asset_response.bytes().await?.as_ref(),
1200                b"console.log('iris localhost');"
1201            );
1202        }
1203
1204        handle.abort();
1205        Ok(())
1206    }
1207
1208    #[tokio::test]
1209    async fn named_iris_site_resolves_from_fips_event_provider_without_relays() -> Result<()> {
1210        let temp_dir = TempDir::new()?;
1211        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1212        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1213        let (index_cid, _) = tree.put(b"fips event provider site").await?;
1214        let root_cid = tree
1215            .put_directory(vec![
1216                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File)
1217            ])
1218            .await?;
1219        let owner = Keys::generate();
1220        let site = "radio";
1221        let root_event = EventBuilder::new(Kind::Custom(30064), "")
1222            .tags(vec![
1223                nostr::Tag::identifier(site),
1224                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["hashtree".to_string()]),
1225                nostr::Tag::custom(nostr::TagKind::custom("hash"), vec![to_hex(&root_cid.hash)]),
1226            ])
1227            .sign_with_keys(&owner)?;
1228        let provider = Arc::new(StaticProvider {
1229            event: Some(VerifiedEvent::try_from(root_event)?),
1230            queries: AtomicUsize::new(0),
1231            publishes: AtomicUsize::new(0),
1232            mode: PubsubProviderMode::LocalOnly,
1233        });
1234
1235        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1236        let port = listener.local_addr()?.port();
1237        let server = HashtreeServer::new(Arc::clone(&store), "127.0.0.1:0".to_string())
1238            .with_public_plaintext_reads(true)
1239            .with_nostr_provider(provider.clone());
1240        assert!(server.state.nostr_relay_urls.is_empty());
1241        assert_eq!(
1242            server.state.nostr_provider.as_ref().unwrap().mode(),
1243            PubsubProviderMode::LocalOnly
1244        );
1245        let handle =
1246            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1247
1248        let npub = owner.public_key().to_bech32()?;
1249        let response = reqwest::Client::new()
1250            .get(format!("http://127.0.0.1:{port}/"))
1251            .header("Host", format!("{site}.{npub}.iris.localhost"))
1252            .header("Accept", "text/html")
1253            .send()
1254            .await?;
1255
1256        assert_eq!(response.status(), reqwest::StatusCode::OK);
1257        assert_eq!(
1258            response.bytes().await?.as_ref(),
1259            b"fips event provider site"
1260        );
1261        assert_eq!(provider.queries.load(Ordering::Relaxed), 1);
1262        handle.abort();
1263        Ok(())
1264    }
1265
1266    #[tokio::test]
1267    async fn loopback_event_publication_uses_configured_provider() -> Result<()> {
1268        let temp_dir = TempDir::new()?;
1269        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1270        let keys = Keys::generate();
1271        let event = EventBuilder::new(Kind::TextNote, "provider test").sign_with_keys(&keys)?;
1272        let provider = Arc::new(StaticProvider {
1273            event: Some(VerifiedEvent::try_from(event.clone())?),
1274            queries: AtomicUsize::new(0),
1275            publishes: AtomicUsize::new(0),
1276            mode: PubsubProviderMode::LocalOnly,
1277        });
1278        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1279        let port = listener.local_addr()?.port();
1280        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
1281            .with_nostr_provider(provider.clone());
1282        assert!(server.state.nostr_relay_urls.is_empty());
1283        let handle =
1284            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1285
1286        let response = reqwest::Client::new()
1287            .post(format!("http://127.0.0.1:{port}/api/nostr/events"))
1288            .json(&event)
1289            .send()
1290            .await?;
1291
1292        assert_eq!(response.status(), reqwest::StatusCode::ACCEPTED);
1293        assert_eq!(provider.publishes.load(Ordering::Relaxed), 1);
1294        handle.abort();
1295        Ok(())
1296    }
1297
1298    #[tokio::test]
1299    async fn loopback_root_publication_is_immediately_resolvable_without_provider_replay(
1300    ) -> Result<()> {
1301        let temp_dir = TempDir::new()?;
1302        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1303        let keys = Keys::generate();
1304        let tree_name = "webvm-e2e";
1305        let root_hash = "11".repeat(32);
1306        let encryption_key = "22".repeat(32);
1307        let event = EventBuilder::new(Kind::Custom(30064), &root_hash)
1308            .tags(vec![
1309                nostr::Tag::identifier(tree_name),
1310                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["hashtree".to_string()]),
1311                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["git".to_string()]),
1312                nostr::Tag::custom(nostr::TagKind::custom("hash"), vec![root_hash.clone()]),
1313                nostr::Tag::custom(nostr::TagKind::custom("key"), vec![encryption_key.clone()]),
1314            ])
1315            .sign_with_keys(&keys)?;
1316        let provider = Arc::new(StaticProvider {
1317            event: None,
1318            queries: AtomicUsize::new(0),
1319            publishes: AtomicUsize::new(0),
1320            mode: PubsubProviderMode::LocalOnly,
1321        });
1322        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1323        let port = listener.local_addr()?.port();
1324        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
1325            .with_nostr_provider(provider.clone());
1326        assert!(server.state.nostr_relay_urls.is_empty());
1327        let handle =
1328            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1329        let client = reqwest::Client::new();
1330
1331        let publish = client
1332            .post(format!("http://127.0.0.1:{port}/api/nostr/events"))
1333            .json(&event)
1334            .send()
1335            .await?;
1336        assert_eq!(publish.status(), reqwest::StatusCode::ACCEPTED);
1337
1338        let npub = keys.public_key().to_bech32()?;
1339        let resolved = client
1340            .get(format!(
1341                "http://127.0.0.1:{port}/api/nostr/resolve/{npub}/{tree_name}?refresh=1"
1342            ))
1343            .send()
1344            .await?
1345            .json::<serde_json::Value>()
1346            .await?;
1347
1348        assert_eq!(resolved["hash"], root_hash);
1349        assert_eq!(resolved["key_tag"], encryption_key);
1350        assert_eq!(resolved["source"], "local-relay");
1351        assert_eq!(provider.publishes.load(Ordering::Relaxed), 1);
1352        assert_eq!(provider.queries.load(Ordering::Relaxed), 1);
1353        handle.abort();
1354        Ok(())
1355    }
1356
1357    #[tokio::test]
1358    async fn nostr_profile_route_returns_latest_metadata_event() -> Result<()> {
1359        let temp_dir = TempDir::new()?;
1360        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1361        let graph_store = {
1362            let _guard = crate::socialgraph::test_lock();
1363            crate::socialgraph::open_test_social_graph_store_with_mapsize(
1364                &temp_dir.path().join("relay-db"),
1365                Some(128 * 1024 * 1024),
1366            )?
1367        };
1368        let backend: Arc<dyn crate::socialgraph::SocialGraphBackend> = graph_store;
1369        let relay = Arc::new(NostrRelay::new(
1370            backend,
1371            temp_dir.path().to_path_buf(),
1372            HashSet::new(),
1373            None,
1374            NostrRelayConfig {
1375                spambox_db_max_bytes: 0,
1376                ..Default::default()
1377            },
1378        )?);
1379
1380        let author = Keys::generate();
1381        let older = EventBuilder::new(
1382            Kind::Metadata,
1383            json!({ "name": "older", "about": "before" }).to_string(),
1384        )
1385        .custom_created_at(Timestamp::from_secs(10))
1386        .sign_with_keys(&author)?;
1387        let newer = EventBuilder::new(
1388            Kind::Metadata,
1389            json!({ "name": "newer", "about": "after" }).to_string(),
1390        )
1391        .custom_created_at(Timestamp::from_secs(20))
1392        .sign_with_keys(&author)?;
1393
1394        relay.ingest_trusted_event(older).await?;
1395        relay.ingest_trusted_event(newer.clone()).await?;
1396
1397        let (port, handle) = spawn_test_server_with_nostr_relay(store, relay).await?;
1398        let response = reqwest::get(format!(
1399            "http://127.0.0.1:{port}/api/nostr/profile/{}",
1400            author.public_key().to_hex()
1401        ))
1402        .await?;
1403
1404        assert_eq!(response.status(), reqwest::StatusCode::OK);
1405        let payload: serde_json::Value = response.json().await?;
1406        assert_eq!(payload["profile"]["name"].as_str(), Some("newer"),);
1407        assert_eq!(payload["profile"]["about"].as_str(), Some("after"));
1408        assert_eq!(payload["created_at"].as_u64(), Some(20));
1409        let expected_event_id = newer.id.to_hex();
1410        assert_eq!(
1411            payload["event_id"].as_str(),
1412            Some(expected_event_id.as_str())
1413        );
1414
1415        handle.abort();
1416        Ok(())
1417    }
1418}
hashtree_cli/server/mod.rs

hashtree_cli/server/
mod.rs
