logicaffeine_cli/browser/
server.rs

1//! A loopback HTTP server for one page bundle, and the relay its pages talk through.
2//!
3//! A module script must come from an HTTP(S) origin — Chrome refuses `type="module"` from
4//! `file://` — so a built page is served the way it ships: `GET` for a file in the bundle, `404`
5//! for anything else, one response per connection. It binds `127.0.0.1:0`, so parallel tests each
6//! get a port of their own and nothing listens beyond the machine.
7//!
8//! `/_relay` is the relay a page's `Publish` and `When a message arrives` reach: a WebSocket that
9//! speaks `logicaffeine_system::relay_proto::RelayFrame` exactly as the system's own relay does —
10//! `Subscribe` answered with `SubAck`, each `Publish` delivered as an `Event` to every OTHER
11//! connection subscribed to its topic. Pages a test opens on one [`RelayHub`] reach each other.
12
13use std::collections::{BTreeMap, HashMap};
14use std::io::{BufRead, BufReader, Read, Write};
15use std::net::{Shutdown, TcpListener, TcpStream};
16use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
17use std::sync::{Arc, Mutex};
18use std::time::Duration;
19
20use base64::Engine as _;
21use logicaffeine_system::relay_proto::RelayFrame;
22
23/// A relay the pages of one test share: every page opened with it publishes into it and hears
24/// what the others publish.
25#[derive(Clone, Default)]
26pub struct RelayHub {
27    state: Arc<Mutex<HubState>>,
28}
29
30impl std::fmt::Debug for RelayHub {
31    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32        let st = self.state.lock().unwrap();
33        f.debug_struct("RelayHub")
34            .field("connections", &st.conns.len())
35            .field("topics", &st.subs.keys().collect::<Vec<_>>())
36            .finish()
37    }
38}
39
40#[derive(Default)]
41struct HubState {
42    /// Each open connection's writer, by the name the page's recorder gave it.
43    conns: HashMap<String, Arc<Mutex<TcpStream>>>,
44    subs: BTreeMap<String, Vec<String>>,
45    /// Per connection: how many messages the hub has read from it, and written to it.
46    counts: HashMap<String, (u64, u64)>,
47    /// Every frame the hub read from a page, in order.
48    received: Vec<RelayFrame>,
49}
50
51impl RelayHub {
52    /// Deliver `data` on `topic` to every page subscribed to it, as another replica's publish
53    /// reaches them.
54    pub fn deliver(&self, topic: &str, data: &[u8]) {
55        let mut st = self.state.lock().unwrap();
56        let to: Vec<String> = st.subs.get(topic).cloned().unwrap_or_default();
57        let frame = RelayFrame::Event {
58            topic: topic.to_string(),
59            data: data.to_vec(),
60        };
61        for conn in &to {
62            st.write(conn, &frame);
63        }
64    }
65
66    /// How many connections are subscribed to `topic`.
67    pub fn subscribers(&self, topic: &str) -> usize {
68        self.state
69            .lock()
70            .unwrap()
71            .subs
72            .get(topic)
73            .map_or(0, Vec::len)
74    }
75
76    /// Every frame the relay read from a page, in the order it read them.
77    pub fn received(&self) -> Vec<RelayFrame> {
78        self.state.lock().unwrap().received.clone()
79    }
80
81    /// How many messages the hub has read from the connection `token` names, and written to it.
82    pub(crate) fn counts(&self, token: &str) -> (u64, u64) {
83        self.state
84            .lock()
85            .unwrap()
86            .counts
87            .get(token)
88            .copied()
89            .unwrap_or((0, 0))
90    }
91
92    fn join(&self, token: &str, writer: TcpStream) {
93        self.state
94            .lock()
95            .unwrap()
96            .conns
97            .insert(token.to_string(), Arc::new(Mutex::new(writer)));
98    }
99
100    /// One whole message read from `conn`: a frame the hub acts on, or bytes no `RelayFrame`
101    /// decodes, which it counts and drops as the system's relay does.
102    fn receive(&self, conn: &str, message: &[u8]) {
103        let mut st = self.state.lock().unwrap();
104        st.counts.entry(conn.to_string()).or_default().0 += 1;
105        let frame = RelayFrame::from_bytes(message);
106        st.received.extend(frame.clone());
107        match frame {
108            Some(RelayFrame::Subscribe { topic }) => {
109                let subs = st.subs.entry(topic.clone()).or_default();
110                if !subs.iter().any(|c| c == conn) {
111                    subs.push(conn.to_string());
112                }
113                st.write(conn, &RelayFrame::SubAck { topic });
114            }
115            Some(RelayFrame::Unsubscribe { topic }) => {
116                if let Some(subs) = st.subs.get_mut(&topic) {
117                    subs.retain(|c| c != conn);
118                }
119            }
120            Some(RelayFrame::Publish { topic, data }) => {
121                let to: Vec<String> = st
122                    .subs
123                    .get(&topic)
124                    .into_iter()
125                    .flatten()
126                    .filter(|c| *c != conn)
127                    .cloned()
128                    .collect();
129                let event = RelayFrame::Event { topic, data };
130                for c in &to {
131                    st.write(c, &event);
132                }
133            }
134            Some(RelayFrame::SubAck { .. } | RelayFrame::Event { .. }) | None => {}
135        }
136    }
137
138    fn leave(&self, conn: &str) {
139        let mut st = self.state.lock().unwrap();
140        st.conns.remove(conn);
141        for subs in st.subs.values_mut() {
142            subs.retain(|c| c != conn);
143        }
144    }
145}
146
147impl HubState {
148    /// Write `frame` to `conn` as one binary WebSocket message.
149    fn write(&mut self, conn: &str, frame: &RelayFrame) {
150        let Some(writer) = self.conns.get(conn).cloned() else {
151            return;
152        };
153        let payload = frame.to_bytes();
154        let mut out = vec![0x82];
155        match payload.len() {
156            n if n < 126 => out.push(n as u8),
157            n if n <= 0xFFFF => {
158                out.push(126);
159                out.extend_from_slice(&(n as u16).to_be_bytes());
160            }
161            n => {
162                out.push(127);
163                out.extend_from_slice(&(n as u64).to_be_bytes());
164            }
165        }
166        out.extend_from_slice(&payload);
167        if writer.lock().unwrap().write_all(&out).is_ok() {
168            self.counts.entry(conn.to_string()).or_default().1 += 1;
169        }
170    }
171}
172
173pub(crate) struct StaticServer {
174    origin: String,
175    stop: Arc<AtomicBool>,
176    port: u16,
177    hub: RelayHub,
178}
179
180impl StaticServer {
181    /// Serve `files` (path → content type and bytes, paths beginning with `/`), and `/_relay` on
182    /// `hub`, until dropped.
183    pub(crate) fn start(
184        files: BTreeMap<String, (String, Vec<u8>)>,
185        hub: RelayHub,
186    ) -> std::io::Result<StaticServer> {
187        let listener = TcpListener::bind("127.0.0.1:0")?;
188        let port = listener.local_addr()?.port();
189        let stop = Arc::new(AtomicBool::new(false));
190        let files = Arc::new(files);
191        let halt = Arc::clone(&stop);
192        let relay = hub.clone();
193        std::thread::Builder::new()
194            .name("page-server".into())
195            .spawn(move || {
196                for conn in listener.incoming() {
197                    if halt.load(Ordering::SeqCst) {
198                        break;
199                    }
200                    let Ok(conn) = conn else { continue };
201                    let files = Arc::clone(&files);
202                    let relay = relay.clone();
203                    let _ = std::thread::Builder::new()
204                        .name("page-server-conn".into())
205                        .spawn(move || respond(conn, &files, &relay));
206                }
207            })?;
208        Ok(StaticServer {
209            origin: format!("http://127.0.0.1:{port}"),
210            stop,
211            port,
212            hub,
213        })
214    }
215
216    /// `http://127.0.0.1:<port>`.
217    pub(crate) fn origin(&self) -> &str {
218        &self.origin
219    }
220
221    /// The relay `/_relay` reaches.
222    pub(crate) fn relay(&self) -> &RelayHub {
223        &self.hub
224    }
225}
226
227impl Drop for StaticServer {
228    fn drop(&mut self) {
229        self.stop.store(true, Ordering::SeqCst);
230        // `incoming()` blocks in `accept`; one connection wakes it to see the flag.
231        let _ = TcpStream::connect(("127.0.0.1", self.port));
232    }
233}
234
235fn respond(conn: TcpStream, files: &BTreeMap<String, (String, Vec<u8>)>, hub: &RelayHub) {
236    let _ = conn.set_read_timeout(Some(Duration::from_secs(10)));
237    let mut reader = BufReader::new(&conn);
238    let mut request_line = String::new();
239    if reader.read_line(&mut request_line).is_err() {
240        return;
241    }
242    let mut websocket_key = None;
243    loop {
244        let mut header = String::new();
245        match reader.read_line(&mut header) {
246            Ok(0) => break,
247            Ok(_) if header == "\r\n" || header == "\n" => break,
248            Ok(_) => {
249                if let Some((name, value)) = header.split_once(':') {
250                    if name.trim().eq_ignore_ascii_case("sec-websocket-key") {
251                        websocket_key = Some(value.trim().to_string());
252                    }
253                }
254            }
255            Err(_) => return,
256        }
257    }
258    let mut parts = request_line.split_whitespace();
259    let method = parts.next().unwrap_or("");
260    let target = parts.next().unwrap_or("/");
261    let path = target.split(['?', '#']).next().unwrap_or("/");
262    if path == "/_relay" {
263        if let Some(key) = websocket_key {
264            drop(reader);
265            let token = target
266                .split_once('?')
267                .and_then(|(_, q)| q.split('&').find_map(|kv| kv.strip_prefix("harness=")))
268                .map(str::to_string)
269                .unwrap_or_else(|| format!("unnamed-{}", UNNAMED.fetch_add(1, Ordering::SeqCst)));
270            relay(conn, &key, &token, hub);
271        }
272        return;
273    }
274    let found = if method == "GET" || method == "HEAD" {
275        files.get(path)
276    } else {
277        None
278    };
279    let mut out = &conn;
280    let head = match found {
281        Some((content_type, bytes)) => format!(
282            "HTTP/1.1 200 OK\r\ncontent-type: {content_type}\r\ncontent-length: {}\r\ncache-control: no-store\r\nconnection: close\r\n\r\n",
283            bytes.len()
284        ),
285        None => "HTTP/1.1 404 Not Found\r\ncontent-type: text/plain\r\ncontent-length: 9\r\nconnection: close\r\n\r\nnot found"
286            .to_string(),
287    };
288    let _ = out.write_all(head.as_bytes());
289    if let (Some((_, bytes)), true) = (found, method == "GET") {
290        let _ = out.write_all(bytes);
291    }
292    let _ = out.flush();
293    let _ = conn.shutdown(Shutdown::Both);
294}
295
296/// A name for a relay connection no recorder named.
297static UNNAMED: AtomicU64 = AtomicU64::new(1);
298
299/// One page's relay connection, named `token`: the WebSocket handshake, then every message it sends
300/// handed to the hub, until it closes.
301fn relay(mut conn: TcpStream, key: &str, token: &str, hub: &RelayHub) {
302    let accept = {
303        let digest = logicaffeine_base::hash::sha1(
304            format!("{key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11").as_bytes(),
305        );
306        base64::engine::general_purpose::STANDARD.encode(digest)
307    };
308    let handshake = format!(
309        "HTTP/1.1 101 Switching Protocols\r\nupgrade: websocket\r\nconnection: Upgrade\r\nsec-websocket-accept: {accept}\r\n\r\n"
310    );
311    if conn.write_all(handshake.as_bytes()).is_err() {
312        return;
313    }
314    let _ = conn.set_read_timeout(None);
315    let Ok(writer) = conn.try_clone() else { return };
316    hub.join(token, writer);
317    let mut message = Vec::new();
318    loop {
319        let Some((fin, opcode, payload)) = read_frame(&mut conn) else {
320            break;
321        };
322        match opcode {
323            0x8 => break,
324            0x9 => {
325                let mut pong = vec![0x8A, payload.len() as u8];
326                pong.extend_from_slice(&payload);
327                let _ = conn.write_all(&pong);
328            }
329            0x0 | 0x1 | 0x2 => {
330                message.extend_from_slice(&payload);
331                if fin {
332                    hub.receive(token, &message);
333                    message.clear();
334                }
335            }
336            _ => {}
337        }
338    }
339    hub.leave(token);
340    let _ = conn.shutdown(Shutdown::Both);
341}
342
343/// One WebSocket frame from a client: `(fin, opcode, unmasked payload)`, or `None` when the
344/// connection ends.
345fn read_frame(conn: &mut TcpStream) -> Option<(bool, u8, Vec<u8>)> {
346    let mut head = [0u8; 2];
347    conn.read_exact(&mut head).ok()?;
348    let fin = head[0] & 0x80 != 0;
349    let opcode = head[0] & 0x0F;
350    let masked = head[1] & 0x80 != 0;
351    let len = match head[1] & 0x7F {
352        126 => {
353            let mut b = [0u8; 2];
354            conn.read_exact(&mut b).ok()?;
355            u16::from_be_bytes(b) as usize
356        }
357        127 => {
358            let mut b = [0u8; 8];
359            conn.read_exact(&mut b).ok()?;
360            u64::from_be_bytes(b) as usize
361        }
362        n => n as usize,
363    };
364    let mut mask = [0u8; 4];
365    if masked {
366        conn.read_exact(&mut mask).ok()?;
367    }
368    let mut payload = vec![0u8; len];
369    conn.read_exact(&mut payload).ok()?;
370    if masked {
371        for (i, b) in payload.iter_mut().enumerate() {
372            *b ^= mask[i % 4];
373        }
374    }
375    Some((fin, opcode, payload))
376}
logicaffeine_cli/browser/server.rs

logicaffeine_cli/browser/
server.rs
