//! The FreenetAdapter — SoulStream's ONLY door to a local Freenet node.
//!
//! Everything here follows the supplement design (§5.3) and the rules the
//! F0–F7+F2b spikes recorded (research/freenet-*/RUN-RECORD.md):
//!
//! - **Off by default, kill-switchable.** `enable()` is the only thing that
//!   spawns work; `disable()` stops the supervisor, drops the connection,
//!   and ends every retry. A missing/stopped/incompatible node must never
//!   affect Soulseek or playback — nothing in here is on any audio path.
//! - **The privilege boundary is this crate.** A loopback Freenet API is
//!   fully privileged and unauthenticated (F0): whatever can reach it owns
//!   the node. The webview therefore gets STATUS ONLY through the Tauri
//!   layer — never a node handle, never raw state bytes.
//! - **Detection is cheap and three-valued** (absent / occupied-not-api /
//!   compatible), with its own short timeouts: a refused loopback connect
//!   costs ~2 s on Windows (F0), so the TCP probe gets 600 ms. A TYPED host
//!   error counts as compatible — local mode answers `ConnectedPeers` with
//!   "not supported" (F0), so refusals prove the protocol.
//! - **One request in flight per connection.** The stdlib's `WebApi`
//!   correlates responses FIFO; the session serializes ops and routes
//!   unsolicited `UpdateNotification`s to a bounded event bus (lagging
//!   subscribers lose oldest events, honestly, rather than growing queues).
//! - **Write rules from the spikes:** a node refuses the FIRST write to a
//!   contract it never held ("auto-fetch triggered", F5) — `update` retries
//!   over ~18 s and callers should prime with a GET (`get_then_update`);
//!   sequence-bearing states must be read-and-bumped by the caller (F5);
//!   a restarted node drops out of update propagation (F2) — services
//!   reconcile by relaying their own state after reconnect.
//! - **Redacted diagnostics.** Logs carry ids, lengths and outcomes; never
//!   state bytes, never key material.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;

use freenet_stdlib::client_api::{
    ClientRequest, ContractRequest, ContractResponse, HostResponse, NodeQuery, WebApi,
};
use freenet_stdlib::prelude::*;
use serde::Serialize;
use tokio::sync::{broadcast, mpsc, oneshot, watch};
use tokio::time::timeout;

pub const DEFAULT_WS_PORT: u16 = 7509;

/// TCP probe budget — see the F0 note about Windows' 2 s refusal cost.
const DETECT_TCP_TIMEOUT: Duration = Duration::from_millis(600);
const DETECT_WS_TIMEOUT: Duration = Duration::from_secs(2);
const DETECT_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
const OP_TIMEOUT: Duration = Duration::from_secs(120);
const PUT_TIMEOUT: Duration = Duration::from_secs(180);
/// "auto-fetch triggered" retry schedule (F5): 6 attempts, 3 s apart.
const AUTOFETCH_RETRIES: u32 = 6;
const AUTOFETCH_BACKOFF: Duration = Duration::from_secs(3);
/// Reconnect backoff while enabled but the node is missing/lost.
const SUPERVISOR_BACKOFF: Duration = Duration::from_secs(5);
const CMD_QUEUE: usize = 32;
const EVENT_QUEUE: usize = 64;

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum AdapterStatus {
    /// The kill switch position. Nothing runs, nothing retries.
    Disabled,
    Detecting,
    /// Nothing listens on the port. NOT an error — Freenet is optional.
    NodeAbsent,
    /// Something listens but does not speak the client API.
    NodeOccupied,
    /// Speaks WebSocket but the protocol exchange failed undecipherably.
    NodeIncompatible { detail: String },
    Connected,
    /// Connection lost while enabled; the supervisor is retrying.
    Lost { detail: String },
}

#[derive(Debug, Clone, Serialize)]
#[serde(tag = "kind", rename_all = "kebab-case")]
pub enum AdapterEvent {
    Status(AdapterStatus),
    /// A subscribed contract changed. Ids and lengths only — the webview
    /// never receives replicated bytes through the event path.
    ContractUpdate { instance: String, update_len: usize },
}

#[derive(Debug, thiserror::Error)]
pub enum AdapterError {
    #[error("freenet layer is disabled")]
    Disabled,
    #[error("freenet node is not connected ({0:?})")]
    NotConnected(AdapterStatus),
    #[error("request timed out")]
    Timeout,
    #[error("adapter is shutting down")]
    ShuttingDown,
    #[error("node refused: {0}")]
    Node(String),
    #[error("{0}")]
    Other(String),
}

type Result<T> = std::result::Result<T, AdapterError>;

/// What `contract_put`/`update` return so callers can address the instance
/// later without recomputing.
#[derive(Debug, Clone, Serialize)]
pub struct ContractIds {
    pub instance: String,
    pub code_hash: String,
}

struct Cmd {
    req: ClientRequest<'static>,
    reply: oneshot::Sender<Result<HostResponse>>,
    op_timeout: Duration,
}

struct Inner {
    ws_port: StdMutex<u16>,
    status: StdMutex<AdapterStatus>,
    events: broadcast::Sender<AdapterEvent>,
    /// Sender into the LIVE session, if any.
    session: StdMutex<Option<mpsc::Sender<Cmd>>>,
    /// Kill switch: flipping to false ends the supervisor and its session.
    enabled: watch::Sender<bool>,
    /// Distinguishes supervisor generations so a stale task can't clobber
    /// the status of a newer enable().
    generation: AtomicU64,
}

#[derive(Clone)]
pub struct FreenetAdapter {
    inner: Arc<Inner>,
}

impl Default for FreenetAdapter {
    fn default() -> Self {
        let (enabled, _) = watch::channel(false);
        let (events, _) = broadcast::channel(EVENT_QUEUE);
        Self {
            inner: Arc::new(Inner {
                ws_port: StdMutex::new(DEFAULT_WS_PORT),
                status: StdMutex::new(AdapterStatus::Disabled),
                events,
                session: StdMutex::new(None),
                enabled,
                generation: AtomicU64::new(0),
            }),
        }
    }
}

/// Spawn on the caller's tokio runtime when there is one (async commands,
/// tests), else on a runtime the adapter owns. `enable()` is reached from
/// Tauri's event-loop thread — the setup hook, sync commands — where a bare
/// `tokio::spawn` is a non-unwinding panic that takes the whole process down
/// (0xc0000409, seen live 2026-09-02 on the first toggle of the switch).
fn spawn_anywhere<F>(fut: F)
where
    F: std::future::Future<Output = ()> + Send + 'static,
{
    match tokio::runtime::Handle::try_current() {
        Ok(handle) => {
            handle.spawn(fut);
        }
        Err(_) => {
            static RT: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
            let rt = RT.get_or_init(|| {
                tokio::runtime::Builder::new_multi_thread()
                    .worker_threads(2)
                    .thread_name("freenet-adapter")
                    .enable_all()
                    .build()
                    .expect("freenet adapter runtime")
            });
            rt.spawn(fut);
        }
    }
}


impl FreenetAdapter {
    pub fn set_ws_port(&self, port: u16) {
        *self.inner.ws_port.lock().unwrap() = port;
    }

    pub fn ws_port(&self) -> u16 {
        *self.inner.ws_port.lock().unwrap()
    }

    pub fn status(&self) -> AdapterStatus {
        self.inner.status.lock().unwrap().clone()
    }

    pub fn is_enabled(&self) -> bool {
        *self.inner.enabled.borrow()
    }

    pub fn subscribe(&self) -> broadcast::Receiver<AdapterEvent> {
        self.inner.events.subscribe()
    }

    fn set_status(&self, generation: u64, s: AdapterStatus) {
        if self.inner.generation.load(Ordering::SeqCst) != generation {
            return; // a newer enable()/disable() owns the status now
        }
        *self.inner.status.lock().unwrap() = s.clone();
        let _ = self.inner.events.send(AdapterEvent::Status(s));
    }

    /// Turn the layer on: spawns the supervisor, which detects, connects,
    /// and reconnects with backoff until `disable()`.
    pub fn enable(&self) {
        if self.is_enabled() {
            return;
        }
        let generation = self.inner.generation.fetch_add(1, Ordering::SeqCst) + 1;
        self.inner.enabled.send_replace(true);
        let this = self.clone();
        spawn_anywhere(async move {
            this.supervisor(generation).await;
        });
        log::info!("freenet adapter enabled (gen {generation})");
    }

    /// The kill switch. Ends the supervisor, drops the session, stops every
    /// retry. Callers see `Disabled` errors immediately.
    pub fn disable(&self) {
        let generation = self.inner.generation.fetch_add(1, Ordering::SeqCst) + 1;
        self.inner.enabled.send_replace(false);
        *self.inner.session.lock().unwrap() = None;
        *self.inner.status.lock().unwrap() = AdapterStatus::Disabled;
        let _ = self.inner.events.send(AdapterEvent::Status(AdapterStatus::Disabled));
        log::info!("freenet adapter disabled (gen {generation})");
    }

    // ------------------------------------------------------------ detect

    /// Three-valued node detection. Usable while disabled (it opens its own
    /// short-lived connection and touches no adapter state).
    pub async fn detect(&self) -> AdapterStatus {
        let port = self.ws_port();
        let addr = format!("127.0.0.1:{port}");
        match timeout(DETECT_TCP_TIMEOUT, tokio::net::TcpStream::connect(&addr)).await {
            Err(_) | Ok(Err(_)) => return AdapterStatus::NodeAbsent,
            Ok(Ok(stream)) => drop(stream),
        }
        let mut api = match timeout(DETECT_WS_TIMEOUT, connect_ws(port)).await {
            Err(_) => return AdapterStatus::NodeOccupied,
            Ok(Err(_)) => return AdapterStatus::NodeOccupied,
            Ok(Ok(api)) => api,
        };
        if api
            .send(ClientRequest::NodeQueries(NodeQuery::ConnectedPeers))
            .await
            .is_err()
        {
            return AdapterStatus::NodeIncompatible { detail: "send failed".into() };
        }
        match timeout(DETECT_QUERY_TIMEOUT, api.recv()).await {
            Ok(Ok(_)) => AdapterStatus::Connected,
            // A decoded, typed refusal IS protocol proof (local mode answers
            // ConnectedPeers with "not supported" — F0).
            Ok(Err(e)) if e.to_string().contains("not supported") => AdapterStatus::Connected,
            Ok(Err(e)) => AdapterStatus::NodeIncompatible { detail: truncate(&e.to_string(), 120) },
            Err(_) => AdapterStatus::NodeIncompatible { detail: "no reply to health query".into() },
        }
    }

    // ------------------------------------------------------------ supervisor

    async fn supervisor(&self, generation: u64) {
        let mut enabled_rx = self.inner.enabled.subscribe();
        loop {
            if self.inner.generation.load(Ordering::SeqCst) != generation || !*enabled_rx.borrow() {
                return;
            }
            self.set_status(generation, AdapterStatus::Detecting);
            let detected = self.detect().await;
            match detected {
                AdapterStatus::Connected => {}
                other => {
                    self.set_status(generation, other);
                    if wait_or_disabled(&mut enabled_rx, SUPERVISOR_BACKOFF).await {
                        return;
                    }
                    continue;
                }
            }
            let port = self.ws_port();
            let api = match timeout(DETECT_WS_TIMEOUT, connect_ws(port)).await {
                Ok(Ok(api)) => api,
                _ => {
                    self.set_status(generation, AdapterStatus::Lost { detail: "connect failed".into() });
                    if wait_or_disabled(&mut enabled_rx, SUPERVISOR_BACKOFF).await {
                        return;
                    }
                    continue;
                }
            };
            let (tx, rx) = mpsc::channel::<Cmd>(CMD_QUEUE);
            *self.inner.session.lock().unwrap() = Some(tx);
            self.set_status(generation, AdapterStatus::Connected);
            log::info!("freenet adapter connected to 127.0.0.1:{port}");
            let reason = self.run_session(api, rx, &mut enabled_rx).await;
            *self.inner.session.lock().unwrap() = None;
            if self.inner.generation.load(Ordering::SeqCst) != generation || !*enabled_rx.borrow() {
                return;
            }
            self.set_status(generation, AdapterStatus::Lost { detail: truncate(&reason, 120) });
            log::warn!("freenet session ended: {}", truncate(&reason, 120));
            if wait_or_disabled(&mut enabled_rx, SUPERVISOR_BACKOFF).await {
                return;
            }
        }
    }

    /// One connection's lifetime. STRICTLY one request in flight (the wire
    /// correlates FIFO); notifications stream out between and during ops.
    async fn run_session(
        &self,
        mut api: WebApi,
        mut rx: mpsc::Receiver<Cmd>,
        enabled_rx: &mut watch::Receiver<bool>,
    ) -> String {
        loop {
            tokio::select! {
                changed = enabled_rx.changed() => {
                    if changed.is_err() || !*enabled_rx.borrow() {
                        return "disabled".into();
                    }
                }
                maybe_cmd = rx.recv() => {
                    let Some(cmd) = maybe_cmd else { return "handle dropped".into() };
                    if api.send(cmd.req).await.is_err() {
                        let _ = cmd.reply.send(Err(AdapterError::Node("send failed".into())));
                        return "send failed".into();
                    }
                    // Await THIS op's response; unsolicited notifications
                    // arriving in between are forwarded, not misdelivered.
                    let deadline = tokio::time::Instant::now() + cmd.op_timeout;
                    loop {
                        let left = deadline.saturating_duration_since(tokio::time::Instant::now());
                        if left.is_zero() {
                            let _ = cmd.reply.send(Err(AdapterError::Timeout));
                            break;
                        }
                        match timeout(left, api.recv()).await {
                            Err(_) => {
                                let _ = cmd.reply.send(Err(AdapterError::Timeout));
                                break;
                            }
                            Ok(Err(e)) => {
                                let _ = cmd.reply.send(Err(AdapterError::Node(e.to_string())));
                                break;
                            }
                            Ok(Ok(HostResponse::ContractResponse(
                                ContractResponse::UpdateNotification { key, update },
                            ))) => {
                                self.forward_notification(&key, &update);
                                continue;
                            }
                            Ok(Ok(resp)) => {
                                let _ = cmd.reply.send(Ok(resp));
                                break;
                            }
                        }
                    }
                }
                streamed = api.recv() => {
                    // Idle traffic: subscription notifications, or the end
                    // of the connection.
                    match streamed {
                        Ok(HostResponse::ContractResponse(
                            ContractResponse::UpdateNotification { key, update },
                        )) => self.forward_notification(&key, &update),
                        Ok(_) => { /* stray late response; correlation is FIFO, drop */ }
                        Err(e) => return e.to_string(),
                    }
                }
            }
        }
    }

    fn forward_notification(&self, key: &ContractKey, update: &UpdateData<'static>) {
        let len = match update {
            UpdateData::State(s) => s.as_ref().len(),
            UpdateData::Delta(d) => d.as_ref().len(),
            UpdateData::StateAndDelta { state, delta } => state.as_ref().len() + delta.as_ref().len(),
            _ => 0,
        };
        // Bounded broadcast: lagging receivers lose oldest events. Honest.
        let _ = self.inner.events.send(AdapterEvent::ContractUpdate {
            instance: key.id().encode(),
            update_len: len,
        });
    }

    // ------------------------------------------------------------ ops

    async fn request(&self, req: ClientRequest<'static>, op_timeout: Duration) -> Result<HostResponse> {
        if !self.is_enabled() {
            return Err(AdapterError::Disabled);
        }
        let tx = {
            let guard = self.inner.session.lock().unwrap();
            guard.clone()
        };
        let Some(tx) = tx else {
            return Err(AdapterError::NotConnected(self.status()));
        };
        let (reply_tx, reply_rx) = oneshot::channel();
        tx.send(Cmd { req, reply: reply_tx, op_timeout })
            .await
            .map_err(|_| AdapterError::ShuttingDown)?;
        reply_rx.await.map_err(|_| AdapterError::ShuttingDown)?
    }

    fn container(code: Vec<u8>, params: &[u8]) -> (ContractContainer, ContractKey) {
        let params = Parameters::from(params.to_vec()).into_owned();
        let c = ContractContainer::Wasm(ContractWasmAPIVersion::V1(WrappedContract::new(
            Arc::new(ContractCode::from(code)),
            params,
        )));
        let k = c.key();
        (c, k)
    }

    /// Compute the ids a (code, params) pair addresses — no network.
    pub fn contract_ids(code: Vec<u8>, params: &[u8]) -> ContractIds {
        let (_, key) = Self::container(code, params);
        ContractIds {
            instance: key.id().encode(),
            code_hash: key.encoded_code_hash(),
        }
    }

    pub async fn contract_put(
        &self,
        code: Vec<u8>,
        params: &[u8],
        state: Vec<u8>,
        subscribe: bool,
    ) -> Result<ContractIds> {
        let (container, key) = Self::container(code, params);
        let ids = ContractIds {
            instance: key.id().encode(),
            code_hash: key.encoded_code_hash(),
        };
        let resp = self
            .request(
                ContractRequest::Put {
                    contract: container,
                    state: WrappedState::new(state),
                    related_contracts: RelatedContracts::new(),
                    subscribe,
                    blocking_subscribe: false,
                }
                .into(),
                PUT_TIMEOUT,
            )
            .await?;
        match resp {
            HostResponse::ContractResponse(ContractResponse::PutResponse { .. }) => Ok(ids),
            other => Err(unexpected("PutResponse", &other)),
        }
    }

    pub async fn contract_get(
        &self,
        instance_b58: &str,
        subscribe: bool,
    ) -> Result<Vec<u8>> {
        let id = parse_instance(instance_b58)?;
        let resp = self
            .request(
                ContractRequest::Get {
                    key: id,
                    return_contract_code: false,
                    subscribe,
                    blocking_subscribe: false,
                }
                .into(),
                OP_TIMEOUT,
            )
            .await
            // Depending on version, a never-put contract comes back either
            // as a typed NotFound or as an error frame "missing contract:
            // <id>" (observed on 0.2.132 local mode). Fold both into one
            // shape so services can match on "contract not found" alone.
            .map_err(|e| match e {
                AdapterError::Node(msg) if msg.contains("missing contract") => {
                    AdapterError::Node("contract not found".into())
                }
                e => e,
            })?;
        match resp {
            HostResponse::ContractResponse(ContractResponse::GetResponse { state, .. }) => {
                Ok(state.to_vec())
            }
            HostResponse::ContractResponse(ContractResponse::NotFound { .. }) => {
                Err(AdapterError::Node("contract not found".into()))
            }
            other => Err(unexpected("GetResponse", &other)),
        }
    }

    pub async fn contract_subscribe(&self, instance_b58: &str) -> Result<bool> {
        let id = parse_instance(instance_b58)?;
        let resp = self
            .request(
                ContractRequest::Subscribe { key: id, summary: None }.into(),
                OP_TIMEOUT,
            )
            .await?;
        match resp {
            HostResponse::ContractResponse(ContractResponse::SubscribeResponse {
                subscribed,
                ..
            }) => Ok(subscribed),
            other => Err(unexpected("SubscribeResponse", &other)),
        }
    }

    /// UPDATE with the recorded first-write behavior handled: a node that
    /// never held the contract answers "auto-fetch triggered" and asks the
    /// client to retry; we do, on the spike's schedule. Returns the
    /// post-merge summary bytes.
    pub async fn contract_update(
        &self,
        code: Vec<u8>,
        params: &[u8],
        delta: Vec<u8>,
    ) -> Result<Vec<u8>> {
        let (_, key) = Self::container(code, params);
        let mut last: Option<AdapterError> = None;
        for attempt in 0..AUTOFETCH_RETRIES {
            if attempt > 0 {
                tokio::time::sleep(AUTOFETCH_BACKOFF).await;
                if !self.is_enabled() {
                    return Err(AdapterError::Disabled); // kill switch ends retries
                }
            }
            let resp = self
                .request(
                    ContractRequest::Update {
                        key,
                        data: UpdateData::Delta(StateDelta::from(delta.clone())),
                    }
                    .into(),
                    OP_TIMEOUT,
                )
                .await;
            match resp {
                Ok(HostResponse::ContractResponse(ContractResponse::UpdateResponse {
                    summary,
                    ..
                })) => return Ok(summary.as_ref().to_vec()),
                Ok(other) => return Err(unexpected("UpdateResponse", &other)),
                Err(AdapterError::Node(msg)) if msg.contains("auto-fetch triggered") => {
                    log::info!(
                        "freenet: first-write priming on attempt {attempt} (auto-fetch in flight)"
                    );
                    last = Some(AdapterError::Node(msg));
                }
                Err(e) => return Err(e),
            }
        }
        Err(last.unwrap_or(AdapterError::Timeout))
    }

    /// The reliable first-write recipe from F5: one GET primes code+state on
    /// this node, then the update applies. Use for a contract this node may
    /// never have held.
    pub async fn get_then_update(
        &self,
        code: Vec<u8>,
        params: &[u8],
        delta: Vec<u8>,
    ) -> Result<Vec<u8>> {
        let ids = Self::contract_ids(code.clone(), params);
        let _ = self.contract_get(&ids.instance, false).await; // prime; absence is fine
        self.contract_update(code, params, delta).await
    }

    // ------------------------------------------------------------ delegates

    fn delegate_container(code: Vec<u8>, params: &[u8]) -> (DelegateContainer, DelegateKey) {
        let params = Parameters::from(params.to_vec()).into_owned();
        let dc = DelegateCode::from(code);
        let d = Delegate::from((&dc, &params));
        let c = DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(d));
        let k = c.key().clone();
        (c, k)
    }

    /// Compute the delegate key a (code, params) pair addresses — no
    /// network. Lets services pin and log their delegate generation.
    pub fn delegate_key_b58(code: Vec<u8>, params: &[u8]) -> String {
        let (_, key) = Self::delegate_container(code, params);
        bs58::encode(key.bytes()).into_string()
    }

    /// Register (idempotent per node) and return the delegate key (b58).
    /// `cipher`/`nonce` are legacy fields the node ignores (core#4140).
    pub async fn delegate_register(&self, code: Vec<u8>, params: &[u8]) -> Result<String> {
        use freenet_stdlib::client_api::DelegateRequest;
        let (container, key) = Self::delegate_container(code, params);
        let resp = self
            .request(
                ClientRequest::DelegateOp(DelegateRequest::RegisterDelegate {
                    delegate: container,
                    cipher: [0u8; 32],
                    nonce: [0u8; 24],
                }),
                OP_TIMEOUT,
            )
            .await?;
        match resp {
            HostResponse::DelegateResponse { .. } => Ok(bs58::encode(key.bytes()).into_string()),
            other => Err(unexpected("DelegateResponse", &other)),
        }
    }

    /// One application message in, the delegate's app-message reply out.
    /// NOTE (F1): migration across delegate code generations must be
    /// verified by reading state back — `RegisterDelegateWithPredecessors`
    /// silently copies nothing in 0.2.132 (GHSA-824h-7x5x-wfmf).
    pub async fn delegate_message(
        &self,
        code: Vec<u8>,
        params: &[u8],
        payload: Vec<u8>,
    ) -> Result<Vec<u8>> {
        use freenet_stdlib::client_api::DelegateRequest;
        let (_, key) = Self::delegate_container(code.clone(), params);
        let params_owned = Parameters::from(params.to_vec()).into_owned();
        let resp = self
            .request(
                ClientRequest::DelegateOp(DelegateRequest::ApplicationMessages {
                    key,
                    params: params_owned,
                    inbound: vec![InboundDelegateMsg::ApplicationMessage(
                        ApplicationMessage::new(payload),
                    )],
                }),
                OP_TIMEOUT,
            )
            .await?;
        match resp {
            HostResponse::DelegateResponse { values, .. } => values
                .into_iter()
                .find_map(|v| match v {
                    OutboundDelegateMsg::ApplicationMessage(m) => Some(m.payload),
                    _ => None,
                })
                .ok_or_else(|| AdapterError::Node("no app message in delegate reply".into())),
            other => Err(unexpected("DelegateResponse", &other)),
        }
    }
}

// ------------------------------------------------------------------ helpers

async fn connect_ws(port: u16) -> std::result::Result<WebApi, String> {
    let uri = format!("ws://127.0.0.1:{port}/v1/contract/command?encodingProtocol=native");
    let cfg = tokio_tungstenite::tungstenite::protocol::WebSocketConfig::default()
        .max_message_size(Some(100 * 1024 * 1024))
        .max_frame_size(Some(16 * 1024 * 1024));
    let (stream, _) = tokio_tungstenite::connect_async_with_config(&uri, Some(cfg), false)
        .await
        .map_err(|e| e.to_string())?;
    Ok(WebApi::start(stream))
}

/// Wait out a backoff, returning true if the adapter was disabled meanwhile.
async fn wait_or_disabled(enabled_rx: &mut watch::Receiver<bool>, dur: Duration) -> bool {
    tokio::select! {
        _ = tokio::time::sleep(dur) => !*enabled_rx.borrow(),
        changed = enabled_rx.changed() => changed.is_err() || !*enabled_rx.borrow(),
    }
}

fn parse_instance(b58: &str) -> Result<ContractInstanceId> {
    ContractInstanceId::try_from(b58.to_string())
        .map_err(|e| AdapterError::Other(format!("bad instance id: {e}")))
}

fn unexpected(wanted: &str, got: &HostResponse) -> AdapterError {
    // Display, not Debug: HostResponse Display is summary-shaped and does
    // not dump state bytes.
    AdapterError::Node(format!("expected {wanted}, got {}", truncate(&got.to_string(), 120)))
}

fn truncate(s: &str, n: usize) -> String {
    if s.len() <= n {
        return s.to_string();
    }
    let mut cut = n;
    while cut > 0 && !s.is_char_boundary(cut) {
        cut -= 1;
    }
    format!("{}…", &s[..cut])
}

// ------------------------------------------------------------------ tests

#[cfg(test)]
mod tests {
    use super::*;

    /// The event-loop path: no tokio context at all. Must not panic, must
    /// flip enabled, and disable() must still end it.
    #[test]
    fn enable_works_outside_any_runtime() {
        std::thread::spawn(|| {
            assert!(tokio::runtime::Handle::try_current().is_err());
            let a = FreenetAdapter::default();
            a.enable();
            assert!(a.is_enabled());
            std::thread::sleep(std::time::Duration::from_millis(300));
            a.disable();
            assert!(!a.is_enabled());
            assert_eq!(a.status(), AdapterStatus::Disabled);
        })
        .join()
        .expect("no panic on the runtime-less path");
    }

    #[tokio::test]
    async fn disabled_is_the_default_and_ops_refuse() {
        let a = FreenetAdapter::default();
        assert_eq!(a.status(), AdapterStatus::Disabled);
        assert!(!a.is_enabled());
        let err = a.contract_get("11111111111111111111111111111111", false).await;
        assert!(matches!(err, Err(AdapterError::Disabled)));
    }

    #[tokio::test]
    async fn detect_absent_on_closed_port() {
        let a = FreenetAdapter::default();
        a.set_ws_port(1); // nothing listens on port 1
        assert_eq!(a.detect().await, AdapterStatus::NodeAbsent);
    }

    #[tokio::test]
    async fn detect_occupied_on_non_api_listener() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        tokio::spawn(async move {
            loop {
                let Ok((mut sock, _)) = listener.accept().await else { break };
                use tokio::io::AsyncWriteExt;
                // Answer like a confused HTTP server, then hang up.
                let _ = sock.write_all(b"HTTP/1.1 404 Not Found\r\n\r\n").await;
            }
        });
        let a = FreenetAdapter::default();
        a.set_ws_port(port);
        assert_eq!(a.detect().await, AdapterStatus::NodeOccupied);
    }

    #[tokio::test]
    async fn kill_switch_ends_everything() {
        let a = FreenetAdapter::default();
        a.set_ws_port(1);
        a.enable();
        assert!(a.is_enabled());
        // Give the supervisor a moment to enter its loop, then kill it.
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.disable();
        assert_eq!(a.status(), AdapterStatus::Disabled);
        assert!(!a.is_enabled());
        let err = a.contract_get("11111111111111111111111111111111", false).await;
        assert!(matches!(err, Err(AdapterError::Disabled)));
        // A stale supervisor tick must not resurrect a status.
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(a.status(), AdapterStatus::Disabled);
    }

    #[tokio::test]
    async fn contract_ids_are_deterministic_offline() {
        let code = vec![0u8, 1, 2, 3];
        let a = FreenetAdapter::contract_ids(code.clone(), b"params");
        let b = FreenetAdapter::contract_ids(code, b"params");
        assert_eq!(a.instance, b.instance);
        assert_eq!(a.code_hash, b.code_hash);
    }
}