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    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
768
769    struct StaticProvider {
770        event: Option<VerifiedEvent>,
771        queries: AtomicUsize,
772        publishes: AtomicUsize,
773        mode: PubsubProviderMode,
774    }
775
776    #[async_trait]
777    impl EventBus for StaticProvider {
778        async fn publish(
779            &self,
780            _event: VerifiedEvent,
781            _source: EventSource,
782        ) -> nostr_pubsub::Result<PublishReport> {
783            self.publishes.fetch_add(1, Ordering::Relaxed);
784            Ok(PublishReport {
785                accepted: true,
786                priority: 0,
787                reason: None,
788            })
789        }
790
791        async fn query(
792            &self,
793            _filters: Vec<nostr_pubsub::Filter>,
794            _options: QueryOptions,
795        ) -> nostr_pubsub::Result<QueryReport> {
796            self.queries.fetch_add(1, Ordering::Relaxed);
797            Ok(QueryReport {
798                events: self
799                    .event
800                    .clone()
801                    .map(|event| QueryEvent {
802                        event,
803                        source: EventSource::fips_endpoint("browser-router"),
804                        priority: 0,
805                    })
806                    .into_iter()
807                    .collect(),
808            })
809        }
810    }
811
812    impl PubsubProvider for StaticProvider {
813        fn mode(&self) -> PubsubProviderMode {
814            self.mode
815        }
816    }
817
818    #[test]
819    fn resource_exhaustion_errors_are_fatal_accept_errors() {
820        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
821            libc::EMFILE
822        )));
823        assert!(is_resource_exhaustion_error(&io::Error::from_raw_os_error(
824            libc::ENFILE
825        )));
826        assert!(!is_resource_exhaustion_error(
827            &io::Error::from_raw_os_error(libc::ECONNRESET)
828        ));
829    }
830
831    #[test]
832    fn upload_queue_semaphores_fit_all_tokio_targets() -> Result<()> {
833        let temp_dir = TempDir::new()?;
834        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
835        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string());
836
837        assert_eq!(
838            server.state.optimistic_upload_queue.available_permits(),
839            256 * 1024 * 1024
840        );
841        assert_eq!(
842            server
843                .state
844                .blossom_upload_replica_queue
845                .available_permits(),
846            256 * 1024 * 1024
847        );
848        assert_eq!(
849            bounded_upload_queue_bytes(u64::MAX),
850            tokio::sync::Semaphore::MAX_PERMITS
851        );
852        Ok(())
853    }
854
855    #[test]
856    fn server_builder_seeds_initial_tree_roots() -> Result<()> {
857        let temp_dir = TempDir::new()?;
858        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
859        let hash = from_hex("1111111111111111111111111111111111111111111111111111111111111111")?;
860        let key = from_hex("2222222222222222222222222222222222222222222222222222222222222222")?;
861        let cid = Cid {
862            hash,
863            key: Some(key),
864        };
865
866        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
867            .with_cached_tree_roots(vec![("npub1example/sites".to_string(), cid.clone())]);
868        let cached = server
869            .state
870            .tree_root_cache
871            .lock()
872            .unwrap()
873            .get("npub1example/sites")
874            .cloned()
875            .expect("seeded root");
876
877        assert_eq!(cached.cid, cid);
878        assert_eq!(cached.source, "embedded-bootstrap");
879        assert!(cached.root_event.is_none());
880        Ok(())
881    }
882
883    #[test]
884    fn iris_localhost_hosts_map_to_existing_tree_routes() {
885        assert_eq!(
886            resolve_virtual_tree_host("NHASH1EXAMPLE.iris.localhost:8080"),
887            Some("/htree/nhash1example".to_string())
888        );
889        assert_eq!(
890            resolve_virtual_tree_host("audio.NPUB1EXAMPLE.iris.localhost:8080"),
891            Some("/htree/npub1example/audio".to_string())
892        );
893        assert_eq!(
894            resolve_virtual_tree_host("audio.extra.npub1example.iris.localhost"),
895            None
896        );
897        assert_eq!(
898            resolve_virtual_tree_host("nhash1example.htree.localhost"),
899            None
900        );
901    }
902
903    #[tokio::test]
904    async fn test_server_serve_file() -> Result<()> {
905        let temp_dir = TempDir::new()?;
906        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
907
908        // Create and upload a test file
909        let test_file = temp_dir.path().join("test.txt");
910        std::fs::write(&test_file, b"Hello, Hashtree!")?;
911
912        let cid = store.upload_file(&test_file)?;
913        let hash = from_hex(&cid)?;
914
915        // Verify we can get it
916        let content = store.get_file(&hash)?;
917        assert!(content.is_some());
918        assert_eq!(content.unwrap(), b"Hello, Hashtree!");
919
920        Ok(())
921    }
922
923    #[tokio::test]
924    async fn test_server_list_pins() -> Result<()> {
925        let temp_dir = TempDir::new()?;
926        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
927
928        let test_file = temp_dir.path().join("test.txt");
929        std::fs::write(&test_file, b"Test")?;
930
931        let cid = store.upload_file(&test_file)?;
932        let hash = from_hex(&cid)?;
933
934        let pins = store.list_pins_raw()?;
935        assert_eq!(pins.len(), 1);
936        assert_eq!(pins[0], hash);
937
938        Ok(())
939    }
940
941    async fn spawn_test_server(
942        store: Arc<HashtreeStore>,
943    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
944        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
945        let port = listener.local_addr()?.port();
946        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string());
947        let handle =
948            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
949        Ok((port, handle))
950    }
951
952    async fn spawn_test_server_with_nostr_relay(
953        store: Arc<HashtreeStore>,
954        relay: Arc<NostrRelay>,
955    ) -> Result<(u16, tokio::task::JoinHandle<Result<()>>)> {
956        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
957        let port = listener.local_addr()?.port();
958        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string()).with_nostr_relay(relay);
959        let handle =
960            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
961        Ok((port, handle))
962    }
963
964    #[tokio::test]
965    async fn unauthenticated_native_store_mutation_is_rejected() -> Result<()> {
966        let temp_dir = TempDir::new()?;
967        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
968        let body = b"unauthorized native store write";
969        let hash = sha256(body);
970        let hash_hex = to_hex(&hash);
971        let (port, handle) = spawn_test_server(Arc::clone(&store)).await?;
972
973        let response = reqwest::Client::new()
974            .put(format!("http://127.0.0.1:{port}/__iris/store/{hash_hex}"))
975            .body(body.to_vec())
976            .send()
977            .await?;
978
979        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
980        assert!(store.get_blob(&hash)?.is_none());
981        handle.abort();
982        Ok(())
983    }
984
985    #[tokio::test]
986    async fn unauthenticated_cache_tree_root_is_rejected() -> Result<()> {
987        let temp_dir = TempDir::new()?;
988        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
989        let (port, handle) = spawn_test_server(store).await?;
990
991        let response = reqwest::Client::new()
992            .post(format!("http://127.0.0.1:{port}/api/cache-tree-root"))
993            .json(&json!({
994                "npub": "npub1example",
995                "treeName": "video",
996                "hash": "988db3f24dc222715f1c1e1fa5876690d3147122243d72d85fd44283867cd61a",
997                "visibility": "public"
998            }))
999            .send()
1000            .await?;
1001
1002        assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
1003        handle.abort();
1004        Ok(())
1005    }
1006
1007    #[tokio::test]
1008    async fn unauthenticated_upload_batch_is_rejected_before_json_extraction() -> Result<()> {
1009        let temp_dir = TempDir::new()?;
1010        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1011        let (port, handle) = spawn_test_server(store).await?;
1012
1013        let response = reqwest::Client::new()
1014            .post(format!("http://127.0.0.1:{port}/upload/batch"))
1015            .header(reqwest::header::CONTENT_TYPE, "application/json")
1016            .body("not valid json")
1017            .send()
1018            .await?;
1019
1020        assert_eq!(response.status(), reqwest::StatusCode::UNAUTHORIZED);
1021        handle.abort();
1022        Ok(())
1023    }
1024
1025    #[tokio::test]
1026    async fn unauthenticated_single_upload_is_rejected_before_body_limit() -> Result<()> {
1027        let temp_dir = TempDir::new()?;
1028        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1029        let (port, handle) = spawn_test_server(store).await?;
1030        let declared_bytes = blossom::MAX_SINGLE_UPLOAD_BODY_BYTES + 1;
1031        let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port)).await?;
1032        stream
1033            .write_all(
1034                format!(
1035                    "PUT /upload HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nContent-Length: {declared_bytes}\r\nConnection: close\r\n\r\n"
1036                )
1037                .as_bytes(),
1038            )
1039            .await?;
1040
1041        // Send headers only: receiving a response proves authentication ran
1042        // before either buffering the declared body or enforcing its limit.
1043        let mut status_line = String::new();
1044        BufReader::new(stream).read_line(&mut status_line).await?;
1045        assert_eq!(status_line.trim_end(), "HTTP/1.1 401 Unauthorized");
1046        handle.abort();
1047        Ok(())
1048    }
1049
1050    #[tokio::test]
1051    async fn upload_options_preflight_does_not_require_auth() -> Result<()> {
1052        let temp_dir = TempDir::new()?;
1053        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1054        let (port, handle) = spawn_test_server(store).await?;
1055
1056        let response = reqwest::Client::new()
1057            .request(
1058                reqwest::Method::OPTIONS,
1059                format!("http://127.0.0.1:{port}/upload"),
1060            )
1061            .send()
1062            .await?;
1063
1064        assert_eq!(response.status(), reqwest::StatusCode::NO_CONTENT);
1065        handle.abort();
1066        Ok(())
1067    }
1068
1069    #[tokio::test]
1070    async fn virtual_tree_hosts_serve_root_assets_and_spa_fallbacks() -> Result<()> {
1071        clear_virtual_tree_hosts_for_test();
1072
1073        let temp_dir = TempDir::new()?;
1074        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1075        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1076
1077        let (index_cid, _) = tree
1078            .put(b"<!doctype html><title>Virtual host ok</title>")
1079            .await?;
1080        let (favicon_cid, _) = tree.put(b"ico").await?;
1081        let (main_js_cid, _) = tree.put(b"console.log('ok');").await?;
1082        let assets_dir = tree
1083            .put_directory(vec![
1084                DirEntry::from_cid("main.js", &main_js_cid).with_link_type(LinkType::File)
1085            ])
1086            .await?;
1087        let root_cid = tree
1088            .put_directory(vec![
1089                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File),
1090                DirEntry::from_cid("favicon.ico", &favicon_cid).with_link_type(LinkType::File),
1091                DirEntry::from_cid("assets", &assets_dir).with_link_type(LinkType::Dir),
1092            ])
1093            .await?;
1094        let nhash = nhash_encode(&root_cid.hash)?;
1095        let host = "tree-test.htree.localhost";
1096        register_virtual_tree_host(host, &format!("/htree/{nhash}"));
1097
1098        let (port, handle) = spawn_test_server(store).await?;
1099        let base_url = format!("http://127.0.0.1:{port}");
1100        let host_header = format!("{host}:{port}");
1101        let client = reqwest::Client::new();
1102
1103        let root_response = client
1104            .get(format!("{base_url}/"))
1105            .header("Host", &host_header)
1106            .header("Accept", "text/html")
1107            .send()
1108            .await?;
1109        assert_eq!(root_response.status(), reqwest::StatusCode::OK);
1110        assert_eq!(
1111            root_response.bytes().await?.as_ref(),
1112            b"<!doctype html><title>Virtual host ok</title>"
1113        );
1114
1115        let favicon_response = client
1116            .get(format!("{base_url}/favicon.ico"))
1117            .header("Host", &host_header)
1118            .send()
1119            .await?;
1120        assert_eq!(favicon_response.status(), reqwest::StatusCode::OK);
1121        assert_eq!(favicon_response.bytes().await?.as_ref(), b"ico");
1122
1123        let js_response = client
1124            .get(format!("{base_url}/assets/main.js"))
1125            .header("Host", &host_header)
1126            .send()
1127            .await?;
1128        assert_eq!(js_response.status(), reqwest::StatusCode::OK);
1129        assert_eq!(js_response.bytes().await?.as_ref(), b"console.log('ok');");
1130
1131        let profile_response = client
1132            .get(format!("{base_url}/users/npub1example"))
1133            .header("Host", &host_header)
1134            .header("Accept", "text/html")
1135            .send()
1136            .await?;
1137        assert_eq!(profile_response.status(), reqwest::StatusCode::OK);
1138        assert_eq!(
1139            profile_response.bytes().await?.as_ref(),
1140            b"<!doctype html><title>Virtual host ok</title>"
1141        );
1142
1143        handle.abort();
1144        clear_virtual_tree_hosts_for_test();
1145
1146        Ok(())
1147    }
1148
1149    #[tokio::test]
1150    async fn iris_localhost_hosts_serve_immutable_and_named_sites() -> Result<()> {
1151        let temp_dir = TempDir::new()?;
1152        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1153        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1154
1155        let (index_cid, _) = tree
1156            .put(b"<!doctype html><title>Iris localhost ok</title>")
1157            .await?;
1158        let (main_js_cid, _) = tree.put(b"console.log('iris localhost');").await?;
1159        let assets_dir = tree
1160            .put_directory(vec![
1161                DirEntry::from_cid("main.js", &main_js_cid).with_link_type(LinkType::File)
1162            ])
1163            .await?;
1164        let root_cid = tree
1165            .put_directory(vec![
1166                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File),
1167                DirEntry::from_cid("assets", &assets_dir).with_link_type(LinkType::Dir),
1168            ])
1169            .await?;
1170
1171        let nhash = nhash_encode(&root_cid.hash)?;
1172        let npub = Keys::generate().public_key().to_bech32()?;
1173        let site = "audio";
1174        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1175        let port = listener.local_addr()?.port();
1176        let server = HashtreeServer::new(Arc::clone(&store), "127.0.0.1:0".to_string())
1177            .with_public_plaintext_reads(true)
1178            .with_cached_tree_roots(vec![(format!("{npub}/{site}"), root_cid)]);
1179        let handle =
1180            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1181        let client = reqwest::Client::new();
1182        let base_url = format!("http://127.0.0.1:{port}");
1183
1184        for host in [
1185            format!("{nhash}.iris.localhost:{port}"),
1186            format!("{site}.{npub}.iris.localhost:{port}"),
1187        ] {
1188            let root_response = client
1189                .get(format!("{base_url}/"))
1190                .header("Host", &host)
1191                .header("Accept", "text/html")
1192                .send()
1193                .await?;
1194            assert_eq!(root_response.status(), reqwest::StatusCode::OK, "{host}");
1195            assert_eq!(
1196                root_response.bytes().await?.as_ref(),
1197                b"<!doctype html><title>Iris localhost ok</title>"
1198            );
1199
1200            let asset_response = client
1201                .get(format!("{base_url}/assets/main.js"))
1202                .header("Host", &host)
1203                .send()
1204                .await?;
1205            assert_eq!(asset_response.status(), reqwest::StatusCode::OK, "{host}");
1206            assert_eq!(
1207                asset_response.bytes().await?.as_ref(),
1208                b"console.log('iris localhost');"
1209            );
1210        }
1211
1212        handle.abort();
1213        Ok(())
1214    }
1215
1216    #[tokio::test]
1217    async fn named_iris_site_resolves_from_fips_event_provider_without_relays() -> Result<()> {
1218        let temp_dir = TempDir::new()?;
1219        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1220        let tree = HashTree::new(HashTreeConfig::new(store.store_arc()).public());
1221        let (index_cid, _) = tree.put(b"fips event provider site").await?;
1222        let root_cid = tree
1223            .put_directory(vec![
1224                DirEntry::from_cid("index.html", &index_cid).with_link_type(LinkType::File)
1225            ])
1226            .await?;
1227        let owner = Keys::generate();
1228        let site = "radio";
1229        let root_event = EventBuilder::new(Kind::Custom(30064), "")
1230            .tags(vec![
1231                nostr::Tag::identifier(site),
1232                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["hashtree".to_string()]),
1233                nostr::Tag::custom(nostr::TagKind::custom("hash"), vec![to_hex(&root_cid.hash)]),
1234            ])
1235            .sign_with_keys(&owner)?;
1236        let provider = Arc::new(StaticProvider {
1237            event: Some(VerifiedEvent::try_from(root_event)?),
1238            queries: AtomicUsize::new(0),
1239            publishes: AtomicUsize::new(0),
1240            mode: PubsubProviderMode::LocalOnly,
1241        });
1242
1243        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1244        let port = listener.local_addr()?.port();
1245        let server = HashtreeServer::new(Arc::clone(&store), "127.0.0.1:0".to_string())
1246            .with_public_plaintext_reads(true)
1247            .with_nostr_provider(provider.clone());
1248        assert!(server.state.nostr_relay_urls.is_empty());
1249        assert_eq!(
1250            server.state.nostr_provider.as_ref().unwrap().mode(),
1251            PubsubProviderMode::LocalOnly
1252        );
1253        let handle =
1254            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1255
1256        let npub = owner.public_key().to_bech32()?;
1257        let response = reqwest::Client::new()
1258            .get(format!("http://127.0.0.1:{port}/"))
1259            .header("Host", format!("{site}.{npub}.iris.localhost"))
1260            .header("Accept", "text/html")
1261            .send()
1262            .await?;
1263
1264        assert_eq!(response.status(), reqwest::StatusCode::OK);
1265        assert_eq!(
1266            response.bytes().await?.as_ref(),
1267            b"fips event provider site"
1268        );
1269        assert_eq!(provider.queries.load(Ordering::Relaxed), 1);
1270        handle.abort();
1271        Ok(())
1272    }
1273
1274    #[tokio::test]
1275    async fn loopback_event_publication_uses_configured_provider() -> Result<()> {
1276        let temp_dir = TempDir::new()?;
1277        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1278        let keys = Keys::generate();
1279        let event = EventBuilder::new(Kind::TextNote, "provider test").sign_with_keys(&keys)?;
1280        let provider = Arc::new(StaticProvider {
1281            event: Some(VerifiedEvent::try_from(event.clone())?),
1282            queries: AtomicUsize::new(0),
1283            publishes: AtomicUsize::new(0),
1284            mode: PubsubProviderMode::LocalOnly,
1285        });
1286        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1287        let port = listener.local_addr()?.port();
1288        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
1289            .with_nostr_provider(provider.clone());
1290        assert!(server.state.nostr_relay_urls.is_empty());
1291        let handle =
1292            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1293
1294        let response = reqwest::Client::new()
1295            .post(format!("http://127.0.0.1:{port}/api/nostr/events"))
1296            .json(&event)
1297            .send()
1298            .await?;
1299
1300        assert_eq!(response.status(), reqwest::StatusCode::ACCEPTED);
1301        assert_eq!(provider.publishes.load(Ordering::Relaxed), 1);
1302        handle.abort();
1303        Ok(())
1304    }
1305
1306    #[tokio::test]
1307    async fn loopback_root_publication_is_immediately_resolvable_without_provider_replay(
1308    ) -> Result<()> {
1309        let temp_dir = TempDir::new()?;
1310        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1311        let keys = Keys::generate();
1312        let tree_name = "webvm-e2e";
1313        let root_hash = "11".repeat(32);
1314        let encryption_key = "22".repeat(32);
1315        let event = EventBuilder::new(Kind::Custom(30064), &root_hash)
1316            .tags(vec![
1317                nostr::Tag::identifier(tree_name),
1318                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["hashtree".to_string()]),
1319                nostr::Tag::custom(nostr::TagKind::custom("l"), vec!["git".to_string()]),
1320                nostr::Tag::custom(nostr::TagKind::custom("hash"), vec![root_hash.clone()]),
1321                nostr::Tag::custom(nostr::TagKind::custom("key"), vec![encryption_key.clone()]),
1322            ])
1323            .sign_with_keys(&keys)?;
1324        let provider = Arc::new(StaticProvider {
1325            event: None,
1326            queries: AtomicUsize::new(0),
1327            publishes: AtomicUsize::new(0),
1328            mode: PubsubProviderMode::LocalOnly,
1329        });
1330        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
1331        let port = listener.local_addr()?.port();
1332        let server = HashtreeServer::new(store, "127.0.0.1:0".to_string())
1333            .with_nostr_provider(provider.clone());
1334        assert!(server.state.nostr_relay_urls.is_empty());
1335        let handle =
1336            tokio::spawn(async move { server.run_with_listener(listener).await.map(|_| ()) });
1337        let client = reqwest::Client::new();
1338
1339        let publish = client
1340            .post(format!("http://127.0.0.1:{port}/api/nostr/events"))
1341            .json(&event)
1342            .send()
1343            .await?;
1344        assert_eq!(publish.status(), reqwest::StatusCode::ACCEPTED);
1345
1346        let npub = keys.public_key().to_bech32()?;
1347        let resolved = client
1348            .get(format!(
1349                "http://127.0.0.1:{port}/api/nostr/resolve/{npub}/{tree_name}?refresh=1"
1350            ))
1351            .send()
1352            .await?
1353            .json::<serde_json::Value>()
1354            .await?;
1355
1356        assert_eq!(resolved["hash"], root_hash);
1357        assert_eq!(resolved["key_tag"], encryption_key);
1358        assert_eq!(resolved["source"], "local-relay");
1359        assert_eq!(provider.publishes.load(Ordering::Relaxed), 1);
1360        assert_eq!(provider.queries.load(Ordering::Relaxed), 1);
1361        handle.abort();
1362        Ok(())
1363    }
1364
1365    #[tokio::test]
1366    async fn nostr_profile_route_returns_latest_metadata_event() -> Result<()> {
1367        let temp_dir = TempDir::new()?;
1368        let store = Arc::new(HashtreeStore::new(temp_dir.path().join("db"))?);
1369        let graph_store = {
1370            let _guard = crate::socialgraph::test_lock().await;
1371            crate::socialgraph::open_test_social_graph_store_with_mapsize(
1372                &temp_dir.path().join("relay-db"),
1373                Some(128 * 1024 * 1024),
1374            )?
1375        };
1376        let backend: Arc<dyn crate::socialgraph::SocialGraphBackend> = graph_store;
1377        let relay = Arc::new(NostrRelay::new(
1378            backend,
1379            temp_dir.path().to_path_buf(),
1380            HashSet::new(),
1381            None,
1382            NostrRelayConfig {
1383                spambox_db_max_bytes: 0,
1384                ..Default::default()
1385            },
1386        )?);
1387
1388        let author = Keys::generate();
1389        let older = EventBuilder::new(
1390            Kind::Metadata,
1391            json!({ "name": "older", "about": "before" }).to_string(),
1392        )
1393        .custom_created_at(Timestamp::from_secs(10))
1394        .sign_with_keys(&author)?;
1395        let newer = EventBuilder::new(
1396            Kind::Metadata,
1397            json!({ "name": "newer", "about": "after" }).to_string(),
1398        )
1399        .custom_created_at(Timestamp::from_secs(20))
1400        .sign_with_keys(&author)?;
1401
1402        relay.ingest_trusted_event(older).await?;
1403        relay.ingest_trusted_event(newer.clone()).await?;
1404
1405        let (port, handle) = spawn_test_server_with_nostr_relay(store, relay).await?;
1406        let response = reqwest::get(format!(
1407            "http://127.0.0.1:{port}/api/nostr/profile/{}",
1408            author.public_key().to_hex()
1409        ))
1410        .await?;
1411
1412        assert_eq!(response.status(), reqwest::StatusCode::OK);
1413        let payload: serde_json::Value = response.json().await?;
1414        assert_eq!(payload["profile"]["name"].as_str(), Some("newer"),);
1415        assert_eq!(payload["profile"]["about"].as_str(), Some("after"));
1416        assert_eq!(payload["created_at"].as_u64(), Some(20));
1417        let expected_event_id = newer.id.to_hex();
1418        assert_eq!(
1419            payload["event_id"].as_str(),
1420            Some(expected_event_id.as_str())
1421        );
1422
1423        handle.abort();
1424        Ok(())
1425    }
1426}
hashtree_cli/server/mod.rs

hashtree_cli/server/
mod.rs
