cairnlang-core 0.5.0

//! The in-process live hub (design.md §10, slice 4a/4c).
//!
//! v1 model: **broadcast-refresh**. Cairn is stateless TEA — app state
//! lives in SQLite (slice 1), so a connection is just the request that
//! produced its page. When a request performs the `Live` effect
//! (`publish(topic)`), every connection subscribed to that topic is
//! re-run against the now-current database and the fresh view is what
//! gets delivered. This is exactly the Turbo-8 broadcast-refresh model
//! the research identified as the minimal-wiring live path; the
//! Slice-2 structural `diff` is the documented *next* optimization
//! (push `Change`s + a client morph instead of the full view).
//!
//! This module is the mechanism, fully testable in process. Real
//! sockets (slice 4d) are a thin shell: write each returned body down
//! that connection's WebSocket/SSE. The publish seam is thread-local
//! (`wasm::drain_published`), so `request` and `drain_and_push` must be
//! called on the same thread — true for the in-process hub and for a
//! per-connection worker; a multi-tenant server swaps the seam for a
//! real bus without touching the language surface.

use crate::wasm::{
    drain_published, drain_resp_headers, reason_phrase, serve_request_db,
    HttpResponse, RunError,
};
use std::collections::HashMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::time::Duration;

/// A live connection: the request that rendered its page, plus the
/// topics that page subscribes to (the app's `subscriptions(Model)`
/// supplies these; the hub is topic-source-agnostic).
struct Conn {
    id: u64,
    topics: Vec<String>,
    method: String,
    path: String,
    body: String,
    /// The serialized prior view (design.md §10, A3). Empty unless a
    /// `live_handler` is set; then it is the `serialize_element` of the
    /// last view this connection was sent, fed back so the canonical
    /// Cairn `diff` runs against it.
    prev: String,
}

/// Holds the lowered app and its open connections; turns `publish`
/// into targeted re-renders. With a `live_handler` set the re-render
/// is a structural diff (design.md §10, A3); otherwise it is the
/// broadcast-refresh full page (A1).
pub struct LiveHub {
    wasm: Vec<u8>,
    handler: String,
    live_handler: Option<String>,
    db_path: String,
    next: u64,
    conns: Vec<Conn>,
}

/// Split a `run_app_live` envelope `"<len>:<payload><rest>"` into
/// `(payload, rest)`. `len` is the byte length of `payload`; `rest` is
/// the trailing serialized view. A body with no ':' (a non-live
/// handler) is returned as `(body, "")` so the broadcast path is
/// unaffected.
fn split_envelope(body: &str) -> (String, String) {
    let Some(colon) = body.find(':') else {
        return (body.to_string(), String::new());
    };
    let Ok(n) = body[..colon].parse::<usize>() else {
        return (body.to_string(), String::new());
    };
    let start = colon + 1;
    let end = start + n;
    if end > body.len() {
        return (body.to_string(), String::new());
    }
    (body[start..end].to_string(), body[end..].to_string())
}

impl LiveHub {
    pub fn new(wasm: Vec<u8>, handler: &str, db_path: &str) -> Self {
        Self {
            wasm,
            handler: handler.into(),
            live_handler: None,
            db_path: db_path.into(),
            next: 1,
            conns: Vec::new(),
        }
    }

    /// Opt this hub into diff-based push (design.md §10, A3): `name` is
    /// the app's `<handler>_live` entry (the `run_app_live` envelope —
    /// full HTML on first render, the Change wire thereafter). Without
    /// this the hub stays broadcast-refresh (A1), unchanged.
    pub fn set_live_handler(&mut self, name: &str) {
        self.live_handler = Some(name.into());
    }

    /// Open a connection for a page request. Returns its id and the
    /// initial rendered response. `topics` is what this page subscribes
    /// to. Any publishes during the initial render are discarded — a
    /// fresh page is its own latest state.
    pub fn connect(
        &mut self,
        topics: Vec<String>,
        method: &str,
        path: &str,
        body: &str,
    ) -> Result<(u64, HttpResponse), RunError> {
        // With diff-push the first render goes through `<handler>_live`
        // with an empty prior: the envelope's payload is the full HTML
        // document (returned to the browser) and its trailing part is
        // the serialized baseline this connection diffs against next.
        let (resp, prev) = if let Some(live) = self.live_handler.clone() {
            let env = serve_request_db(
                &self.wasm,
                &live,
                &self.db_path,
                method,
                path,
                "",
            )?;
            let (html, ser) = split_envelope(&env.body);
            (
                HttpResponse {
                    status: env.status,
                    body: html,
                },
                ser,
            )
        } else {
            let resp = serve_request_db(
                &self.wasm,
                &self.handler,
                &self.db_path,
                method,
                path,
                body,
            )?;
            (resp, String::new())
        };
        let _ = drain_published();
        let _ = drain_resp_headers(); // SSE stream carries no app headers
        let id = self.next;
        self.next += 1;
        self.conns.push(Conn {
            id,
            topics,
            method: method.into(),
            path: path.into(),
            body: body.into(),
            prev,
        });
        Ok((id, resp))
    }

    /// Like [`connect`](Self::connect) but the page's topics come from
    /// the app itself: a `subscriptions(req) -> Response` handler whose
    /// body is the whitespace-delimited topic list (the v1 ABI form of
    /// design.md §10's `subscriptions: Fn(Model) -> List<Topic>` —
    /// pure, explicit, reviewed, no magic dependency capture). An empty
    /// body means the page subscribes to nothing.
    pub fn connect_subscribed(
        &mut self,
        subs_handler: &str,
        method: &str,
        path: &str,
        body: &str,
    ) -> Result<(u64, HttpResponse), RunError> {
        let subs = serve_request_db(
            &self.wasm,
            subs_handler,
            &self.db_path,
            method,
            path,
            body,
        )?;
        let _ = drain_published();
        let _ = drain_resp_headers();
        let topics: Vec<String> = subs
            .body
            .split_whitespace()
            .map(str::to_string)
            .collect();
        self.connect(topics, method, path, body)
    }

    /// A one-shot request that is not itself a live connection (e.g. the
    /// POST that mutates state and `publish`es). Publishes it performs
    /// are queued on this thread for the next `drain_and_push`.
    pub fn request(
        &self,
        method: &str,
        path: &str,
        body: &str,
    ) -> Result<HttpResponse, RunError> {
        serve_request_db(
            &self.wasm,
            &self.handler,
            &self.db_path,
            method,
            path,
            body,
        )
    }

    /// Drain the topics published on this thread since the last call;
    /// for every connection subscribed to any of them, re-render
    /// (stateless: the same request against the new DB state) and
    /// return `(conn_id, fresh response)` to write down its socket.
    /// Unsubscribed connections are untouched.
    pub fn drain_and_push(
        &mut self,
    ) -> Result<Vec<(u64, HttpResponse)>, RunError> {
        let topics = drain_published();
        if topics.is_empty() {
            return Ok(Vec::new());
        }
        // Snapshot the subscribed connections first so the wasm calls
        // below don't alias `self.conns` (the live path writes back
        // each conn's new serialized prior).
        #[allow(clippy::type_complexity)]
        let hits: Vec<(usize, u64, String, String, String, String)> = self
            .conns
            .iter()
            .enumerate()
            .filter(|(_, c)| c.topics.iter().any(|t| topics.contains(t)))
            .map(|(i, c)| {
                (
                    i,
                    c.id,
                    c.method.clone(),
                    c.path.clone(),
                    c.body.clone(),
                    c.prev.clone(),
                )
            })
            .collect();
        let mut out = Vec::new();
        match self.live_handler.clone() {
            Some(live) => {
                for (i, id, method, path, _body, prev) in hits {
                    // Re-render through `<handler>_live` with the prior
                    // view as `req.body`: the envelope is the Change
                    // wire vs that prior, plus the new serialized view.
                    let env = serve_request_db(
                        &self.wasm,
                        &live,
                        &self.db_path,
                        &method,
                        &path,
                        &prev,
                    )?;
                    let (changes, ser) = split_envelope(&env.body);
                    self.conns[i].prev = ser;
                    out.push((
                        id,
                        HttpResponse {
                            status: env.status,
                            body: changes,
                        },
                    ));
                }
            }
            None => {
                for (_, id, method, path, body, _prev) in hits {
                    // Broadcast-refresh (A1): the conn's stored request
                    // re-rendered whole, unchanged.
                    let resp = serve_request_db(
                        &self.wasm,
                        &self.handler,
                        &self.db_path,
                        &method,
                        &path,
                        &body,
                    )?;
                    out.push((id, resp));
                }
            }
        }
        // Subscriber re-renders are SSE-only; discard any headers they
        // set so they never leak into the next normal HTTP response.
        let _ = drain_resp_headers();
        Ok(out)
    }

    /// Drop a connection by id — called when its socket write fails
    /// (the client went away). After this the connection is no longer
    /// re-rendered by `drain_and_push`. Unknown ids are a no-op (an
    /// already-pruned conn double-reported as dead is not an error).
    pub fn disconnect(&mut self, id: u64) {
        self.conns.retain(|c| c.id != id);
    }

    /// Open connection count (for observability/tests).
    pub fn connections(&self) -> usize {
        self.conns.len()
    }

    /// Diff-push mode (a `live_handler` is set)? Then pushes are the
    /// Change wire and the SSE stream sends no initial frame — the
    /// browser already has the server-rendered document and the
    /// baseline prior is held here. Broadcast mode sends the full page.
    pub fn is_diff(&self) -> bool {
        self.live_handler.is_some()
    }
}

/// Format an HTML body as one Server-Sent Events message: every line
/// prefixed `data: `, terminated by a blank line. SSE is the v1 live
/// transport — it is a plain long-lived HTTP response (no handshake,
/// no framing protocol), so it composes with the existing blocking
/// HTTP serve far more simply than WebSocket. The browser rejoins
/// multi-line `data:` with `\n`.
pub fn sse_frame(body: &str) -> String {
    let mut out = String::with_capacity(body.len() + 16);
    for line in body.split('\n') {
        out.push_str("data: ");
        out.push_str(line);
        out.push('\n');
    }
    out.push('\n');
    out
}

/// The entire client side of liveness: open an `EventSource` to the
/// live endpoint, **carrying this page's path+query**, and on each
/// pushed message either apply the length-prefixed Change wire (A3
/// diff-push: `^\d+:` — parse, resolve each leaf-to-root path over
/// `document.body.firstChild`, and SetText/Replace/Append/Truncate via
/// real DOM ops, building nodes from the same Element codec) or, for a
/// full-page broadcast (A1), swap `document.body.innerHTML`. One
/// discriminator, one shim for both modes — framework-emitted, never
/// authored (the "no wiring" payoff). The host serve loop appends this
/// to every page render.
pub fn live_client_script(live_path: &str) -> String {
    // Raw string (no format! brace-escaping); `__LP__` is the only
    // interpolation. JS is `;`-separated so the Rust line-continuations
    // keep it one logical line. The reader threads a byte offset; the
    // wire is `count ":" change…`, a change is an op byte then the path
    // `n ":" d ":" …` then the kind's payload (a length-prefixed string
    // / a recursively-encoded Element / a Truncate length).
    const JS: &str = r#"<script>(function(){
var p=encodeURIComponent(location.pathname+location.search);
var es=new EventSource("__LP__?path="+p);
function root(){return document.body.firstChild;}
function ri(s,o){var j=s.indexOf(":",o);return [parseInt(s.slice(o,j),10),j+1];}
function rs(s,o){var x=ri(s,o);return [s.slice(x[1],x[1]+x[0]),x[1]+x[0]];}
function rp(s,o){var x=ri(s,o),n=x[0],q=x[1],a=[];for(var i=0;i<n;i++){var y=ri(s,q);a.push(y[0]);q=y[1];}return [a,q];}
function re(s,o){var t=s[o];o++;if(t=="T"){var x=rs(s,o);return [document.createTextNode(x[0]),x[1]];}var g=rs(s,o);o=g[1];var c=ri(s,o),nk=c[0];o=c[1];var e=document.createElement(g[0]);for(var i=0;i<nk;i++){var k=re(s,o);e.appendChild(k[0]);o=k[1];}return [e,o];}
function resolve(a){var c=root();for(var i=a.length-1;i>=0;i--){c=c.childNodes[a[i]];}return c;}
function apply(s){var x=ri(s,0),n=x[0],o=x[1];for(var i=0;i<n;i++){var op=s[o];o++;var pr=rp(s,o),a=pr[0];o=pr[1];if(op=="S"){var v=rs(s,o);o=v[1];resolve(a).textContent=v[0];}else if(op=="R"){var m=re(s,o);o=m[1];resolve(a).replaceWith(m[0]);}else if(op=="A"){var m2=re(s,o);o=m2[1];resolve(a).appendChild(m2[0]);}else if(op=="T"){var l=ri(s,o);o=l[1];var el=resolve(a);while(el.childNodes.length>l[0]){el.removeChild(el.lastChild);}}}}
es.onmessage=function(ev){var d=ev.data;if(/^[0-9]+:/.test(d)){apply(d);}else{document.body.innerHTML=d;}};
})();</script>"#;
    JS.replace("__LP__", live_path)
}

/// Percent-decode a query-string component. `encodeURIComponent` (the
/// shim's encoder) emits `%XX` for reserved bytes and never `+` for
/// space, so we decode `%XX` and pass everything else through. Invalid
/// escapes are left literal — a best-effort decode never panics.
fn urldecode(s: &str) -> String {
    let b = s.as_bytes();
    let mut out = Vec::with_capacity(b.len());
    let mut i = 0;
    while i < b.len() {
        if b[i] == b'%' && i + 2 < b.len() {
            let hi = (b[i + 1] as char).to_digit(16);
            let lo = (b[i + 2] as char).to_digit(16);
            if let (Some(h), Some(l)) = (hi, lo) {
                out.push((h * 16 + l) as u8);
                i += 3;
                continue;
            }
        }
        out.push(b[i]);
        i += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}

/// First value of `key` in an `&`-separated query string, decoded.
fn query_param(query: &str, key: &str) -> Option<String> {
    query.split('&').find_map(|kv| {
        let (k, v) = kv.split_once('=')?;
        (k == key).then(|| urldecode(v))
    })
}

/// The concurrent SSE server (design.md §10, A1). Liveness is gated on
/// the served handler's `requires Live` (decided upstream); this is the
/// network shell over the proven [`LiveHub`].
///
/// **Concurrency model: a single-threaded accept loop with a parked-
/// socket registry — deliberately not thread-per-connection.** The
/// publish seam (`wasm::drain_published`) is thread-local; running every
/// wasm call (page renders, the SSE initial render, mutating handlers,
/// and `drain_and_push`) on this one thread makes publish→push correct
/// for free. The only concurrency actually required is *holding N SSE
/// responses open while still accepting requests*, which is a registry
/// of parked `TcpStream`s, not concurrent handler execution. A slow SSE
/// reader can stall the loop in `write_all`; bounded by a per-push write
/// timeout + prune-on-error. A thread-pool/async server is the
/// escalation when a real need forces it (design.md §9, Principle 10),
/// not stubbed here.
pub fn serve_http_live(
    wasm: Vec<u8>,
    handler: &str,
    subs: Option<&str>,
    live: Option<&str>,
    addr: &str,
    db_path: &str,
    live_path: &str,
) -> Result<(), RunError> {
    let listener =
        TcpListener::bind(addr).map_err(|e| RunError::Wasmtime(e.to_string()))?;
    let mut hub = LiveHub::new(wasm, handler, db_path);
    // `<handler>_live` present → structural diff push (A3); absent →
    // broadcast-refresh (A1). Authoring picks neither; the convention does.
    if let Some(l) = live {
        hub.set_live_handler(l);
    }
    serve_http_live_on(listener, hub, subs, live_path)
}

/// The accept loop, over a pre-bound listener so it is loopback-testable
/// (the public entry binds, then calls this). Returns only on a listener
/// error; like `serve_http`, the loop itself otherwise runs forever.
fn serve_http_live_on(
    listener: TcpListener,
    mut hub: LiveHub,
    subs: Option<&str>,
    live_path: &str,
) -> Result<(), RunError> {
    // conn id -> its parked long-lived SSE socket.
    let mut parked: HashMap<u64, TcpStream> = HashMap::new();

    for stream in listener.incoming() {
        let Ok(mut stream) = stream else { continue };
        let mut reader = BufReader::new(&stream);

        let mut line = String::new();
        if reader.read_line(&mut line).is_err() {
            continue;
        }
        let mut parts = line.split_whitespace();
        let method = parts.next().unwrap_or("GET").to_string();
        let target = parts.next().unwrap_or("/").to_string();

        let mut content_length = 0usize;
        loop {
            let mut h = String::new();
            if reader.read_line(&mut h).is_err() {
                break;
            }
            let h = h.trim_end();
            if h.is_empty() {
                break;
            }
            if let Some((name, value)) = h.split_once(':') {
                if name.eq_ignore_ascii_case("content-length") {
                    content_length = value.trim().parse().unwrap_or(0);
                }
            }
        }
        let mut body = String::new();
        if content_length > 0 {
            let mut buf = vec![0u8; content_length];
            if reader.read_exact(&mut buf).is_ok() {
                body = String::from_utf8_lossy(&buf).into_owned();
            }
        }
        drop(reader);

        let (path, query) = target.split_once('?').unwrap_or((&target, ""));

        // --- The live endpoint: register a long-lived SSE connection. ---
        if path == live_path {
            let page = query_param(query, "path").unwrap_or_else(|| "/".into());
            let opened = match subs {
                Some(s) => hub.connect_subscribed(s, "GET", &page, ""),
                None => hub.connect(Vec::new(), "GET", &page, ""),
            };
            let Ok((id, resp)) = opened else { continue };
            // SSE is a plain long-lived HTTP response. In broadcast mode
            // push the current render immediately (self-heals the
            // GET↔first-publish race); the body is shim-free so it
            // never spawns a 2nd EventSource. In diff mode send no
            // initial frame — the browser already has the server
            // -rendered document and the baseline prior is held here.
            let head = "HTTP/1.1 200 OK\r\n\
                Content-Type: text/event-stream\r\n\
                Cache-Control: no-cache\r\n\
                Connection: keep-alive\r\n\r\n";
            let _ = stream.set_write_timeout(Some(Duration::from_secs(5)));
            let ok = stream.write_all(head.as_bytes()).is_ok()
                && (hub.is_diff()
                    || stream
                        .write_all(sse_frame(&resp.body).as_bytes())
                        .is_ok());
            if ok {
                parked.insert(id, stream);
            } else {
                hub.disconnect(id);
            }
            continue;
        }

        // --- A normal request: run it, then fan any publish out. ---
        let (status, mut out_body) =
            match hub.request(&method, &target, &body) {
                Ok(r) => (r.status, r.body),
                Err(e) => (500, format!("cairn handler error: {e}")),
            };
        // The host appends the live shim to page documents (GET). It is
        // NOT in pushed SSE frames — the EventSource is created once;
        // refreshes only replace body content. Zero authored client code.
        if method.eq_ignore_ascii_case("GET") {
            out_body.push_str(&live_client_script(live_path));
        }
        // The Resp effect: emit headers this request set (same thread,
        // right after the handler), before the blank line.
        let extra: String = drain_resp_headers()
            .into_iter()
            .map(|(n, v)| format!("{n}: {v}\r\n"))
            .collect();
        let resp = format!(
            "HTTP/1.1 {status} {}\r\nContent-Length: {}\r\n\
             Content-Type: text/html; charset=utf-8\r\n{extra}\r\n{}",
            reason_phrase(status),
            out_body.len(),
            out_body
        );
        let _ = stream.write_all(resp.as_bytes());
        drop(stream);

        // Fan the publish (if any) out to every subscribed parked conn.
        let pushes = match hub.drain_and_push() {
            Ok(p) => p,
            Err(_) => continue,
        };
        for (id, r) in pushes {
            let frame = sse_frame(&r.body);
            let dead = match parked.get_mut(&id) {
                Some(sock) => sock.write_all(frame.as_bytes()).is_err(),
                None => false,
            };
            if dead {
                parked.remove(&id);
                hub.disconnect(id);
            }
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::edit::{Editor, ModuleSpec};
    use crate::node::{Param, Produces};
    use crate::store::Store;
    use crate::ty::{Confidence, Effect, Type};
    use crate::web;
    use crate::wasm::lower;
    use crate::ExprSpec;
    use std::collections::BTreeSet;

    /// The counter app used by every live test: `app(req)` — `/add`
    /// bumps a SQLite counter and `publish("items")`, anything else
    /// renders `count: <v>`; `subs(req)` — `/b` subscribes to "other",
    /// everything else to "items" (pure, explicit; design.md §10).
    /// Lowered to wasm; the served handler is `app`, its subscriptions
    /// handler `subs`.
    fn counter_app_wasm() -> Vec<u8> {
        let lit = |s: &str| ExprSpec::Str(s.into());
        let r = |n: &str| ExprSpec::Ref(n.into());
        let cat = |a: ExprSpec, b: ExprSpec| {
            ExprSpec::StrConcat(Box::new(a), Box::new(b))
        };
        let field = |b: ExprSpec, t: &str, f: &str| ExprSpec::Field {
            base: Box::new(b),
            type_name: t.into(),
            field: f.into(),
        };
        let dbq = |sql: &str| ExprSpec::DbQuery {
            sql: Box::new(lit(sql)),
            params: Box::new(ExprSpec::ListEmpty { elem: Type::String }),
        };
        let resp = |body: ExprSpec| ExprSpec::Record {
            type_name: "Response".into(),
            fields: vec![
                ("status".into(), ExprSpec::Lit(200)),
                ("body".into(), body),
            ],
        };
        let create =
            "CREATE TABLE IF NOT EXISTS ctr(id INTEGER PRIMARY KEY, v INTEGER)";
        let mut effs = BTreeSet::new();
        effs.insert(Effect::Db);
        effs.insert(Effect::Live);

        let app = crate::FunctionSpec {
            name: "app".into(),
            type_params: vec![],
            params: vec![Param {
                name: "req".into(),
                ty: Type::Named("Request".into()),
                min_confidence: Confidence::External,
            }],
            produces: Produces {
                ty: Type::Named("Response".into()),
                confidence: Confidence::External,
            },
            requires: effs,
            on_failure: vec![],
            steps: vec![
                crate::StepSpec {
                    binding: "path".into(),
                    value: field(r("req"), "Request", "path"),
                },
                crate::StepSpec {
                    binding: "c".into(),
                    value: dbq(create),
                },
            ],
            result: ExprSpec::If {
                cond: Box::new(ExprSpec::StrEq(
                    Box::new(r("path")),
                    Box::new(lit("/add")),
                )),
                then_branch: Box::new(resp(cat(
                    cat(
                        lit("ok"),
                        dbq("INSERT INTO ctr(id,v) VALUES(1,1) \
                             ON CONFLICT(id) DO UPDATE SET v=v+1"),
                    ),
                    ExprSpec::NumberToStr(Box::new(ExprSpec::Publish(
                        Box::new(lit("items")),
                    ))),
                ))),
                else_branch: Box::new(resp(cat(
                    lit("count: "),
                    dbq("SELECT COALESCE((SELECT v FROM ctr WHERE id=1),0)"),
                ))),
            },
        };

        let subs = crate::FunctionSpec {
            name: "subs".into(),
            type_params: vec![],
            params: vec![Param {
                name: "req".into(),
                ty: Type::Named("Request".into()),
                min_confidence: Confidence::External,
            }],
            produces: Produces {
                ty: Type::Named("Response".into()),
                confidence: Confidence::External,
            },
            requires: BTreeSet::new(),
            on_failure: vec![],
            steps: vec![crate::StepSpec {
                binding: "path".into(),
                value: field(r("req"), "Request", "path"),
            }],
            result: ExprSpec::If {
                cond: Box::new(ExprSpec::StrEq(
                    Box::new(r("path")),
                    Box::new(lit("/b")),
                )),
                then_branch: Box::new(resp(lit("other"))),
                else_branch: Box::new(resp(lit("items"))),
            },
        };

        let e = Editor::new(Store::open_in_memory().unwrap());
        let (m, report) = e
            .apply_module(&ModuleSpec {
                name: "live".into(),
                types: web::types(),
                functions: vec![app, subs],
            })
            .unwrap();
        assert!(report.ok(), "violations: {:?}", report.violations);
        lower(e.store(), &m).unwrap()
    }

    /// publish → targeted re-render. A connection subscribed to "items"
    /// is refreshed after a request publishes it; one subscribed to
    /// "other" is not. The whole live loop, in process, no sockets.
    #[test]
    fn publish_refreshes_only_subscribed_connections() {
        let wasm = counter_app_wasm();

        let mut path = std::env::temp_dir();
        path.push(format!("cairn-live-{}.db", std::process::id()));
        let _ = std::fs::remove_file(&path);
        let dbp = path.to_str().unwrap();

        let mut hub = LiveHub::new(wasm, "app", dbp);
        // Topics derived from the app's own `subs` handler, not hardcoded.
        let (a, a0) = hub
            .connect_subscribed("subs", "GET", "/", "")
            .unwrap();
        assert!(a0.body.contains("count: 0"), "initial render: {}", a0.body);
        let (b, _) = hub
            .connect_subscribed("subs", "GET", "/b", "")
            .unwrap();
        assert_eq!(hub.connections(), 2);

        // A request that mutates and publishes "items".
        hub.request("POST", "/add", "").unwrap();
        let pushes = hub.drain_and_push().unwrap();

        assert_eq!(pushes.len(), 1, "only the 'items' subscriber refreshes");
        assert_eq!(pushes[0].0, a, "connection A (subscribed to items)");
        assert_ne!(pushes[0].0, b, "connection B must not be pushed");
        assert!(
            pushes[0].1.body.contains("count: 1"),
            "refresh reflects new state: {}",
            pushes[0].1.body
        );

        // No publish since → nothing to push.
        assert!(hub.drain_and_push().unwrap().is_empty());

        // Drop the only "items" subscriber (its socket died). A
        // subsequent publish of "items" must push nothing — the pruned
        // connection is no longer re-rendered — and the count drops.
        hub.disconnect(a);
        assert_eq!(hub.connections(), 1, "B remains after A is pruned");
        hub.request("POST", "/add", "").unwrap();
        assert!(
            hub.drain_and_push().unwrap().is_empty(),
            "a disconnected connection is never refreshed"
        );
        hub.disconnect(a); // double-report of a dead conn is a no-op
        assert_eq!(hub.connections(), 1);
        std::fs::remove_file(&path).unwrap();
    }

    #[test]
    fn sse_frame_and_client_shim() {
        // Multi-line body → one `data:` line each, blank-line terminated.
        assert_eq!(
            sse_frame("<div>a</div>\n<p>b</p>"),
            "data: <div>a</div>\ndata: <p>b</p>\n\n"
        );
        assert_eq!(sse_frame(""), "data: \n\n");
        // The shim is an EventSource against the live path carrying the
        // page path+query, with both modes: apply the Change wire when
        // it matches `^\d+:`, else swap full HTML.
        let s = live_client_script("/__live/");
        assert!(s.contains("new EventSource(\"/__live/?path=\"+p)"));
        assert!(s.contains("encodeURIComponent(location.pathname+location.search)"));
        // diff-push branch
        assert!(s.contains("function apply(s)"));
        assert!(s.contains("/^[0-9]+:/.test(d)"));
        assert!(s.contains("resolve(a).textContent=v[0]"));
        assert!(s.contains("resolve(a).replaceWith"));
        assert!(s.contains("childNodes[a[i]]"));
        // broadcast fallback
        assert!(s.contains("document.body.innerHTML=d"));
        assert!(s.starts_with("<script>") && s.ends_with("</script>"));
    }

    /// The whole A1 loop over a real loopback socket: an SSE client
    /// connects, a separate POST publishes, and the publish is delivered
    /// down the parked SSE socket as a fresh frame — plus a normal GET
    /// page carries the auto-injected live shim. Proves the single-
    /// thread + parked-registry model end to end without a browser.
    #[test]
    fn serve_http_live_pushes_a_refresh_over_a_real_socket() {
        let wasm = counter_app_wasm();
        let mut path = std::env::temp_dir();
        path.push(format!("cairn-live-srv-{}.db", std::process::id()));
        let _ = std::fs::remove_file(&path);
        let dbp = path.to_str().unwrap().to_string();

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let hub = LiveHub::new(wasm, "app", &dbp);
        // The server thread owns every wasm call (and the thread-local
        // publish seam); the client below does pure TCP — exactly the
        // single-thread invariant serve_http_live relies on.
        std::thread::spawn(move || {
            let _ = serve_http_live_on(listener, hub, Some("subs"), "/__live/");
        });

        // Read from `s` (5s deadline) until `needle` appears.
        fn wait_for(s: &mut TcpStream, needle: &str) -> String {
            s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
            let mut acc = String::new();
            let mut buf = [0u8; 1024];
            loop {
                match s.read(&mut buf) {
                    Ok(0) => break,
                    Ok(n) => {
                        acc.push_str(&String::from_utf8_lossy(&buf[..n]));
                        if acc.contains(needle) {
                            break;
                        }
                    }
                    Err(_) => break,
                }
            }
            acc
        }
        let connect = || loop {
            if let Ok(s) = TcpStream::connect(("127.0.0.1", port)) {
                return s;
            }
            std::thread::sleep(Duration::from_millis(10));
        };

        // 1. Open the SSE connection for page "/": event-stream head +
        //    the current render as the first frame.
        let mut sse = connect();
        sse.write_all(b"GET /__live/?path=%2F HTTP/1.1\r\nHost: x\r\n\r\n")
            .unwrap();
        let first = wait_for(&mut sse, "count: 0");
        assert!(first.contains("text/event-stream"), "SSE head: {first}");
        assert!(first.contains("data: count: 0"), "first frame: {first}");

        // 2. A POST that mutates state and publishes "items".
        let mut post = connect();
        post.write_all(
            b"POST /add HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n",
        )
        .unwrap();
        let _ = wait_for(&mut post, "\r\n\r\n");

        // 3. The publish fans out: the parked SSE socket gets a fresh
        //    frame reflecting the new state — zero authored client code.
        let pushed = wait_for(&mut sse, "count: 1");
        assert!(pushed.contains("data: count: 1"), "no push: {pushed}");

        // 4. A normal GET page carries the auto-injected live shim and
        //    reflects persisted state.
        let mut page = connect();
        page.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
        let doc = wait_for(&mut page, "</script>");
        assert!(doc.contains("count: 1"), "page state: {doc}");
        assert!(
            doc.contains("new EventSource(\"/__live/?path=\"+p)"),
            "shim not injected: {doc}"
        );

        let _ = std::fs::remove_file(&path);
    }

    /// A3 over a real socket: with the `_live` handler the server sends
    /// the SSE head (no initial frame — the diff baseline is held
    /// server-side), and a publish delivers the length-prefixed Change
    /// wire (not the full page) down the parked socket. The document
    /// still carries the dual-mode shim.
    #[test]
    fn serve_http_live_pushes_the_change_wire_over_a_real_socket() {
        let wasm = tea_live_counter_wasm();
        let mut path = std::env::temp_dir();
        path.push(format!("cairn-diffsrv-{}.db", std::process::id()));
        let _ = std::fs::remove_file(&path);
        let dbp = path.to_str().unwrap().to_string();

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let mut hub = LiveHub::new(wasm, "app", &dbp);
        hub.set_live_handler("app_live");
        std::thread::spawn(move || {
            let _ = serve_http_live_on(listener, hub, Some("subs"), "/__live/");
        });

        fn wait_for(s: &mut TcpStream, needle: &str) -> String {
            s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
            let mut acc = String::new();
            let mut buf = [0u8; 1024];
            loop {
                match s.read(&mut buf) {
                    Ok(0) => break,
                    Ok(n) => {
                        acc.push_str(&String::from_utf8_lossy(&buf[..n]));
                        if acc.contains(needle) {
                            break;
                        }
                    }
                    Err(_) => break,
                }
            }
            acc
        }
        let connect = || loop {
            if let Ok(s) = TcpStream::connect(("127.0.0.1", port)) {
                return s;
            }
            std::thread::sleep(Duration::from_millis(10));
        };

        // 1. The document: server-rendered page + the dual-mode shim.
        let mut page = connect();
        page.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
        let doc = wait_for(&mut page, "</script>");
        assert!(doc.contains("<div>count: 0</div>"), "page: {doc}");
        assert!(
            doc.contains("new EventSource(\"/__live/?path=\"+p)")
                && doc.contains("function apply(s)"),
            "dual-mode shim missing: {doc}"
        );

        // 2. SSE: head only (diff mode sends no initial frame).
        let mut sse = connect();
        sse.write_all(b"GET /__live/?path=%2F HTTP/1.1\r\nHost: x\r\n\r\n")
            .unwrap();
        let head = wait_for(&mut sse, "\r\n\r\n");
        assert!(head.contains("text/event-stream"), "SSE head: {head}");

        // 3. The action publishes; the parked SSE gets the Change wire.
        let mut act = connect();
        act.write_all(b"GET /inc HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
        let _ = wait_for(&mut act, "\r\n\r\n");
        let pushed = wait_for(&mut sse, "count: 1");
        assert!(
            pushed.contains("data: 1:S1:0:8:count: 1"),
            "expected the Change wire, got: {pushed:?}"
        );
        assert!(!pushed.contains("<div>"), "must not be full HTML");

        let _ = std::fs::remove_file(&path);
    }

    /// A counter whose page/action handler `app` mutates+publishes (A1
    /// style) and whose `app_live(req) = render_live(req.body, cload,
    /// cview)` is the render-only diff entry; `subs` → topic "c". The
    /// served handler is `app`, the live handler `app_live`.
    fn tea_live_counter_wasm() -> Vec<u8> {
        let lit = |s: &str| ExprSpec::Str(s.into());
        let r = |n: &str| ExprSpec::Ref(n.into());
        let cat = |a: ExprSpec, b: ExprSpec| {
            ExprSpec::StrConcat(Box::new(a), Box::new(b))
        };
        let field = |b: ExprSpec, t: &str, f: &str| ExprSpec::Field {
            base: Box::new(b),
            type_name: t.into(),
            field: f.into(),
        };
        let dbq = |sql: &str| ExprSpec::DbQuery {
            sql: Box::new(lit(sql)),
            params: Box::new(ExprSpec::ListEmpty { elem: Type::String }),
        };
        let call = |f: &str, a: Vec<ExprSpec>| ExprSpec::Call {
            func: f.into(),
            args: a,
        };
        let resp = |body: ExprSpec| ExprSpec::Record {
            type_name: "Response".into(),
            fields: vec![
                ("status".into(), ExprSpec::Lit(200)),
                ("body".into(), body),
            ],
        };
        let create =
            "CREATE TABLE IF NOT EXISTS ctr(id INTEGER PRIMARY KEY, v INTEGER)";
        let mut dbl = BTreeSet::new();
        dbl.insert(Effect::Db);
        dbl.insert(Effect::Live);
        let mut dbo = BTreeSet::new();
        dbo.insert(Effect::Db);

        // cload() -> Counter @ Db
        let cload = crate::FunctionSpec {
            name: "cload".into(),
            type_params: vec![],
            params: vec![],
            produces: Produces {
                ty: Type::Named("Counter".into()),
                confidence: Confidence::External,
            },
            requires: dbo.clone(),
            on_failure: vec![],
            steps: vec![crate::StepSpec {
                binding: "c".into(),
                value: dbq(create),
            }],
            result: ExprSpec::Record {
                type_name: "Counter".into(),
                fields: vec![(
                    "value".into(),
                    ExprSpec::StrToNumber(Box::new(dbq(
                        "SELECT COALESCE((SELECT v FROM ctr WHERE id=1),0)",
                    ))),
                )],
            },
        };
        // cview(m) -> Element : <div>count: N</div>
        let cview = crate::FunctionSpec {
            name: "cview".into(),
            type_params: vec![],
            params: vec![Param {
                name: "m".into(),
                ty: Type::Named("Counter".into()),
                min_confidence: Confidence::External,
            }],
            produces: Produces {
                ty: Type::Named("Element".into()),
                confidence: Confidence::External,
            },
            requires: BTreeSet::new(),
            on_failure: vec![],
            steps: vec![],
            result: ExprSpec::Variant {
                type_name: "Element".into(),
                case: "El".into(),
                fields: vec![
                    ("tag".into(), lit("div")),
                    (
                        "kids".into(),
                        ExprSpec::List(vec![ExprSpec::Variant {
                            type_name: "Element".into(),
                            case: "Text".into(),
                            fields: vec![(
                                "content".into(),
                                cat(
                                    lit("count: "),
                                    ExprSpec::NumberToStr(Box::new(field(
                                        r("m"),
                                        "Counter",
                                        "value",
                                    ))),
                                ),
                            )],
                        }]),
                    ),
                ],
            },
        };
        // app(req): /inc -> bump + publish("c"); else render the page.
        let app = crate::FunctionSpec {
            name: "app".into(),
            type_params: vec![],
            params: vec![Param {
                name: "req".into(),
                ty: Type::Named("Request".into()),
                min_confidence: Confidence::External,
            }],
            produces: Produces {
                ty: Type::Named("Response".into()),
                confidence: Confidence::External,
            },
            requires: dbl.clone(),
            on_failure: vec![],
            steps: vec![
                crate::StepSpec {
                    binding: "path".into(),
                    value: field(r("req"), "Request", "path"),
                },
                crate::StepSpec {
                    binding: "c".into(),
                    value: dbq(create),
                },
            ],
            result: ExprSpec::If {
                cond: Box::new(ExprSpec::StrEq(
                    Box::new(r("path")),
                    Box::new(lit("/inc")),
                )),
                then_branch: Box::new(resp(cat(
                    cat(
                        lit("ok"),
                        dbq("INSERT INTO ctr(id,v) VALUES(1,1) \
                             ON CONFLICT(id) DO UPDATE SET v=v+1"),
                    ),
                    ExprSpec::NumberToStr(Box::new(ExprSpec::Publish(
                        Box::new(lit("c")),
                    ))),
                ))),
                else_branch: Box::new(resp(call(
                    "render_html",
                    vec![call("cview", vec![call("cload", vec![])])],
                ))),
            },
        };
        // app_live(req) = render_live(req.body, &cload, &cview)
        let app_live = crate::FunctionSpec {
            name: "app_live".into(),
            type_params: vec![],
            params: vec![Param {
                name: "req".into(),
                ty: Type::Named("Request".into()),
                min_confidence: Confidence::External,
            }],
            produces: Produces {
                ty: Type::Named("Response".into()),
                confidence: Confidence::External,
            },
            requires: dbo.clone(),
            on_failure: vec![],
            steps: vec![],
            result: call(
                "render_live",
                vec![
                    field(r("req"), "Request", "body"),
                    ExprSpec::FuncRef("cload".into()),
                    ExprSpec::FuncRef("cview".into()),
                ],
            ),
        };
        // subs(req) -> "c"
        let subs = crate::FunctionSpec {
            name: "subs".into(),
            type_params: vec![],
            params: vec![Param {
                name: "req".into(),
                ty: Type::Named("Request".into()),
                min_confidence: Confidence::External,
            }],
            produces: Produces {
                ty: Type::Named("Response".into()),
                confidence: Confidence::External,
            },
            requires: BTreeSet::new(),
            on_failure: vec![],
            steps: vec![],
            result: resp(lit("c")),
        };

        let mut types = web::types();
        types.push(crate::edit::TypeDefSpec::Record {
            name: "Counter".into(),
            fields: vec![("value".into(), Type::Number)],
        });
        let mut funcs = web::functions();
        funcs.extend([cload, cview, app, app_live, subs]);

        let e = Editor::new(Store::open_in_memory().unwrap());
        let (m, report) = e
            .apply_module(&ModuleSpec {
                name: "tealive".into(),
                types,
                functions: funcs,
            })
            .unwrap();
        assert!(report.ok(), "violations: {:?}", report.violations);
        lower(e.store(), &m).unwrap()
    }

    /// A3 (design.md §10): with a `live_handler` set the hub pushes the
    /// length-prefixed **Change wire** (a structural diff vs the prior
    /// view), not the full page; the connection's prior is threaded so
    /// successive publishes diff against the last sent view. The render
    /// path is idempotent — it never re-applies the message.
    #[test]
    fn diff_push_sends_the_change_wire_not_the_full_page() {
        let wasm = tea_live_counter_wasm();
        let mut path = std::env::temp_dir();
        path.push(format!("cairn-diffpush-{}.db", std::process::id()));
        let _ = std::fs::remove_file(&path);
        let dbp = path.to_str().unwrap();

        let mut hub = LiveHub::new(wasm, "app", dbp);
        hub.set_live_handler("app_live");

        // connect: render-only first render → the HTML document (not an
        // envelope; the hub split it) and a serialized baseline. The
        // baseline render must NOT have mutated state.
        let (id, doc) = hub
            .connect_subscribed("subs", "GET", "/", "")
            .unwrap();
        assert_eq!(
            doc.body, "<div>count: 0</div>",
            "first render is the HTML page at current (un-mutated) state"
        );

        // The action mutates + publishes "c".
        hub.request("GET", "/inc", "").unwrap();
        let pushes = hub.drain_and_push().unwrap();
        assert_eq!(pushes.len(), 1);
        assert_eq!(pushes[0].0, id);
        // Exactly the Change wire: 1 change, SetText at path [0],
        // length-prefixed content "count: 1". Not HTML.
        assert_eq!(pushes[0].1.body, "1:S1:0:8:count: 1");
        assert!(!pushes[0].1.body.contains("<div>"));

        // Prior threading: a second publish diffs against count:1, not
        // the original baseline.
        hub.request("GET", "/inc", "").unwrap();
        let p2 = hub.drain_and_push().unwrap();
        assert_eq!(p2.len(), 1);
        assert_eq!(p2[0].1.body, "1:S1:0:8:count: 2");

        // No publish → nothing.
        assert!(hub.drain_and_push().unwrap().is_empty());
        std::fs::remove_file(&path).unwrap();
    }
}