tauri_plugin_phyto/

runtime.rs

use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use tauri::{
    plugin::{Builder, TauriPlugin},
    Manager, Runtime, WebviewWindow,
};
use tiny_http::{Header, Response, Server};
use uuid::Uuid;

/// Wire protocol version. Kept manually in lockstep with the
/// `PROTOCOL_VERSION` constants in `crates/phyto-core/src/protocol.rs` and
/// the TS packages. The driver fetches this via `GET /info` and exits cleanly
/// on mismatch — see `packages/driver-tauri/src/index.ts`. Bump only on
/// breaking wire-format changes.
pub const PROTOCOL_VERSION: u32 = 1;

/// `Cargo.toml`-defined version of this crate. Returned alongside
/// `protocol_version` on `GET /info` so the driver can surface it in error
/// messages when versions mismatch.
const PLUGIN_VERSION: &str = env!("CARGO_PKG_VERSION");

/// Configuration for the Phyto automation server.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PhytoConfig {
    /// Port for the HTTP automation server (default: 9876)
    #[serde(default = "default_port")]
    pub port: u16,
}

impl Default for PhytoConfig {
    fn default() -> Self {
        Self {
            port: default_port(),
        }
    }
}

fn default_port() -> u16 {
    9876
}

/// Response body for command results
#[derive(Debug, Serialize)]
struct CommandResponse {
    ok: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    value: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    error: Option<String>,
}

/// Callback result received from the webview via Tauri IPC
#[derive(Debug, Clone, Deserialize)]
struct CallbackResult {
    ok: bool,
    #[serde(default)]
    value: Option<serde_json::Value>,
    #[serde(default)]
    error: Option<String>,
}

/// Shared state for pending command callbacks.
/// Wrapped in Arc so both the HTTP server thread and Tauri command can share it.
#[derive(Clone)]
pub struct PendingCallbacks {
    inner: Arc<Mutex<HashMap<String, std::sync::mpsc::Sender<CallbackResult>>>>,
}

impl PendingCallbacks {
    fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    fn insert(&self, id: String, sender: std::sync::mpsc::Sender<CallbackResult>) {
        self.inner.lock().unwrap().insert(id, sender);
    }

    fn remove(&self, id: &str) {
        self.inner.lock().unwrap().remove(id);
    }

    fn send(&self, id: &str, result: CallbackResult) {
        let callbacks = self.inner.lock().unwrap();
        if let Some(sender) = callbacks.get(id) {
            let _ = sender.send(result);
        }
    }
}

/// Shared readiness flag. Managed as Tauri state so both the HTTP server
/// thread and the `signal_ready` IPC command can access it.
#[derive(Clone)]
struct ReadinessFlag(Arc<AtomicBool>);

/// Per-condition probe state, used to produce a targeted remediation message
/// when the readiness probe times out.
///
/// - `loc_ok` is updated from Rust each tick (via `webview.url()`), since the
///   webview handle is held by the probe thread.
/// - `tauri_ok` and `harness_ok` are reported from JS via the
///   `report_probe_state` IPC command, since they can only be observed inside
///   the page.
/// - `failure_reason` is set once on timeout and read by `/health` so the
///   external driver sees an actionable error instead of a perpetual
///   `"loading"`.
#[derive(Clone)]
struct ProbeState {
    loc_ok: Arc<AtomicBool>,
    tauri_ok: Arc<AtomicBool>,
    harness_ok: Arc<AtomicBool>,
    failure_reason: Arc<Mutex<Option<String>>>,
}

impl ProbeState {
    fn new() -> Self {
        Self {
            loc_ok: Arc::new(AtomicBool::new(false)),
            tauri_ok: Arc::new(AtomicBool::new(false)),
            harness_ok: Arc::new(AtomicBool::new(false)),
            failure_reason: Arc::new(Mutex::new(None)),
        }
    }
}

// ─── Tauri command: receives command results from the webview via IPC ───

#[tauri::command]
fn eval_callback(
    state: tauri::State<'_, PendingCallbacks>,
    id: String,
    ok: bool,
    value: Option<serde_json::Value>,
    error: Option<String>,
) -> Result<(), String> {
    let result = CallbackResult { ok, value, error };
    state.send(&id, result);
    Ok(())
}

/// Tauri IPC command called by the readiness probe script.
/// Uses Tauri IPC instead of HTTP fetch to avoid mixed-content blocks
/// when the app runs under `tauri://localhost` (custom-protocol).
#[tauri::command]
fn signal_ready(state: tauri::State<'_, ReadinessFlag>) -> Result<(), String> {
    log::info!("[phyto] Readiness signal received via IPC — marking ready");
    state.0.store(true, Ordering::SeqCst);
    Ok(())
}

/// Tauri IPC command called by the readiness probe script each tick to report
/// the two JS-observable conditions. `loc_ok` is computed Rust-side from
/// `webview.url()`, so it isn't an argument here.
///
/// This lets `start_server` format a targeted remediation message on probe
/// timeout (e.g. "Phyto harness never loaded — confirm @phyto/vite-plugin is
/// wired up") instead of a generic "probe timed out".
#[tauri::command]
fn report_probe_state(
    state: tauri::State<'_, ProbeState>,
    tauri_ok: bool,
    harness_ok: bool,
) -> Result<(), String> {
    state.tauri_ok.store(tauri_ok, Ordering::SeqCst);
    state.harness_ok.store(harness_ok, Ordering::SeqCst);
    Ok(())
}

// ─── HTTP helpers ───────────────────────────────────────────────────

fn cors_headers() -> Vec<Header> {
    vec![
        Header::from_bytes("Access-Control-Allow-Origin", "*").unwrap(),
        Header::from_bytes("Access-Control-Allow-Methods", "GET, POST, OPTIONS").unwrap(),
        Header::from_bytes("Access-Control-Allow-Headers", "Content-Type").unwrap(),
        Header::from_bytes("Content-Type", "application/json").unwrap(),
    ]
}

fn json_response(status: u16, body: &str) -> Response<std::io::Cursor<Vec<u8>>> {
    let data = body.as_bytes().to_vec();
    let mut response = Response::from_data(data).with_status_code(status);
    for header in cors_headers() {
        response.add_header(header);
    }
    response
}

/// Start the HTTP automation server in a background thread.
///
/// The server includes an IPC readiness probe: before `/health` returns 200,
/// it verifies that the webview's Tauri IPC bridge is functional by sending
/// a test eval and waiting for the callback. This prevents the harness from
/// thinking the app is ready when the page hasn't fully loaded yet
/// (common under QEMU + Xvfb where WebKitGTK init is slow).
fn start_server<R: Runtime>(
    ipc_ready: Arc<AtomicBool>,
    pending: PendingCallbacks,
    probe_state: ProbeState,
    webview: WebviewWindow<R>,
) {
    let port = {
        let config = webview.state::<PhytoConfig>();
        config.port
    };

    let addr = format!("0.0.0.0:{}", port);
    let server = match Server::http(&addr) {
        Ok(s) => {
            log::info!("[phyto] Automation server listening on http://localhost:{}", port);
            s
        }
        Err(e) => {
            log::error!("[phyto] Failed to start automation server on {}: {}", addr, e);
            return;
        }
    };

    // Readiness probe thread: injects JS into the webview that signals
    // readiness via Tauri IPC (not HTTP fetch, which is blocked by
    // mixed-content policies when the app uses custom-protocol / tauri://).
    //
    // The probe also tracks per-condition state so that on timeout we can
    // surface a targeted remediation message (e.g. "webview stuck on
    // about:blank — typically means tauri/custom-protocol isn't enabled")
    // instead of a generic "timed out". `loc_ok` is observed Rust-side via
    // `webview.url()`; `tauri_ok` and `harness_ok` are reported from JS via
    // the `report_probe_state` IPC command.
    {
        let ipc_ready = ipc_ready.clone();
        let webview = webview.clone();
        let probe_state = probe_state.clone();
        thread::spawn(move || {
            let start = Instant::now();
            let deadline = start + Duration::from_secs(120);
            let mut probe_count = 0u32;
            let mut warned_about_blank = false;
            while Instant::now() < deadline {
                // Observe loc_ok from Rust: cheaper and more direct than
                // round-tripping through JS, and works even before the page
                // can run script.
                let loc_ok = webview
                    .url()
                    .ok()
                    .map(|u| u.as_str() != "about:blank")
                    .unwrap_or(false);
                probe_state.loc_ok.store(loc_ok, Ordering::SeqCst);

                // JS reports the two conditions only it can see, then signals
                // readiness when both the IPC bridge and the harness are up.
                let script = format!(
                    r#"try {{
  var __phyto_tauri = !!window.__TAURI_INTERNALS__;
  var __phyto_harness = !!window.__phyto_harness__;
  if ({probe_count} % 10 === 0) console.log('[phyto-probe] tauri=' + __phyto_tauri + ' harness=' + __phyto_harness);
  if (__phyto_tauri) {{
    window.__TAURI_INTERNALS__.invoke('plugin:phyto|report_probe_state', {{ tauri_ok: __phyto_tauri, harness_ok: __phyto_harness }});
    if (__phyto_harness) {{
      window.__TAURI_INTERNALS__.invoke('plugin:phyto|signal_ready');
    }}
  }}
}} catch(e) {{ if ({probe_count} % 10 === 0) console.log('[phyto-probe] error: ' + e.message); }}"#,
                    probe_count = probe_count,
                );

                let _ = webview.eval(&script);
                probe_count += 1;

                // Early warning: if we're still on about:blank at ~10s, log a
                // targeted hint immediately so the failure mode is obvious in
                // the log stream long before the 120s timeout fires.
                if !warned_about_blank
                    && !loc_ok
                    && start.elapsed() >= Duration::from_secs(10)
                {
                    log::warn!(
                        "[phyto] Readiness probe: webview still on about:blank after 10s — \
                         typically means tauri/custom-protocol isn't enabled in your test \
                         feature. Add it directly, or upgrade tauri-plugin-phyto to a version \
                         that pulls it in transitively."
                    );
                    warned_about_blank = true;
                }

                thread::sleep(Duration::from_millis(500));

                if ipc_ready.load(Ordering::SeqCst) {
                    log::info!("[phyto] Readiness probe succeeded after {} probes", probe_count);
                    return;
                }
            }

            // Timeout: pick a targeted remediation message from the last
            // observed state and surface it via both the log and /health.
            let loc_ok = probe_state.loc_ok.load(Ordering::SeqCst);
            let tauri_ok = probe_state.tauri_ok.load(Ordering::SeqCst);
            let harness_ok = probe_state.harness_ok.load(Ordering::SeqCst);
            let reason: &'static str = if !loc_ok {
                "webview stuck on about:blank — typically means tauri/custom-protocol isn't \
                 enabled in your test feature. Add it directly, or upgrade tauri-plugin-phyto \
                 to a version that pulls it in transitively."
            } else if !tauri_ok {
                "Tauri IPC bridge never initialized — check that the main window finished \
                 loading."
            } else if !harness_ok {
                "Phyto harness never loaded — confirm @phyto/vite-plugin is wired up in your \
                 vite config."
            } else {
                // All three flipped true but ipc_ready never did. Shouldn't
                // happen in practice (JS invokes signal_ready in the same
                // tick it sees both flags), but cover it anyway.
                "all probe conditions met but signal_ready never landed — Tauri IPC may be \
                 dropping invocations."
            };
            log::error!(
                "[phyto] Readiness probe timed out after 120s ({} probes): {}",
                probe_count,
                reason
            );
            *probe_state.failure_reason.lock().unwrap() = Some(reason.to_string());
        });
    }

    thread::spawn(move || {
        for mut request in server.incoming_requests() {
            let method_str = request.method().to_string();
            let url = request.url().to_string();

            // Handle CORS preflight
            if method_str == "OPTIONS" {
                let _ = request.respond(json_response(204, ""));
                continue;
            }

            // POST /__ready_probe — legacy HTTP-based readiness signal (kept for
            // apps that serve from http:// origins where fetch works fine).
            if method_str == "POST" && url == "/__ready_probe" {
                let mut probe_body = String::new();
                let _ = request.as_reader().read_to_string(&mut probe_body);

                let is_loaded = probe_body.contains("\"loaded\":true");
                if is_loaded {
                    log::info!("[phyto] Readiness probe received loaded=true via HTTP — marking ready");
                    ipc_ready.store(true, Ordering::SeqCst);
                }
                let _ = request.respond(json_response(200, r#"{"ok":true}"#));
                continue;
            }

            // GET /health — only returns 200 once webview JS context is confirmed working.
            // Once the probe has timed out and stored a `failure_reason`,
            // `/health` switches from "loading" (which the driver would keep
            // polling) to "failed" + reason, so callers get an actionable
            // error instead of a perpetual loading state.
            if method_str == "GET" && url == "/health" {
                let failure = probe_state.failure_reason.lock().unwrap().clone();
                if let Some(reason) = failure {
                    let body = serde_json::json!({
                        "status": "failed",
                        "reason": reason,
                    })
                    .to_string();
                    let _ = request.respond(json_response(503, &body));
                } else if ipc_ready.load(Ordering::SeqCst) {
                    let body = serde_json::json!({ "status": "ok" }).to_string();
                    let _ = request.respond(json_response(200, &body));
                } else {
                    let body = serde_json::json!({ "status": "loading" }).to_string();
                    let _ = request.respond(json_response(503, &body));
                }
                continue;
            }

            // GET /info — returns wire-protocol and plugin metadata so the
            // external driver can verify compatibility before issuing any
            // commands. Always responds (no readiness gate) — the protocol
            // version doesn't change at runtime.
            if method_str == "GET" && url == "/info" {
                let body = serde_json::json!({
                    "protocol_version": PROTOCOL_VERSION,
                    "plugin_version": PLUGIN_VERSION,
                    "plugin": "tauri-plugin-phyto",
                })
                .to_string();
                let _ = request.respond(json_response(200, &body));
                continue;
            }

            // POST /command — forward a declarative command to the in-page harness
            if method_str == "POST" && url == "/command" {
                let mut body = String::new();
                if request.as_reader().read_to_string(&mut body).is_err() {
                    let _ = request.respond(json_response(
                        400,
                        r#"{"ok":false,"error":"Failed to read request body"}"#,
                    ));
                    continue;
                }

                // Validate it's valid JSON (the harness will handle the rest)
                let command_json: serde_json::Value = match serde_json::from_str(&body) {
                    Ok(v) => v,
                    Err(e) => {
                        let err = serde_json::json!({
                            "ok": false,
                            "error": format!("Invalid JSON: {}", e)
                        });
                        let _ = request.respond(json_response(400, &err.to_string()));
                        continue;
                    }
                };

                // Build a script that calls the harness's execute() method
                // and sends the result back via the existing IPC callback channel
                let callback_id = Uuid::new_v4().to_string();
                let (tx, rx) = std::sync::mpsc::channel::<CallbackResult>();
                pending.insert(callback_id.clone(), tx);

                let command_str = serde_json::to_string(&command_json).unwrap();
                let wrapped_script = format!(
                    r#"(async () => {{
    const __phyto_id = "{}";
    try {{
        if (!window.__phyto_harness__) {{
            throw new Error("Phyto harness not available — is the vite plugin installed?");
        }}
        const __phyto_result = await window.__phyto_harness__.execute({});
        await window.__TAURI_INTERNALS__.invoke('plugin:phyto|eval_callback', {{
            id: __phyto_id,
            ok: true,
            value: __phyto_result,
            error: null
        }});
    }} catch (__phyto_err) {{
        await window.__TAURI_INTERNALS__.invoke('plugin:phyto|eval_callback', {{
            id: __phyto_id,
            ok: false,
            value: null,
            error: __phyto_err.message || String(__phyto_err)
        }});
    }}
}})()"#,
                    callback_id, command_str
                );

                let webview_clone = webview.clone();
                let eval_result = webview_clone.eval(&wrapped_script);

                if let Err(e) = eval_result {
                    pending.remove(&callback_id);
                    let err = serde_json::json!({
                        "ok": false,
                        "error": format!("Failed to evaluate command in webview: {}", e)
                    });
                    let _ = request.respond(json_response(500, &err.to_string()));
                    continue;
                }

                match rx.recv_timeout(std::time::Duration::from_secs(30)) {
                    Ok(result) => {
                        pending.remove(&callback_id);
                        let response = CommandResponse {
                            ok: result.ok,
                            value: result.value,
                            error: result.error,
                        };
                        let body = serde_json::to_string(&response).unwrap();
                        let _ = request.respond(json_response(200, &body));
                    }
                    Err(_) => {
                        pending.remove(&callback_id);
                        let err = serde_json::json!({
                            "ok": false,
                            "error": "Command timed out after 30 seconds"
                        });
                        let _ = request.respond(json_response(500, &err.to_string()));
                    }
                }
                continue;
            }

            // 404 for everything else
            let _ = request.respond(json_response(
                404,
                r#"{"ok":false,"error":"Not found"}"#,
            ));
        }
    });
}

/// Initialize the Phyto plugin.
///
/// # Usage in your Tauri app
///
/// ```rust,ignore
/// tauri::Builder::default()
///     .plugin(tauri_plugin_phyto::init(Default::default()))
/// ```
pub fn init<R: Runtime>(config: PhytoConfig) -> TauriPlugin<R> {
    Builder::new("phyto")
        .invoke_handler(tauri::generate_handler![
            eval_callback,
            signal_ready,
            report_probe_state,
        ])
        .setup(move |app, _api| {
            let pending = PendingCallbacks::new();
            let ipc_ready = Arc::new(AtomicBool::new(false));
            let probe_state = ProbeState::new();

            // Share state so Tauri commands and HTTP server can access it
            app.manage(pending.clone());
            app.manage(ReadinessFlag(ipc_ready.clone()));
            app.manage(probe_state.clone());
            app.manage(config.clone());

            let app_handle = app.clone();
            let pending_clone = pending.clone();
            let probe_state_clone = probe_state.clone();

            // Start the HTTP server once the main window is ready.
            // Poll for the window with retries to handle slow WebKitGTK init
            // (e.g. inside QEMU + Xvfb where startup can take several seconds).
            thread::spawn(move || {
                let deadline = Instant::now() + Duration::from_secs(10);
                loop {
                    if let Some(window) = app_handle.get_webview_window("main") {
                        start_server(ipc_ready, pending_clone, probe_state_clone, window);
                        return;
                    }
                    // Fallback: try any available window
                    let windows = app_handle.webview_windows();
                    if let Some((_label, window)) = windows.into_iter().next() {
                        start_server(ipc_ready, pending_clone, probe_state_clone, window);
                        return;
                    }
                    if Instant::now() >= deadline {
                        log::error!("[phyto] No webview window found after 10s — automation server not started");
                        return;
                    }
                    thread::sleep(Duration::from_millis(250));
                }
            });

            Ok(())
        })
        .build()
}