diff --git a/crates/zap-stream/Cargo.toml b/crates/zap-stream/Cargo.toml index 217995c..e36ba39 100644 --- a/crates/zap-stream/Cargo.toml +++ b/crates/zap-stream/Cargo.toml @@ -42,4 +42,5 @@ sha2.workspace = true pretty_env_logger = "0.5.0" clap = { version = "4.5.16", features = ["derive"] } futures-util = "0.3.31" -matchit = "0.8.4" \ No newline at end of file +matchit = "0.8.4" +mustache = "0.9.0" \ No newline at end of file diff --git a/crates/zap-stream/index.html b/crates/zap-stream/index.html index b952b71..8881675 100644 --- a/crates/zap-stream/index.html +++ b/crates/zap-stream/index.html @@ -9,9 +9,122 @@ color: white; font-family: monospace; } + .container { + padding: 20px; + max-width: 1200px; + margin: 0 auto; + } + .stream-list { + margin: 20px 0; + } + .stream-item { + background: #333; + margin: 10px 0; + padding: 15px; + border-radius: 5px; + } + .stream-title { + font-size: 18px; + font-weight: bold; + margin-bottom: 5px; + } + .stream-link { + color: #00ff00; + text-decoration: none; + } + .stream-link:hover { + text-decoration: underline; + } + .video-player { + margin: 20px 0; + max-width: 800px; + } + video { + width: 100%; + max-width: 800px; + background: #000; + } + .no-streams { + color: #999; + font-style: italic; + } + .player-section { + margin-top: 30px; + border-top: 1px solid #555; + padding-top: 20px; + } + -

Welcome to %%PUBLIC_URL%%

+
+

Welcome to {{public_url}}

+ +

Active Streams

+ {{#has_streams}} +
+ {{#streams}} +
+
{{title}}
+ {{#summary}}
{{summary}}
{{/summary}} +
+ 📺 {{live_url}} + {{#viewer_count}}👥 {{viewer_count}} viewers{{/viewer_count}} +
+ +
+ {{/streams}} +
+ {{/has_streams}} + {{^has_streams}} +
No active streams
+ {{/has_streams}} + +
+

Stream Player

+
+ +
+
+ + +
+
+
+ + \ No newline at end of file diff --git a/crates/zap-stream/src/api.rs b/crates/zap-stream/src/api.rs index 2039295..2c36838 100644 --- a/crates/zap-stream/src/api.rs +++ b/crates/zap-stream/src/api.rs @@ -658,6 +658,16 @@ impl Api { pub fn get_viewer_count(&self, stream_id: &str) -> usize { self.overseer.viewer_tracker().get_viewer_count(stream_id) } + + /// Get active streams from database + pub async fn get_active_streams(&self) -> Result> { + self.db.list_live_streams().await + } + + /// Get the public URL from settings + pub fn get_public_url(&self) -> String { + self.settings.public_url.clone() + } } #[derive(Deserialize, Serialize)] diff --git a/crates/zap-stream/src/http.rs b/crates/zap-stream/src/http.rs index 9a8c2c6..f471fb4 100644 --- a/crates/zap-stream/src/http.rs +++ b/crates/zap-stream/src/http.rs @@ -11,29 +11,129 @@ use hyper::service::Service; use hyper::{Method, Request, Response}; use log::error; use nostr_sdk::{serde_json, Alphabet, Event, Kind, PublicKey, SingleLetterTag, TagKind}; +use serde::{Serialize, Deserialize}; use std::future::Future; use std::path::PathBuf; use std::pin::Pin; -use tokio::fs::File; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::fs::File; +use tokio::sync::RwLock; use tokio_util::io::ReaderStream; use zap_stream_core::viewer::ViewerTracker; +#[derive(Serialize)] +struct StreamData { + id: String, + title: String, + #[serde(skip_serializing_if = "Option::is_none")] + summary: Option, + live_url: String, + #[serde(skip_serializing_if = "Option::is_none")] + viewer_count: Option, +} + +#[derive(Serialize)] +struct IndexTemplateData { + public_url: String, + has_streams: bool, + #[serde(skip_serializing_if = "Vec::is_empty")] + streams: Vec, +} + +struct CachedStreams { + data: IndexTemplateData, + cached_at: Instant, +} + +pub type StreamCache = Arc>>; + #[derive(Clone)] pub struct HttpServer { - index: String, + index_template: String, files_dir: PathBuf, api: Api, + stream_cache: StreamCache, } impl HttpServer { - pub fn new(index: String, files_dir: PathBuf, api: Api) -> Self { + pub fn new(index_template: String, files_dir: PathBuf, api: Api, stream_cache: StreamCache) -> Self { Self { - index, + index_template, files_dir, api, + stream_cache, } } + async fn get_cached_or_fetch_streams(&self) -> Result { + Self::get_cached_or_fetch_streams_static(&self.stream_cache, &self.api).await + } + + async fn get_cached_or_fetch_streams_static(stream_cache: &StreamCache, api: &Api) -> Result { + const CACHE_DURATION: Duration = Duration::from_secs(60); // 1 minute + + // Check if we have valid cached data + { + let cache = stream_cache.read().await; + if let Some(ref cached) = *cache { + if cached.cached_at.elapsed() < CACHE_DURATION { + return Ok(cached.data.clone()); + } + } + } + + // Cache is expired or missing, fetch new data + let active_streams = api.get_active_streams().await?; + let public_url = api.get_public_url(); + + let template_data = if !active_streams.is_empty() { + let streams: Vec = active_streams + .into_iter() + .map(|stream| { + let viewer_count = api.get_viewer_count(&stream.id); + StreamData { + id: stream.id.clone(), + title: stream.title.unwrap_or_else(|| format!("Stream {}", &stream.id[..8])), + summary: stream.summary, + live_url: format!("/{}/live.m3u8", stream.id), + viewer_count: if viewer_count > 0 { Some(viewer_count) } else { None }, + } + }) + .collect(); + + IndexTemplateData { + public_url, + has_streams: true, + streams, + } + } else { + IndexTemplateData { + public_url, + has_streams: false, + streams: Vec::new(), + } + }; + + // Update cache + { + let mut cache = stream_cache.write().await; + *cache = Some(CachedStreams { + data: template_data.clone(), + cached_at: Instant::now(), + }); + } + + Ok(template_data) + } + + async fn render_index(&self) -> Result { + let template_data = self.get_cached_or_fetch_streams().await?; + let template = mustache::compile_str(&self.index_template)?; + let rendered = template.render_to_string(&template_data)?; + Ok(rendered) + } + async fn handle_hls_playlist( api: &Api, req: &Request, @@ -162,16 +262,52 @@ impl Service> for HttpServer { if req.method() == Method::GET && req.uri().path() == "/" || req.uri().path() == "/index.html" { - let index = self.index.clone(); + let stream_cache = self.stream_cache.clone(); + let api = self.api.clone(); + + // Compile template outside async move for better performance + let template = match mustache::compile_str(&self.index_template) { + Ok(t) => t, + Err(e) => { + error!("Failed to compile template: {}", e); + return Box::pin(async move { + Ok(Response::builder() + .status(500) + .body(BoxBody::default()).unwrap()) + }); + } + }; + return Box::pin(async move { - Ok(Response::builder() - .header("content-type", "text/html") - .header("server", "zap-stream-core") - .body( - Full::new(Bytes::from(index)) - .map_err(|e| match e {}) - .boxed(), - )?) + // Use the existing method to get cached template data + let template_data = Self::get_cached_or_fetch_streams_static(&stream_cache, &api).await; + + match template_data { + Ok(data) => { + match template.render_to_string(&data) { + Ok(index_html) => Ok(Response::builder() + .header("content-type", "text/html") + .header("server", "zap-stream-core") + .body( + Full::new(Bytes::from(index_html)) + .map_err(|e| match e {}) + .boxed(), + )?), + Err(e) => { + error!("Failed to render template: {}", e); + Ok(Response::builder() + .status(500) + .body(BoxBody::default())?) + } + } + } + Err(e) => { + error!("Failed to fetch template data: {}", e); + Ok(Response::builder() + .status(500) + .body(BoxBody::default())?) + } + } }); } diff --git a/crates/zap-stream/src/main.rs b/crates/zap-stream/src/main.rs index 2c3921b..ebbe5e6 100644 --- a/crates/zap-stream/src/main.rs +++ b/crates/zap-stream/src/main.rs @@ -12,6 +12,7 @@ use std::str::FromStr; use std::sync::Arc; use std::time::Duration; use tokio::net::TcpListener; +use tokio::sync::RwLock; use tokio::task::JoinHandle; use tokio::time::sleep; use url::Url; @@ -23,7 +24,7 @@ use zap_stream_core::ingress::srt; use zap_stream_core::ingress::test; use crate::api::Api; -use crate::http::HttpServer; +use crate::http::{HttpServer, StreamCache}; use crate::monitor::BackgroundMonitor; use crate::overseer::ZapStreamOverseer; use crate::settings::Settings; @@ -69,11 +70,13 @@ async fn main() -> Result<()> { } let http_addr: SocketAddr = settings.listen_http.parse()?; - let index_html = include_str!("../index.html").replace("%%PUBLIC_URL%%", &settings.public_url); + let index_template = include_str!("../index.html"); let api = Api::new(overseer.clone(), settings.clone()); + // Create shared stream cache + let stream_cache: StreamCache = Arc::new(RwLock::new(None)); // HTTP server - let server = HttpServer::new(index_html, PathBuf::from(settings.output_dir), api); + let server = HttpServer::new(index_template.to_string(), PathBuf::from(settings.output_dir), api, stream_cache); tasks.push(tokio::spawn(async move { let listener = TcpListener::bind(&http_addr).await?;