1use std::io::{BufRead, BufReader, Write};
6use std::os::unix::fs::FileTypeExt;
7use std::os::unix::net::UnixListener;
8use std::sync::atomic::{AtomicBool, Ordering};
9use std::sync::{Arc, Mutex};
10use std::time::Duration;
11
12use serde_json::json;
13use tiny_http::{Header, Response, Server};
14
15use crate::stats::WebMetrics;
16
17const PORT: u16 = 50005;
18const UNIX_SOCKET_PATH: &str = "/tmp/scx_flow.sock";
19const POLL_INTERVAL: Duration = Duration::from_millis(200);
20
21struct WebState {
22 metrics: WebMetrics,
23}
24
25fn unix_handle_client(
26 mut stream: std::os::unix::net::UnixStream,
27 state: &Arc<Mutex<WebState>>,
28 html: &str,
29) {
30 let clone = match stream.try_clone() {
31 Ok(c) => c,
32 Err(_) => return,
33 };
34 let mut reader = BufReader::new(clone);
35 let mut request_line = String::new();
36 if reader.read_line(&mut request_line).is_err() {
37 return;
38 }
39
40 let parts: Vec<&str> = request_line.split_whitespace().collect();
41 if parts.len() < 2 {
42 return;
43 }
44 let path = parts[1];
45
46 let metrics = {
47 let st = match state.lock() {
48 Ok(s) => s,
49 Err(_) => return,
50 };
51 st.metrics.clone()
52 };
53
54 let (body, content_type) = match path {
55 "/" => (html.as_bytes().to_vec(), "text/html; charset=utf-8"),
56 "/api/stats" => {
57 let stats = serde_json::to_value(&metrics.stats).unwrap_or_default();
58 let per_cpu = serde_json::to_value(&metrics.per_cpu).unwrap_or_default();
59 let merged = json!({
60 "stats": stats,
61 "per_cpu": per_cpu,
62 "carriage_filling_count": metrics.carriage_filling_count,
63 });
64 let j = serde_json::to_string(&merged).unwrap_or_else(|_| "{}".into());
65 (j.into_bytes(), "application/json")
66 }
67 _ => {
68 let _ = write!(
69 stream,
70 "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n"
71 );
72 return;
73 }
74 };
75 let len = body.len();
76 let _ = write!(
77 stream,
78 "HTTP/1.1 200 OK\r\nContent-Type: {}\r\nContent-Length: {}\r\nCache-Control: no-store\r\nConnection: close\r\n\r\n",
79 content_type, len
80 );
81 let _ = stream.write_all(&body);
82 let _ = stream.flush();
83}
84
85pub fn start(metrics_rx: crossbeam::channel::Receiver<WebMetrics>, shutdown: Arc<AtomicBool>) {
86 log::info!("Web UI thread started");
87
88 let html = include_str!("../ui/index.html").to_string();
89 let state = Arc::new(Mutex::new(WebState {
90 metrics: WebMetrics::default(),
91 }));
92
93 let state_clone = state.clone();
94 let shutdown_clone = shutdown.clone();
95 std::thread::spawn(move || {
96 while !shutdown_clone.load(Ordering::Relaxed) {
97 match metrics_rx.recv_timeout(POLL_INTERVAL) {
98 Ok(m) => {
99 if let Ok(mut st) = state_clone.lock() {
100 st.metrics = m;
101 }
102 }
103 Err(crossbeam::channel::RecvTimeoutError::Timeout) => {}
104 Err(_) => break,
105 }
106 }
107 });
108
109 let html_for_unix = html.to_owned();
110 let mut server: Option<tiny_http::Server> = None;
111 let mut tcp_addr = String::new();
112
113 if let Ok(s) = Server::http(&format!("[::1]:{}", PORT)) {
114 tcp_addr = format!("[::1]:{}", PORT);
115 server = Some(s);
116 }
117
118 if server.is_none() {
119 if let Ok(s) = Server::http(&format!("127.0.0.1:{}", PORT)) {
120 tcp_addr = format!("127.0.0.1:{}", PORT);
121 server = Some(s);
122 }
123 }
124
125 if let Some(server) = server {
126 log::info!(
127 "Web UI listening on http://{}/ — disable with --no-webui",
128 tcp_addr
129 );
130
131 let no_cache = Header::from_bytes("Cache-Control", "no-store").unwrap();
132 let html_type = Header::from_bytes("Content-Type", "text/html; charset=utf-8").unwrap();
133 let json_type = Header::from_bytes("Content-Type", "application/json").unwrap();
134
135 while !shutdown.load(Ordering::Relaxed) {
136 if let Ok(Some(request)) = server.recv_timeout(Duration::from_millis(200)) {
137 let metrics = {
138 let st = match state.lock() {
139 Ok(s) => s,
140 Err(_) => continue,
141 };
142 st.metrics.clone()
143 };
144 match request.url() {
145 "/" => {
146 let resp = Response::from_string(&html)
147 .with_header(html_type.clone())
148 .with_header(no_cache.clone());
149 let _ = request.respond(resp);
150 }
151 "/api/stats" => {
152 let stats = serde_json::to_value(&metrics.stats).unwrap_or_default();
153 let per_cpu = serde_json::to_value(&metrics.per_cpu).unwrap_or_default();
154 let merged = json!({
155 "stats": stats,
156 "per_cpu": per_cpu,
157 "carriage_filling_count": metrics.carriage_filling_count,
158 });
159 let json = serde_json::to_string(&merged).unwrap_or_else(|_| "{}".into());
160 let resp = Response::from_string(json)
161 .with_header(json_type.clone())
162 .with_header(no_cache.clone());
163 let _ = request.respond(resp);
164 }
165 _ => {
166 let _ = request.respond(Response::empty(404));
167 }
168 }
169 }
170 }
171 } else {
172 log::warn!(
173 "Web UI: TCP blocked (spawned by scx_loader?), falling back to {}",
174 UNIX_SOCKET_PATH
175 );
176 if let Ok(meta) = std::fs::symlink_metadata(UNIX_SOCKET_PATH) {
177 if meta.file_type().is_socket() {
178 let _ = std::fs::remove_file(UNIX_SOCKET_PATH);
179 }
180 }
181
182 let listener = match UnixListener::bind(UNIX_SOCKET_PATH) {
183 Ok(l) => l,
184 Err(e) => {
185 log::warn!("Web UI: Unix socket bind failed: {}", e);
186 log::warn!("Web UI disabled. Use --no-webui to silence.");
187 return;
188 }
189 };
190
191 log::info!(
192 "Web UI listening on unix:{} — access via: sudo socat TCP-LISTEN:{} UNIX-CONNECT:{}",
193 UNIX_SOCKET_PATH,
194 PORT,
195 UNIX_SOCKET_PATH
196 );
197 log::info!("Or run: sudo /usr/bin/scx_flow for direct TCP access");
198
199 listener.set_nonblocking(true).expect("set_nonblocking");
200 while !shutdown.load(Ordering::Relaxed) {
201 match listener.accept() {
202 Ok((stream, _)) => {
203 let state = state.clone();
204 let html = html_for_unix.clone();
205 std::thread::spawn(move || unix_handle_client(stream, &state, &html));
206 }
207 Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
208 std::thread::sleep(Duration::from_millis(100));
209 }
210 Err(_) => break,
211 }
212 }
213 }
214
215 log::info!("Web UI stopped");
216}