Premiere Pro Bridge

//! Adom Bridge bridge for Adobe Premiere Pro.
//!
//! Two halves:
//!   * this Rust server — Adom Bridge spawns it, POSTs `{command, args}` verbs at it,
//!     and polls `/health` for the status chip;
//!   * a UXP plugin running INSIDE Premiere (see `plugin/`) — it dials back into this
//!     server over a loopback WebSocket (`/ws`) and executes verbs against the
//!     `premierepro` scripting API. No mouse, no keyboard, no window automation.
//!
//! A verb arrives here, gets forwarded to the plugin as `{type:"call", id, verb, args}`,
//! and the plugin's `{type:"result", id, ok, result|error}` is wrapped with the
//! `_hint` / `_next` / `related` / `pitfalls` the calling AI actually reads.

mod ab;
mod catalog;
mod host;
mod plugin_pkg;

use axum::{
    body::Bytes,
    extract::{
        ws::{Message, WebSocket, WebSocketUpgrade},
        Query, State,
    },
    http::{header, HeaderMap, Method, StatusCode},
    response::{IntoResponse, Response},
    routing::{any, get},
    Json, Router,
};
use futures_util::{SinkExt, StreamExt};
use serde_json::{json, Value};
use std::{
    collections::HashMap,
    path::PathBuf,
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc,
    },
    time::{Duration, Instant},
};
use tokio::sync::{mpsc, oneshot, Mutex};

pub const VERSION: &str = env!("CARGO_PKG_VERSION");
pub const PREFIX: &str = "premiere_";

/// Protocol version spoken between this server and the UXP plugin. Bump on any
/// breaking change to the call/result envelope; the plugin reports its own in `hello`.
pub const PROTOCOL: u64 = 1;

pub struct PluginConn {
    runtime: String,
    tx: mpsc::UnboundedSender<String>,
    hello: Value,
    connected_at: Instant,
    conn_id: u64,
}

pub struct AppState {
    pub port: u16,
    pub token: String,
    pub data_dir: PathBuf,
    pub started: Instant,
    /// Live in-app connections keyed by runtime: "uxp" (primary) and "cep" (helper).
    plugins: Mutex<HashMap<String, PluginConn>>,
    /// In-flight calls: id -> (connection that owns it, waiter).
    pending: Mutex<HashMap<u64, (u64, oneshot::Sender<Value>)>>,
    next_id: AtomicU64,
    /// Plugin-pushed events (selection changed, export finished...), newest last.
    events: Mutex<Vec<Value>>,
    event_seq: AtomicU64,
}

pub type Shared = Arc<AppState>;

impl AppState {
    pub async fn runtime_connected(&self, runtime: &str) -> bool {
        self.plugins.lock().await.contains_key(runtime)
    }

    /// The primary (UXP) plugin; "connected" in user-facing terms means this one.
    pub async fn plugin_connected(&self) -> bool {
        self.runtime_connected("uxp").await
    }

    pub async fn plugin_hello(&self) -> Option<(Value, u64)> {
        self.runtime_hello("uxp").await
    }

    pub async fn runtime_hello(&self, runtime: &str) -> Option<(Value, u64)> {
        self.plugins
            .lock()
            .await
            .get(runtime)
            .map(|p| (p.hello.clone(), p.connected_at.elapsed().as_secs()))
    }

    /// Send one verb to an in-app runtime ("uxp" or "cep") and wait for its result.
    pub async fn call_plugin(&self, verb: &str, args: Value, timeout: Duration) -> Result<Value, CallError> {
        self.call_runtime("uxp", verb, args, timeout).await
    }

    pub async fn call_runtime(&self, runtime: &str, verb: &str, args: Value, timeout: Duration) -> Result<Value, CallError> {
        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
        let (tx, rx) = oneshot::channel();
        {
            let guard = self.plugins.lock().await;
            let Some(conn) = guard.get(runtime) else {
                return Err(CallError::NotConnected);
            };
            self.pending.lock().await.insert(id, (conn.conn_id, tx));
            let msg = json!({"type": "call", "id": id, "verb": verb, "args": args}).to_string();
            if conn.tx.send(msg).is_err() {
                self.pending.lock().await.remove(&id);
                return Err(CallError::NotConnected);
            }
        }
        match tokio::time::timeout(timeout, rx).await {
            Ok(Ok(v)) => Ok(v),
            Ok(Err(_)) => Err(CallError::Disconnected),
            Err(_) => {
                self.pending.lock().await.remove(&id);
                Err(CallError::Timeout(timeout.as_secs()))
            }
        }
    }

    pub async fn recent_events(&self, since: u64) -> (Vec<Value>, u64) {
        let ev = self.events.lock().await;
        let out = ev
            .iter()
            .filter(|e| e["seq"].as_u64().unwrap_or(0) > since)
            .cloned()
            .collect();
        (out, self.event_seq.load(Ordering::Relaxed))
    }
}

#[derive(Debug)]
pub enum CallError {
    NotConnected,
    Disconnected,
    Timeout(u64),
}

// ───────────────────────────── entry ─────────────────────────────

#[tokio::main]
async fn main() {
    let mut port: u16 = 0;
    let mut args = std::env::args().skip(1);
    while let Some(a) = args.next() {
        if a == "--port" {
            port = args.next().and_then(|p| p.parse().ok()).unwrap_or(0);
        } else if let Some(p) = a.strip_prefix("--port=") {
            port = p.parse().unwrap_or(0);
        } else if a == "--version" {
            println!("premiere-bridge {VERSION}");
            return;
        }
    }
    let bind_host = std::env::var("ADOM_BIND_HOST").unwrap_or_else(|_| "127.0.0.1".into());
    let listener = tokio::net::TcpListener::bind((bind_host.as_str(), port))
        .await
        .unwrap_or_else(|e| panic!("bind {bind_host}:{port}: {e}"));
    let port = listener.local_addr().map(|a| a.port()).unwrap_or(port);

    let data_dir = host::data_dir();
    let _ = std::fs::create_dir_all(&data_dir);
    let token = host::load_or_create_token(&data_dir);

    let state: Shared = Arc::new(AppState {
        port,
        token,
        data_dir,
        started: Instant::now(),
        plugins: Mutex::new(HashMap::new()),
        pending: Mutex::new(HashMap::new()),
        next_id: AtomicU64::new(1),
        events: Mutex::new(Vec::new()),
        event_seq: AtomicU64::new(0),
    });

    // The plugin finds us by reading this file (port changes are picked up on its next
    // reconnect attempt), so write it before accepting anything.
    host::write_connection_file(&state);

    let app = Router::new()
        .route("/health", get(health))
        .route("/ws", get(ws_upgrade))
        .route("/", any(command))
        .route("/command", any(command))
        .with_state(state.clone());

    eprintln!("[premiere-bridge {VERSION}] listening on {bind_host}:{port} data={}", state.data_dir.display());
    axum::serve(listener, app).await.expect("server");
}

// ───────────────────────────── /health ─────────────────────────────

async fn health(State(st): State<Shared>) -> Json<Value> {
    let running = host::premiere_running().await;
    let hello = st.plugin_hello().await;
    let cep = st.runtime_connected("cep").await;
    let (led, summary, tooltip) = match (&hello, running) {
        (Some((h, _)), _) => (
            "green",
            "Connected".to_string(),
            format!(
                "Premiere Pro {} connected via UXP plugin {}. Project: {}",
                h["premiereVersion"].as_str().unwrap_or("?"),
                h["pluginVersion"].as_str().unwrap_or("?"),
                h["project"].as_str().unwrap_or("(none open)")
            ),
        ),
        (None, true) if cep => (
            "yellow",
            "Helper only".to_string(),
            "Premiere is running and the Adom CEP helper is connected, but the main UXP plugin isn't. Open Window > UXP Plugins > Adom Bridge once (it then auto-restores).".to_string(),
        ),
        (None, true) => (
            "yellow",
            "Plugin not connected".to_string(),
            "Premiere is running but the Adom UXP plugin hasn't dialed in. Run premiere_install_plugin, then open Window > UXP Plugins > Adom Bridge once.".to_string(),
        ),
        (None, false) => (
            "yellow",
            "Premiere not running".to_string(),
            "Bridge is up. Premiere Pro isn't running — premiere_launch starts it.".to_string(),
        ),
    };
    Json(json!({
        "ok": true,
        "name": "premiere",
        "version": VERSION,
        "led": led,
        "summary": summary,
        "tooltip": tooltip,
        "pluginConnected": hello.is_some(),
        "cepHelperConnected": cep,
        "premiereRunning": running,
        "uptimeSec": st.started.elapsed().as_secs(),
    }))
}

// ───────────────────────────── /command ─────────────────────────────

async fn command(State(st): State<Shared>, method: Method, headers: HeaderMap, body: Bytes) -> Response {
    // Loopback is not an auth boundary: a web page can POST here. Refuse anything that
    // isn't a JSON POST *before* touching the body, and never send ACAO.
    if method != Method::POST {
        return (StatusCode::METHOD_NOT_ALLOWED, [(header::ALLOW, "POST")], "POST only").into_response();
    }
    let is_json = headers
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .map(|v| v.split(';').next().unwrap_or("").trim().eq_ignore_ascii_case("application/json"))
        .unwrap_or(false);
    if !is_json {
        return (
            StatusCode::UNSUPPORTED_MEDIA_TYPE,
            "Expected request with `Content-Type: application/json`",
        )
            .into_response();
    }
    let req: Value = match serde_json::from_slice(&body) {
        Ok(v) => v,
        Err(e) => {
            return Json(json!({"success": false, "error": format!("bad JSON: {e}"), "errorCode": "bad_request"}))
                .into_response()
        }
    };
    let verb = req["command"].as_str().unwrap_or("").to_string();
    let args = match &req["args"] {
        Value::Null => json!({}),
        Value::String(s) => serde_json::from_str(s).unwrap_or_else(|_| json!({})),
        v => v.clone(),
    };
    Json(dispatch(&st, &verb, args).await).into_response()
}

/// Verb names are accepted with or without the `premiere_` prefix. Boxed because local
/// composite verbs (preview_frame, launch) dispatch other verbs.
pub fn dispatch<'a>(st: &'a Shared, verb: &'a str, args: Value) -> std::pin::Pin<Box<dyn std::future::Future<Output = Value> + Send + 'a>> {
    Box::pin(async move {
        let bare = verb.strip_prefix(PREFIX).unwrap_or(verb);
        let mut out = match host::local_verb(st, bare, &args).await {
            Some(v) => v,
            None => forward(st, bare, args).await,
        };
        catalog::decorate(bare, &mut out);
        out
    })
}

async fn forward(st: &Shared, bare: &str, args: Value) -> Value {
    let Some(spec) = catalog::find(bare) else {
        return json!({
            "success": false,
            "errorCode": "unknown_verb",
            "error": format!("unknown verb premiere_{bare}"),
            "_hint": "Call premiere_describe for the full verb catalog. Anything the catalog doesn't cover can be done with premiere_eval (raw JS against the premierepro UXP API).",
        });
    };
    let timeout = Duration::from_secs(
        args["timeoutSec"].as_u64().unwrap_or(spec.timeout_sec).clamp(1, 3600),
    );
    let args = host::fill_defaults(st, bare, args);
    let runtime = match spec.runtime {
        catalog::Runtime::Uxp | catalog::Runtime::Local => "uxp",
        catalog::Runtime::Cep => "cep",
        // True QE razor beats the UXP emulation when the helper is there.
        catalog::Runtime::PreferCep => {
            if st.runtime_connected("cep").await { "cep" } else { "uxp" }
        }
    };
    match st.call_runtime(runtime, bare, args, timeout).await {
        Ok(res) => {
            if res["ok"].as_bool().unwrap_or(false) {
                let mut out = json!({"success": true});
                merge_result(&mut out, res.get("result").cloned().unwrap_or(Value::Null));
                if let Some(w) = res.get("warnings") {
                    out["warnings"] = w.clone();
                }
                out
            } else {
                json!({
                    "success": false,
                    "error": res["error"].as_str().unwrap_or("plugin error"),
                    "errorCode": res["errorCode"].as_str().unwrap_or("premiere_error"),
                    "stack": res.get("stack").cloned().unwrap_or(Value::Null),
                })
            }
        }
        Err(CallError::NotConnected) if runtime == "cep" => host::cep_not_connected_response(st).await,
        Err(CallError::NotConnected) => host::not_connected_response(st).await,
        Err(CallError::Disconnected) => json!({
            "success": false,
            "errorCode": "plugin_disconnected",
            "error": "Premiere's UXP plugin disconnected mid-call (Premiere quit, crashed, or the plugin reloaded).",
            "_hint": "Check premiere_status. If Premiere is still up, the plugin reconnects within ~3s — retry once. If the verb mutated the timeline, call premiere_get_sequence before retrying so you don't apply it twice.",
        }),
        Err(CallError::Timeout(s)) => json!({
            "success": false,
            "errorCode": "plugin_timeout",
            "stillRunning": true,
            "statusVerb": "premiere_status",
            "error": format!("Premiere didn't answer within {s}s. A modal dialog in Premiere blocks scripting, or the operation is genuinely long."),
            "_hint": "Call premiere_status (shows whether Premiere is responsive). For long exports use premiere_export_sequence (it returns immediately and you poll premiere_export_status). You can pass a bigger timeoutSec.",
        }),
    }
}

/// Plugin results that are objects get flattened into the response; anything else
/// lands under `result`.
fn merge_result(out: &mut Value, result: Value) {
    match result {
        Value::Object(map) => {
            for (k, v) in map {
                if k != "success" {
                    out[k] = v;
                }
            }
        }
        Value::Null => {}
        other => out["result"] = other,
    }
}

// ───────────────────────────── /ws (plugin) ─────────────────────────────

async fn ws_upgrade(
    State(st): State<Shared>,
    Query(q): Query<HashMap<String, String>>,
    ws: WebSocketUpgrade,
) -> Response {
    // WebSockets are NOT covered by CORS, so any web page could open one. The token
    // (only readable from the user's ~/.adom dir) is what keeps strangers out.
    if q.get("token").map(String::as_str) != Some(st.token.as_str()) {
        return (StatusCode::UNAUTHORIZED, "bad token").into_response();
    }
    ws.max_message_size(256 * 1024 * 1024)
        .on_upgrade(move |sock| plugin_session(st, sock))
}

static CONN_SEQ: AtomicU64 = AtomicU64::new(1);

async fn plugin_session(st: Shared, sock: WebSocket) {
    let (mut sink, mut stream) = sock.split();
    let (tx, mut rx) = mpsc::unbounded_channel::<String>();
    let conn_id = CONN_SEQ.fetch_add(1, Ordering::Relaxed);

    let writer = tokio::spawn(async move {
        while let Some(msg) = rx.recv().await {
            if sink.send(Message::Text(msg)).await.is_err() {
                break;
            }
        }
    });

    // Keepalive: UXP sockets die silently when Premiere sleeps; a ping every 15s
    // surfaces that as a disconnect instead of a hung call.
    let ka_tx = tx.clone();
    let keepalive = tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_secs(15)).await;
            if ka_tx.send(json!({"type": "ping"}).to_string()).is_err() {
                break;
            }
        }
    });

    while let Some(Ok(msg)) = stream.next().await {
        let text = match msg {
            Message::Text(t) => t,
            Message::Binary(b) => String::from_utf8_lossy(&b).into_owned(),
            Message::Close(_) => break,
            _ => continue,
        };
        let Ok(v) = serde_json::from_str::<Value>(&text) else { continue };
        match v["type"].as_str() {
            Some("hello") => {
                let runtime = if v["runtime"].as_str() == Some("cep") { "cep" } else { "uxp" }.to_string();
                eprintln!("[premiere-bridge] {runtime} connected: {v}");
                // A newer connection replaces an older one of the same runtime (plugin
                // reloaded, a second panel instance).
                st.plugins.lock().await.insert(
                    runtime.clone(),
                    PluginConn { runtime, tx: tx.clone(), hello: v.clone(), connected_at: Instant::now(), conn_id },
                );
                let _ = tx.send(json!({"type": "welcome", "bridgeVersion": VERSION, "protocol": PROTOCOL}).to_string());
            }
            Some("result") => {
                if let Some(id) = v["id"].as_u64() {
                    if let Some((_, waiter)) = st.pending.lock().await.remove(&id) {
                        let _ = waiter.send(v);
                    }
                }
            }
            Some("event") => {
                let seq = st.event_seq.fetch_add(1, Ordering::Relaxed) + 1;
                let mut ev = v.clone();
                ev["seq"] = json!(seq);
                let mut events = st.events.lock().await;
                events.push(ev);
                let len = events.len();
                if len > 200 {
                    events.drain(..len - 200);
                }
            }
            Some("hello_update") => {
                if let Some(p) = st.plugins.lock().await.values_mut().find(|p| p.conn_id == conn_id) {
                    {
                        if let (Some(dst), Some(src)) = (p.hello.as_object_mut(), v["fields"].as_object()) {
                            for (k, val) in src {
                                dst.insert(k.clone(), val.clone());
                            }
                        }
                    }
                }
            }
            _ => {}
        }
    }

    keepalive.abort();
    writer.abort();
    let mut guard = st.plugins.lock().await;
    let gone: Vec<String> = guard.values().filter(|p| p.conn_id == conn_id).map(|p| p.runtime.clone()).collect();
    for rt in &gone {
        guard.remove(rt);
        eprintln!("[premiere-bridge] {rt} disconnected");
    }
    drop(guard);
    // Dropping this connection's waiters wakes its in-flight calls with Disconnected
    // instead of letting them sit until their timeout.
    if !gone.is_empty() {
        st.pending.lock().await.retain(|_, (owner, _)| *owner != conn_id);
    }
}