spaceshell/app/src-tauri/src/bridge.rs

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use base64::Engine;
use serde_json::Value;
use spacesh_proto::codec::{read_frame, write_frame};
use spacesh_proto::message::{SplitDir, Edge, PresetSlot};
use spacesh_proto::ids::{GroupId, WorkspaceId};
use spacesh_proto::{Cmd, Envelope, Evt, SurfaceId};
use tauri::ipc::Channel;
use tauri::{AppHandle, Emitter};
use tokio::net::UnixStream;
use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf};
use tokio::sync::{mpsc, oneshot, Mutex};

pub struct Bridge {
    next_id: AtomicU64,
    /// For respawning/reconnecting the daemon connection after it drops.
    app: AppHandle,
    sock: PathBuf,
    /// Bumped on every successful reconnect; lets concurrent failing requests
    /// collapse into a single reconnect (single-flight).
    gen: AtomicU64,
    /// Outbound frames to the daemon. Swapped on reconnect.
    tx: Mutex<mpsc::Sender<Envelope>>,
    /// Serializes reconnect attempts.
    reconnect_lock: Mutex<()>,
    /// The current reader task; aborted on reconnect so a stale connection can't
    /// keep delivering duplicate output (which doubled keystroke echo).
    reader: Mutex<tokio::task::JoinHandle<()>>,
    /// Pending request id → reply slot.
    pending: Arc<Mutex<HashMap<u64, oneshot::Sender<Envelope>>>>,
    /// surface id → output channel into the webview.
    out_channels: Arc<Mutex<HashMap<String, Channel<Vec<u8>>>>>,
}

fn socket_path() -> Result<PathBuf> {
    // Honor SPACESH_SOCK so the GUI matches a daemon/CLI started with the same
    // override (mirrors the daemon's lifecycle::socket_path and the CLI client).
    if let Ok(p) = std::env::var("SPACESH_SOCK") {
        if !p.is_empty() {
            return Ok(PathBuf::from(p));
        }
    }
    Ok(dirs::home_dir().context("no home")?.join(".spacesh").join("sock"))
}

/// Locate the `spaceshd` binary. The Tauri app is its own cargo workspace, so in
/// `tauri dev` the app binary lives in `app/src-tauri/target/debug/` while the
/// daemon is built into the repo-root `target/debug/` — they are NOT siblings.
/// Try, in order: SPACESHD_BIN override, a sibling (release/bundled layout),
/// the repo-root dev/release target relative to the app binary, then PATH.
fn find_daemon() -> PathBuf {
    if let Ok(p) = std::env::var("SPACESHD_BIN") {
        if !p.is_empty() {
            return PathBuf::from(p);
        }
    }
    if let Ok(exe) = std::env::current_exe() {
        let sibling = exe.with_file_name("spaceshd");
        if sibling.exists() {
            return sibling;
        }
        if let Some(dir) = exe.parent() {
            // app/src-tauri/target/{debug,release} → repo-root target/{debug,release}
            for rel in ["../../../../target/debug/spaceshd", "../../../../target/release/spaceshd"] {
                let cand = dir.join(rel);
                if cand.exists() {
                    return cand;
                }
            }
        }
    }
    PathBuf::from("spaceshd") // last resort: rely on PATH
}

async fn ensure_daemon(sock: &PathBuf) -> Result<UnixStream> {
    if let Ok(s) = UnixStream::connect(sock).await {
        return Ok(s);
    }
    // Lazy start: spawn the daemon binary, then poll for the socket.
    let daemon = find_daemon();
    match std::process::Command::new(&daemon).spawn() {
        Ok(_) => {}
        Err(e) => anyhow::bail!("could not spawn daemon at {}: {e}", daemon.display()),
    }
    for _ in 0..100 {
        if let Ok(s) = UnixStream::connect(sock).await {
            return Ok(s);
        }
        tokio::time::sleep(tokio::time::Duration::from_millis(30)).await;
    }
    anyhow::bail!("daemon spawned ({}) but did not bind {} in time", daemon.display(), sock.display())
}

/// Connect (spawning the daemon if needed) and start the reader/writer tasks,
/// returning the outbound sender. Shared `pending`/`out_channels` are reused so
/// replies and live output keep routing across reconnects.
async fn spawn_connection(
    sock: &PathBuf,
    app: &AppHandle,
    pending: Arc<Mutex<HashMap<u64, oneshot::Sender<Envelope>>>>,
    out_channels: Arc<Mutex<HashMap<String, Channel<Vec<u8>>>>>,
) -> Result<(mpsc::Sender<Envelope>, tokio::task::JoinHandle<()>)> {
    let stream = ensure_daemon(sock).await?;
    let (read_half, write_half) = stream.into_split();
    let (tx, rx) = mpsc::channel::<Envelope>(256);
    spawn_writer(write_half, rx);
    let reader = spawn_reader(read_half, app.clone(), pending, out_channels);
    Ok((tx, reader))
}

impl Bridge {
    pub async fn connect(app: AppHandle) -> Result<Self> {
        let sock = socket_path()?;
        let pending: Arc<Mutex<HashMap<u64, oneshot::Sender<Envelope>>>> = Arc::default();
        let out_channels: Arc<Mutex<HashMap<String, Channel<Vec<u8>>>>> = Arc::default();
        let (tx, reader) = spawn_connection(&sock, &app, pending.clone(), out_channels.clone()).await?;
        let bridge = Self {
            next_id: AtomicU64::new(1),
            app,
            sock,
            gen: AtomicU64::new(0),
            tx: Mutex::new(tx),
            reconnect_lock: Mutex::new(()),
            reader: Mutex::new(reader),
            pending,
            out_channels,
        };
        bridge.ensure_matching_daemon().await;
        Ok(bridge)
    }

    /// If a previously-running daemon is a different build than the one bundled
    /// with this GUI (the daemon outlives the GUI, so an old one can still be
    /// serving the socket), restart it so our matching daemon takes over. No-op
    /// for unstamped dev builds or when the daemon is too old to report a build.
    async fn ensure_matching_daemon(&self) {
        let want = option_env!("SPACESH_BUILD").unwrap_or("dev");
        if want == "dev" {
            return;
        }
        let got = match self.request(Cmd::Health).await {
            Ok(env) => data_of(env)
                .ok()
                .and_then(|v| v.get("build").and_then(|b| b.as_str()).map(str::to_string))
                .unwrap_or_default(),
            Err(_) => return,
        };
        if got.is_empty() || got == want {
            return;
        }
        // Stale daemon — stop it and respawn our bundled one (matching code).
        let seen = self.gen.load(Ordering::Acquire);
        let _ = self.request(Cmd::Shutdown).await; // daemon exits; reply may not arrive
        let _ = self.reconnect(seen).await; // ensure_daemon respawns the bundled daemon
    }

    /// Re-establish the daemon connection. Single-flight: callers pass the `gen`
    /// they observed; if another caller already reconnected (gen advanced), this
    /// is a no-op so we never open duplicate connections.
    async fn reconnect(&self, seen_gen: u64) -> Result<()> {
        let _guard = self.reconnect_lock.lock().await;
        if self.gen.load(Ordering::Acquire) != seen_gen {
            return Ok(());
        }
        // Drop in-flight reply slots — their connection is gone; they'll error out.
        self.pending.lock().await.clear();
        // Kill the old reader FIRST so it can't keep delivering output on a stale
        // connection alongside the new one (the cause of doubled keystroke echo).
        self.reader.lock().await.abort();
        let (new_tx, new_reader) = spawn_connection(&self.sock, &self.app, self.pending.clone(), self.out_channels.clone()).await?;
        *self.tx.lock().await = new_tx;
        *self.reader.lock().await = new_reader;
        self.gen.fetch_add(1, Ordering::Release);
        let _ = self.app.emit("spacesh:reconnected", ());
        Ok(())
    }

    /// Send one request and await its reply with a timeout. Errors if the writer
    /// is gone, the reply slot is dropped, or no reply arrives in time.
    async fn send_once(&self, id: u64, env: Envelope) -> Result<Envelope> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.pending.lock().await.insert(id, reply_tx);
        let tx = self.tx.lock().await.clone();
        if tx.send(env).await.is_err() {
            self.pending.lock().await.remove(&id);
            anyhow::bail!("daemon writer closed");
        }
        match tokio::time::timeout(Duration::from_secs(5), reply_rx).await {
            Ok(Ok(env)) => Ok(env),
            Ok(Err(_)) => anyhow::bail!("reply slot dropped"),
            Err(_) => {
                self.pending.lock().await.remove(&id);
                anyhow::bail!("request timed out")
            }
        }
    }

    pub async fn request(&self, cmd: Cmd) -> Result<Envelope> {
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let seen_gen = self.gen.load(Ordering::Acquire);
        let env = Envelope::Req { id, cmd };
        match self.send_once(id, env.clone()).await {
            Ok(reply) => Ok(reply),
            Err(_) => {
                // Connection likely dropped — reconnect (respawns the daemon if
                // it exited) and retry once.
                self.reconnect(seen_gen).await?;
                self.send_once(id, env).await
            }
        }
    }

    pub async fn register_output(&self, surface_id: String, channel: Channel<Vec<u8>>) {
        self.out_channels.lock().await.insert(surface_id, channel);
    }

    pub async fn unregister_output(&self, surface_id: &str) {
        self.out_channels.lock().await.remove(surface_id);
    }
}

fn spawn_writer(mut write_half: OwnedWriteHalf, mut rx: mpsc::Receiver<Envelope>) {
    tokio::spawn(async move {
        while let Some(env) = rx.recv().await {
            if write_frame(&mut write_half, &env).await.is_err() {
                break;
            }
        }
    });
}

fn spawn_reader(
    mut read_half: OwnedReadHalf,
    app: AppHandle,
    pending: Arc<Mutex<HashMap<u64, oneshot::Sender<Envelope>>>>,
    out_channels: Arc<Mutex<HashMap<String, Channel<Vec<u8>>>>>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        loop {
            match read_frame(&mut read_half).await {
                Ok(Some(env)) => match env {
                    Envelope::Res { id, .. } => {
                        if let Some(slot) = pending.lock().await.remove(&id) {
                            let _ = slot.send(env);
                        }
                    }
                    Envelope::Evt(Evt::Output { surface_id, bytes }) => {
                        if let Some(ch) = out_channels.lock().await.get(&surface_id.0) {
                            let _ = ch.send(bytes);
                        }
                    }
                    Envelope::Evt(other) => {
                        // exit / surface_created / surface_closed → emit to webview.
                        let _ = app.emit("spacesh:evt", &other);
                    }
                    Envelope::Req { .. } => {}
                },
                Ok(None) | Err(_) => {
                    let _ = app.emit("spacesh:disconnected", ());
                    break;
                }
            }
        }
    })
}

// ---- Tauri commands ----

type BridgeState<'a> = tauri::State<'a, Bridge>;

fn data_of(env: Envelope) -> Result<Value, String> {
    match env {
        Envelope::Res { ok: true, data, .. } => Ok(data),
        Envelope::Res { ok: false, error, .. } => {
            Err(error.map(|e| format!("{}: {}", e.code, e.msg)).unwrap_or_else(|| "error".into()))
        }
        _ => Err("unexpected reply".into()),
    }
}

#[tauri::command]
pub async fn open(state: BridgeState<'_>, path: String) -> Result<Value, String> {
    data_of(state.request(Cmd::Open { path }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn new_surface(
    state: BridgeState<'_>,
    workspace_id: String,
    command: Option<String>,
    args: Vec<String>,
    cols: u16,
    rows: u16,
) -> Result<Value, String> {
    let cmd = Cmd::NewSurface {
        workspace_id: spacesh_proto::WorkspaceId(workspace_id),
        command,
        args,
        cols,
        rows,
    };
    data_of(state.request(cmd).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn input(state: BridgeState<'_>, surface_id: String, data: Vec<u8>) -> Result<Value, String> {
    let b64 = base64::engine::general_purpose::STANDARD.encode(&data);
    data_of(state.request(Cmd::Input { surface_id: SurfaceId(surface_id), bytes: b64 }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn resize(state: BridgeState<'_>, surface_id: String, cols: u16, rows: u16) -> Result<Value, String> {
    data_of(state.request(Cmd::Resize { surface_id: SurfaceId(surface_id), cols, rows }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn attach(state: BridgeState<'_>, surface_id: String, on_output: Channel<Vec<u8>>) -> Result<Value, String> {
    state.register_output(surface_id.clone(), on_output).await;
    data_of(state.request(Cmd::Attach { surface_id: SurfaceId(surface_id) }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn detach(state: BridgeState<'_>, surface_id: String) -> Result<Value, String> {
    state.unregister_output(&surface_id).await;
    data_of(state.request(Cmd::Detach { surface_id: SurfaceId(surface_id) }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn status(state: BridgeState<'_>) -> Result<Value, String> {
    data_of(state.request(Cmd::Status).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn close_surface(state: BridgeState<'_>, surface_id: String) -> Result<Value, String> {
    data_of(state.request(Cmd::Close { surface_id: SurfaceId(surface_id) }).await.map_err(|e| e.to_string())?)
}

// ---- M2 commands ----

#[tauri::command]
pub async fn split_surface(state: BridgeState<'_>, surface_id: String, dir: String, command: Option<String>, args: Vec<String>) -> Result<Value, String> {
    let dir = if dir == "down" { SplitDir::Down } else { SplitDir::Right };
    data_of(state.request(Cmd::SplitSurface { surface_id: SurfaceId(surface_id), dir, command, args }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn set_ratios(state: BridgeState<'_>, workspace_id: String, node_path: Vec<u32>, ratios: Vec<f32>) -> Result<Value, String> {
    data_of(state.request(Cmd::SetRatios { workspace_id: WorkspaceId(workspace_id), node_path, ratios }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn move_surface(state: BridgeState<'_>, surface_id: String, target_surface_id: String, edge: String) -> Result<Value, String> {
    let edge = match edge.as_str() { "left" => Edge::Left, "top" => Edge::Top, "bottom" => Edge::Bottom, _ => Edge::Right };
    data_of(state.request(Cmd::MoveSurface { surface_id: SurfaceId(surface_id), target_surface_id: SurfaceId(target_surface_id), edge }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn apply_preset(state: BridgeState<'_>, workspace_id: String, preset_id: String, slots: Vec<PresetSlot>) -> Result<Value, String> {
    data_of(state.request(Cmd::ApplyPreset { workspace_id: WorkspaceId(workspace_id), preset_id, slots }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn restart_surface(state: BridgeState<'_>, surface_id: String) -> Result<Value, String> {
    data_of(state.request(Cmd::RestartSurface { surface_id: SurfaceId(surface_id) }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn close_workspace(state: BridgeState<'_>, workspace_id: String) -> Result<Value, String> {
    data_of(state.request(Cmd::CloseWorkspace { workspace_id: WorkspaceId(workspace_id) }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn set_workspace_meta(state: BridgeState<'_>, workspace_id: String, name: Option<String>, group_id: Option<String>, unread: Option<bool>, order: Option<u32>, pinned: Option<bool>) -> Result<Value, String> {
    // group_id: None from JS means "no change"; an explicit null is sent as Some("") to mean "ungroup".
    let gid = match group_id {
        None => None,
        Some(s) if s.is_empty() => Some(None),
        Some(s) => Some(Some(GroupId(s))),
    };
    data_of(state.request(Cmd::SetWorkspaceMeta { workspace_id: WorkspaceId(workspace_id), name, group_id: gid, unread, order, pinned }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn create_group(state: BridgeState<'_>, name: String, color: String) -> Result<Value, String> {
    data_of(state.request(Cmd::CreateGroup { name, color }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn set_group(state: BridgeState<'_>, group_id: String, name: Option<String>, color: Option<String>, order: Option<u32>) -> Result<Value, String> {
    data_of(state.request(Cmd::SetGroup { group_id: GroupId(group_id), name, color, order }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn delete_group(state: BridgeState<'_>, group_id: String) -> Result<Value, String> {
    data_of(state.request(Cmd::DeleteGroup { group_id: GroupId(group_id) }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn focus(state: BridgeState<'_>, surface_id: String) -> Result<Value, String> {
    data_of(state.request(Cmd::Focus { surface_id: SurfaceId(surface_id) }).await.map_err(|e| e.to_string())?)
}

// ---- SP4 zoom command ----

#[tauri::command]
pub async fn set_zoom(state: BridgeState<'_>, workspace_id: String, surface_id: Option<String>) -> Result<Value, String> {
    let cmd = Cmd::SetZoom {
        workspace_id: spacesh_proto::WorkspaceId(workspace_id),
        surface_id: surface_id.map(spacesh_proto::SurfaceId),
    };
    data_of(state.request(cmd).await.map_err(|e| e.to_string())?)
}

// ---- M3 event log commands ----

#[tauri::command]
pub async fn event_log(state: BridgeState<'_>, limit: Option<u32>) -> Result<Value, String> {
    data_of(state.request(Cmd::EventLog { limit }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn mark_read(state: BridgeState<'_>, target: Value) -> Result<Value, String> {
    let target: spacesh_proto::MarkReadTarget = serde_json::from_value(target).map_err(|e| format!("invalid mark_read target: {e}"))?;
    data_of(state.request(Cmd::MarkRead { target }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn clear_events(state: BridgeState<'_>) -> Result<Value, String> {
    data_of(state.request(Cmd::ClearEvents).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn health(state: BridgeState<'_>) -> Result<Value, String> {
    data_of(state.request(Cmd::Health).await.map_err(|e| e.to_string())?)
}

// ---- Settings commands ----

#[tauri::command]
pub async fn get_config(state: BridgeState<'_>) -> Result<Value, String> {
    data_of(state.request(Cmd::GetConfig).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn set_config(
    state: BridgeState<'_>,
    default_shell: Option<String>,
    font_family: Option<String>,
    font_size: Option<u16>,
    theme: Option<String>,
    accent: Option<String>,
) -> Result<Value, String> {
    data_of(state.request(Cmd::SetConfig { default_shell, font_family, font_size, theme, accent }).await.map_err(|e| e.to_string())?)
}

#[tauri::command]
pub async fn shutdown_daemon(state: BridgeState<'_>) -> Result<Value, String> {
    data_of(state.request(Cmd::Shutdown).await.map_err(|e| e.to_string())?)
}