fn parse_sse_event(event: &str) -> (Option<&str>, Option<String>) {
    let mut kind = None;
    let mut data = Vec::new();
    for raw in event.lines() {
        let line = raw.trim_end_matches('\r');
        if let Some(value) = line.strip_prefix("event:") {
            kind = Some(value.trim());
        } else if let Some(value) = line.strip_prefix("data:") {
            data.push(value.trim_start());
        }
    }
    (kind, (!data.is_empty()).then(|| data.join("\n")))
}

type LocalAdminHttpClient = Client<HttpsConnector<HttpConnector>, Body>;

#[derive(Debug)]
struct LocalAdminCertificateVerifier {
    expected: Vec<[u8; 32]>,
    signature_verifier: Arc<WebPkiServerVerifier>,
}
impl ServerCertVerifier for LocalAdminCertificateVerifier {
    fn verify_server_cert(
        &self,
        end_entity: &CertificateDer<'_>,
        _intermediates: &[CertificateDer<'_>],
        _server_name: &ServerName<'_>,
        _ocsp_response: &[u8],
        _now: UnixTime,
    ) -> std::result::Result<ServerCertVerified, rustls::Error> {
        let actual = Sha256::digest(end_entity.as_ref());
        if !self
            .expected
            .iter()
            .any(|expected| actual[..] == expected[..])
        {
            return Err(rustls::Error::General(
                "local administration TLS certificate is not present in data-dir/certs".into(),
            ));
        }
        Ok(ServerCertVerified::assertion())
    }
    fn verify_tls12_signature(
        &self,
        message: &[u8],
        cert: &CertificateDer<'_>,
        dss: &DigitallySignedStruct,
    ) -> std::result::Result<HandshakeSignatureValid, rustls::Error> {
        self.signature_verifier
            .verify_tls12_signature(message, cert, dss)
    }
    fn verify_tls13_signature(
        &self,
        message: &[u8],
        cert: &CertificateDer<'_>,
        dss: &DigitallySignedStruct,
    ) -> std::result::Result<HandshakeSignatureValid, rustls::Error> {
        self.signature_verifier
            .verify_tls13_signature(message, cert, dss)
    }
    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
        self.signature_verifier.supported_verify_schemes()
    }
}

fn local_admin_public_roots() -> RootCertStore {
    let mut roots = RootCertStore::empty();
    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
    roots
}

fn local_admin_certificate_pins(data_dir: &Path) -> Result<Vec<[u8; 32]>> {
    let directory = data_dir.join("certs");
    let mut pins = Vec::new();
    for entry in std::fs::read_dir(&directory).with_context(|| {
        format!(
            "cannot read local certificate directory {}",
            directory.display()
        )
    })? {
        let entry = entry?;
        let path = entry.path();
        if path.extension().and_then(|value| value.to_str()) != Some("pem") {
            continue;
        }
        let metadata = entry.metadata()?;
        if !metadata.is_file() || metadata.len() > 512 * 1024 {
            continue;
        }
        let bytes = std::fs::read(&path)
            .with_context(|| format!("cannot read local certificate {}", path.display()))?;
        let mut reader = bytes.as_slice();
        let Some(Ok(certificate)) = rustls_pemfile::certs(&mut reader).next() else {
            continue;
        };
        let digest = Sha256::digest(certificate.as_ref());
        let mut pin = [0_u8; 32];
        pin.copy_from_slice(&digest);
        pins.push(pin);
    }
    pins.sort_unstable();
    pins.dedup();
    ensure!(
        !pins.is_empty(),
        "no local TLS certificate was found in {}; start rust-proxy serve first",
        directory.display()
    );
    Ok(pins)
}

fn local_admin_scheme_from_probe(
    status: StatusCode,
    location: Option<&str>,
) -> Result<&'static str> {
    if status == StatusCode::PERMANENT_REDIRECT {
        ensure!(
            location.is_some_and(|value| value.starts_with("https://")),
            "administration HTTP probe returned 308 without an HTTPS location"
        );
        return Ok("https");
    }
    ensure!(
        status.is_success(),
        "administration health probe returned {status}"
    );
    Ok("http")
}

async fn detect_local_admin_scheme(addr: SocketAddr) -> Result<&'static str> {
    let mut connector = HttpConnector::new();
    connector.set_connect_timeout(Some(Duration::from_secs(3)));
    let client: Client<_, Body> = Client::builder(TokioExecutor::new()).build(connector);
    let request = Request::builder()
        .method(Method::GET)
        .uri(format!("http://{addr}/healthz"))
        .header("accept", "application/json")
        .body(Body::empty())?;
    let response = tokio::time::timeout(Duration::from_secs(5), client.request(request))
        .await
        .context("administration health probe timed out")??;
    let location = response
        .headers()
        .get(http::header::LOCATION)
        .and_then(|value| value.to_str().ok());
    local_admin_scheme_from_probe(response.status(), location)
}

#[derive(Clone)]
struct LocalAdminClient {
    addr: SocketAddr,
    scheme: &'static str,
    client: LocalAdminHttpClient,
}
impl LocalAdminClient {
    async fn connect(data_dir: &Path, addr: SocketAddr) -> Result<Self> {
        ensure!(
            addr.ip().is_loopback(),
            "CLI administration requires a loopback address"
        );
        let scheme = detect_local_admin_scheme(addr).await?;
        let provider = Arc::new(rustls::crypto::ring::default_provider());
        let mut config = ClientConfig::builder_with_provider(provider.clone())
            .with_safe_default_protocol_versions()?
            .with_root_certificates(local_admin_public_roots())
            .with_no_client_auth();
        if scheme == "https" {
            let data_dir = data_dir.to_path_buf();
            let expected =
                tokio::task::spawn_blocking(move || local_admin_certificate_pins(&data_dir))
                    .await??;
            let signature_verifier = WebPkiServerVerifier::builder_with_provider(
                Arc::new(local_admin_public_roots()),
                provider,
            )
            .build()?;
            config
                .dangerous()
                .set_certificate_verifier(Arc::new(LocalAdminCertificateVerifier {
                    expected,
                    signature_verifier,
                }));
        }
        let mut http = HttpConnector::new();
        http.enforce_http(false);
        http.set_connect_timeout(Some(Duration::from_secs(3)));
        http.set_nodelay(true);
        let connector = HttpsConnectorBuilder::new()
            .with_tls_config(config)
            .https_or_http()
            .enable_http1()
            .wrap_connector(http);
        Ok(Self {
            addr,
            scheme,
            client: Client::builder(TokioExecutor::new()).build(connector),
        })
    }
    fn uri(&self, path: &str) -> String {
        format!("{}://{}{}", self.scheme, self.addr, path)
    }
    fn origin(&self) -> String {
        format!("{}://{}", self.scheme, self.addr)
    }
}

async fn live_logs_command(dir: PathBuf, addr: SocketAddr) -> Result<()> {
    ensure!(
        addr.ip().is_loopback(),
        "--livelogs requires the local loopback administration listener"
    );
    auth::prepare_directory(&dir)?;
    let path = dir.join("config.sqlite3");
    ensure!(
        path.is_file(),
        "database does not exist at {}; start rust-proxy serve first",
        path.display()
    );
    auth::prepare_database(&path)?;
    let store = Store::open(&path, &Config::default()).await?;
    let admin = LocalAdminClient::connect(&dir, addr).await?;
    let result = async {
        let record = store.admin_record().await?.context("create the administrator first")?;
        let mut session = store
            .create_live_log_session(record.generation)
            .await?
            .context("credentials changed; retry --livelogs")?;
        let (command_tx, mut command_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        std::thread::spawn(move || {
            let stdin = io::stdin();
            for line in stdin.lock().lines() {
                match line {
                    Ok(line) => {
                        if command_tx.send(line).is_err() {
                            break;
                        }
                    }
                    Err(_) => break,
                }
            }
        });
        let interactive = io::stdin().is_terminal() && io::stdout().is_terminal();
        println!("RustProxy live logs 0.28.2 - all vhosts. Type 'help' for interactive filters; Ctrl-C or 'quit' exits.");
        if interactive { print!("> "); io::stdout().flush()?; }
        let mut filters = LiveLogFilters::default();
        let mut history = std::collections::BTreeMap::<u64, serde_json::Value>::new();
        let mut last_id = 0_u64;
        let mut command_input_open = true;

        'reconnect: loop {
            let request = Request::builder()
                .method(Method::GET)
                .uri(admin.uri("/api/live-logs"))
                .header("cookie", format!("{}={}", auth::COOKIE_NAME, session.secret))
                .header("accept", "text/event-stream")
                .body(Body::empty())?;
            let response = match tokio::time::timeout(Duration::from_secs(10), admin.client.request(request)).await {
                Ok(Ok(response)) if response.status().is_success() => response,
                Ok(Ok(response)) => {
                    eprintln!("live-log stream returned {}; reconnecting", response.status());
                    let replacement = store.create_live_log_session(record.generation).await?.context("administrator credentials changed; --livelogs stopped")?;
                    let old = std::mem::replace(&mut session, replacement); let _ = store.end_session(&old.session).await;
                    tokio::time::sleep(Duration::from_secs(1)).await; continue;
                }
                Ok(Err(error)) => { eprintln!("live-log stream unavailable: {error}; reconnecting"); tokio::time::sleep(Duration::from_secs(1)).await; continue; }
                Err(_) => { eprintln!("live-log stream connect timeout; reconnecting"); tokio::time::sleep(Duration::from_secs(1)).await; continue; }
            };
            // The server subscribes before the SSE response is returned. Read the
            // retained history only after that point, so requests cannot disappear
            // between the initial snapshot and the live subscription. Buffered live
            // duplicates are ignored by request id below.
            let history_limit = if last_id == 0 { 25 } else { 1024 };
            if let Ok(snapshot) = api(
                &admin,
                &session,
                Method::GET,
                &format!("/api/logs?limit={history_limit}"),
                None,
            )
            .await
            {
                if let Ok(snapshot) = serde_json::from_str::<serde_json::Value>(&snapshot) {
                    if let Some(entries) = snapshot.get("entries").and_then(|value| value.as_array()) {
                        let mut entries = entries.clone();
                        entries.reverse();
                        for entry in entries {
                            let id = entry.get("id").and_then(|value| value.as_u64()).unwrap_or(0);
                            if id > 0 && id <= last_id {
                                continue;
                            }
                            if id > 0 {
                                last_id = id;
                                history.insert(id, entry.clone());
                            }
                            if !filters.paused && live_log_matches(&entry, &filters) {
                                print_live_entry(&entry, filters.detail);
                            }
                        }
                    }
                }
            }
            let mut body = response.into_body();
            let mut buffer = Vec::new();
            loop {
                tokio::select! {
                    _ = tokio::signal::ctrl_c() => break 'reconnect,
                    command = command_rx.recv(), if command_input_open => {
                        let Some(command) = command else {
                            command_input_open = false;
                            continue;
                        };
                        match handle_live_command(&command, &mut filters, &history) {
                            Ok(true) => {}
                            Ok(false) => break 'reconnect,
                            Err(error) => eprintln!("command error: {error:#}"),
                        }
                        if interactive { print!("> "); io::stdout().flush()?; }
                    }
                    frame = body.frame() => {
                        let Some(frame) = frame else {
                            eprintln!("live-log stream closed; reconnecting");
                            tokio::time::sleep(Duration::from_millis(500)).await;
                            continue 'reconnect;
                        };
                        let frame = match frame {
                            Ok(frame) => frame,
                            Err(error) => { eprintln!("live-log stream error: {error}; reconnecting"); tokio::time::sleep(Duration::from_millis(500)).await; continue 'reconnect; }
                        };
                        let Ok(data) = frame.into_data() else { continue; };
                        buffer.extend_from_slice(&data);
                        ensure!(buffer.len() <= 2 * 1024 * 1024, "live-log stream frame buffer exceeded 2 MiB");
                        while let Some(raw) = take_sse_event(&mut buffer) {
                            let raw = std::str::from_utf8(&raw).context("live-log SSE event is not UTF-8")?;
                            let (kind, data) = parse_sse_event(raw);
                            match (kind, data) {
                                (Some("log"), Some(data)) => {
                                    let entry: serde_json::Value = serde_json::from_str(&data)?;
                                    let id = entry.get("id").and_then(|value| value.as_u64()).unwrap_or(0);
                                    if id > 0 && id <= last_id { continue; }
                                    if id > 0 { last_id = id; history.insert(id, entry.clone()); }
                                    while history.len() > 1024 { if let Some(id) = history.keys().next().copied() { history.remove(&id); } else { break; } }
                                    if !filters.paused && live_log_matches(&entry, &filters) { print_live_entry(&entry, filters.detail); }
                                }
                                (Some("gap"), Some(data)) => {
                                    let dropped = serde_json::from_str::<serde_json::Value>(&data).ok().and_then(|value| value.get("dropped").and_then(|v| v.as_u64())).unwrap_or(0);
                                    eprintln!("*** live-log receiver lagged: {dropped} entries were dropped before display ***");
                                }
                                _ => {}
                            }
                        }
                        if interactive { io::stdout().flush()?; }
                    }
                }
            }
        }
        let _ = store.end_session(&session.session).await;
        Ok::<_, anyhow::Error>(())
    }.await;
    store.close().await;
    result
}

async fn api(
    admin: &LocalAdminClient,
    session: &auth::SessionToken,
    method: Method,
    path: &str,
    body: Option<String>,
) -> Result<String> {
    let request = Request::builder()
        .method(method)
        .uri(admin.uri(path))
        .header(
            "cookie",
            format!("{}={}", auth::COOKIE_NAME, session.secret),
        )
        .header("origin", admin.origin())
        .header("x-rustproxy-request", "1")
        .header("x-csrf-token", &session.session.csrf)
        .header("content-type", "application/json")
        .body(Body::from(body.unwrap_or_default()))?;
    tokio::time::timeout(Duration::from_secs(15), async {
        let response = admin
            .client
            .request(request)
            .await
            .context("admin API unavailable; start rust-proxy serve first")?;
        let status = response.status();
        let bytes = Limited::new(response.into_body(), config::MAX_CONFIG_BYTES + 4096)
            .collect()
            .await
            .map_err(|e| anyhow::anyhow!("admin body error: {e}"))?
            .to_bytes();
        let text = String::from_utf8(bytes.to_vec())?;
        if !status.is_success() {
            bail!("admin API returned {status}: {text}");
        }
        Ok(text)
    })
    .await
    .context("admin API timed out")?
}
async fn read_config_file(file: PathBuf) -> Result<Config> {
    ensure!(
        tokio::fs::metadata(&file).await?.len() <= config::MAX_CONFIG_BYTES as u64,
        "config file exceeds 256 KiB"
    );
    // Bound actual reads as well as metadata; replacement/append races cannot grow memory.
    let data = tokio::task::spawn_blocking(move || {
        let file = std::fs::File::open(file)?;
        let mut bytes = Vec::new();
        file.take((config::MAX_CONFIG_BYTES + 1) as u64)
            .read_to_end(&mut bytes)?;
        anyhow::ensure!(
            bytes.len() <= config::MAX_CONFIG_BYTES,
            "config file exceeds 256 KiB"
        );
        Ok::<_, anyhow::Error>(bytes)
    })
    .await??;
    Ok(serde_json::from_slice(&data)?)
}