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

hashtree_cli/server/
mod.rs
