aev-bridge 0.1.1

use anyhow::{anyhow, Context, Result};
use axum::body::Body;
use axum::extract::{Path, Query, State};
use axum::http::{header, HeaderMap, HeaderValue, Request, StatusCode, Uri};
use axum::middleware::{self, Next};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{Html, IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use clap::Parser;
use futures_core::Stream;
use reqwest::Method;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::convert::Infallible;
use std::ffi::c_char;
use std::net::{SocketAddr, UdpSocket};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::TcpListener;
use uuid::Uuid;

pub const PLUGIN_ID: &str = "aev-bridge";
pub const PLUGIN_NAME: &str = "aev-bridge";
pub const CRATE_NAME: &str = "aev-bridge";
pub const ENTRYPOINT: &str = "aev_plugin_init";
pub const PROTOCOL: &str = "aev.plugin.v1";
pub const DEFAULT_BIND: &str = "0.0.0.0:8780";
pub const DEFAULT_UPSTREAM: &str = "http://127.0.0.1:8765";
pub const PLUGIN_INIT_JSON: &str = r#"{"id":"aev-bridge","name":"aev-bridge","package":"aev-bridge","entrypoint":"aev_plugin_init","protocol":"aev.plugin.v1","capabilities":["tool","bridge","mobile","http","sse","bearer-auth"]}"#;
const PLUGIN_INIT_JSON_NUL: &str = "{\"id\":\"aev-bridge\",\"name\":\"aev-bridge\",\"package\":\"aev-bridge\",\"entrypoint\":\"aev_plugin_init\",\"protocol\":\"aev.plugin.v1\",\"capabilities\":[\"tool\",\"bridge\",\"mobile\",\"http\",\"sse\",\"bearer-auth\"]}\0";

#[no_mangle]
pub extern "C" fn aev_plugin_init() -> *const c_char {
    PLUGIN_INIT_JSON_NUL.as_ptr().cast()
}

#[derive(Parser, Debug, Clone)]
#[command(name = "aev-bridge", about = "Start the Aev phone bridge for aev-web.")]
pub struct Cli {
    #[arg(long, env = "AEV_BRIDGE_BIND", default_value = DEFAULT_BIND)]
    pub bind: SocketAddr,
    #[arg(long, env = "AEV_BRIDGE_TOKEN")]
    pub token: Option<String>,
    #[arg(long, env = "AEV_BRIDGE_UPSTREAM", default_value = DEFAULT_UPSTREAM)]
    pub upstream: String,
    #[arg(long, env = "AEV_BRIDGE_UPSTREAM_TOKEN")]
    pub upstream_token: Option<String>,
    #[arg(long, env = "AEV_BRIDGE_PUBLIC_URL")]
    pub public_url: Option<String>,
    #[arg(long, env = "AEV_BRIDGE_POLL_MS", default_value_t = 300)]
    pub poll_ms: u64,
}

#[derive(Debug, Clone)]
pub struct ServerConfig {
    pub bind: SocketAddr,
    pub token: String,
    pub upstream: String,
    pub upstream_token: String,
    pub public_url: Option<String>,
    pub poll_ms: u64,
}

impl From<Cli> for ServerConfig {
    fn from(cli: Cli) -> Self {
        let upstream_token = cli
            .upstream_token
            .or_else(|| std::env::var("AEV_WEB_TOKEN").ok())
            .unwrap_or_default();
        Self {
            bind: cli.bind,
            token: normalize_token(cli.token),
            upstream: normalize_upstream(&cli.upstream),
            upstream_token: upstream_token.trim().to_string(),
            public_url: cli
                .public_url
                .map(|url| url.trim().trim_end_matches('/').to_string()),
            poll_ms: cli.poll_ms.clamp(100, 5_000),
        }
    }
}

#[derive(Debug, Clone, Serialize)]
pub struct PluginManifest {
    pub id: &'static str,
    pub name: &'static str,
    pub package: &'static str,
    pub entrypoint: &'static str,
    pub protocol: &'static str,
    pub capabilities: Vec<&'static str>,
    pub default_bind: &'static str,
    pub default_upstream: &'static str,
}

pub fn plugin_manifest() -> PluginManifest {
    PluginManifest {
        id: PLUGIN_ID,
        name: PLUGIN_NAME,
        package: CRATE_NAME,
        entrypoint: ENTRYPOINT,
        protocol: PROTOCOL,
        capabilities: vec!["tool", "bridge", "mobile", "http", "sse", "bearer-auth"],
        default_bind: DEFAULT_BIND,
        default_upstream: DEFAULT_UPSTREAM,
    }
}

pub async fn serve(config: ServerConfig) -> Result<()> {
    if config.upstream_token.trim().is_empty() {
        return Err(anyhow!(
            "missing upstream aev-web token; set AEV_WEB_TOKEN or --upstream-token"
        ));
    }

    let bridge_url = phone_url(&config);
    let state = AppState::new(config)?;
    let app = router(state.clone());
    let listener = TcpListener::bind(state.bind)
        .await
        .with_context(|| format!("failed to bind aev-bridge to {}", state.bind))?;
    let local_addr = listener
        .local_addr()
        .context("failed to read aev-bridge listener address")?;

    eprintln!("aev-bridge listening on http://{local_addr}");
    eprintln!("phone URL: {bridge_url}/?token={}", state.token);
    eprintln!("upstream: {}", state.upstream);

    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal())
        .await
        .context("aev-bridge server failed")
}

pub fn router(state: AppState) -> Router {
    let auth_token = state.token.clone();
    let api = Router::new()
        .route("/manifest", get(manifest_handler))
        .route("/upstream/health", get(upstream_health_handler))
        .route(
            "/sessions",
            get(list_sessions_handler).post(create_session_handler),
        )
        .route(
            "/sessions/{id}",
            get(get_session_handler).delete(delete_session_handler),
        )
        .route("/sessions/{id}/events", get(output_events_handler))
        .route("/sessions/{id}/input", post(write_input_handler))
        .route_layer(middleware::from_fn_with_state(auth_token, require_token));

    Router::new()
        .route("/", get(index_handler))
        .route("/health", get(health_handler))
        .nest("/api", api)
        .with_state(state)
}

#[derive(Clone)]
pub struct AppState {
    bind: SocketAddr,
    token: Arc<str>,
    upstream: Arc<str>,
    upstream_token: Arc<str>,
    poll: Duration,
    client: reqwest::Client,
}

impl AppState {
    pub fn new(config: ServerConfig) -> Result<Self> {
        Ok(Self {
            bind: config.bind,
            token: Arc::<str>::from(config.token),
            upstream: Arc::<str>::from(normalize_upstream(&config.upstream)),
            upstream_token: Arc::<str>::from(config.upstream_token),
            poll: Duration::from_millis(config.poll_ms.clamp(100, 5_000)),
            client: reqwest::Client::builder()
                .build()
                .context("failed to build bridge HTTP client")?,
        })
    }

    fn upstream_url(&self, path: &str) -> String {
        format!("{}{}", self.upstream, path)
    }

    fn upstream_auth(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        request.bearer_auth(self.upstream_token.as_ref())
    }
}

#[derive(Debug, Deserialize)]
pub struct EventQuery {
    pub since: Option<u64>,
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct UpstreamOutput {
    pub id: Uuid,
    pub cursor: u64,
    pub next: u64,
    pub data: String,
    pub dropped: bool,
    pub alive: bool,
}

#[derive(Debug, Serialize)]
struct HealthResponse {
    id: &'static str,
    status: &'static str,
    protocol: &'static str,
    upstream: String,
}

#[derive(Debug, Serialize)]
struct ErrorResponse {
    error: String,
}

#[derive(Debug)]
struct ApiError {
    status: StatusCode,
    message: String,
}

impl ApiError {
    fn bad_gateway(message: impl Into<String>) -> Self {
        Self {
            status: StatusCode::BAD_GATEWAY,
            message: message.into(),
        }
    }

    fn upstream(error: reqwest::Error) -> Self {
        Self::bad_gateway(error.to_string())
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        (
            self.status,
            Json(ErrorResponse {
                error: self.message,
            }),
        )
            .into_response()
    }
}

async fn index_handler() -> Html<&'static str> {
    Html(INDEX_HTML)
}

async fn health_handler(State(state): State<AppState>) -> Json<HealthResponse> {
    Json(HealthResponse {
        id: PLUGIN_ID,
        status: "ok",
        protocol: PROTOCOL,
        upstream: state.upstream.to_string(),
    })
}

async fn manifest_handler() -> Json<PluginManifest> {
    Json(plugin_manifest())
}

async fn upstream_health_handler(State(state): State<AppState>) -> Result<Response, ApiError> {
    proxy_to_upstream(&state, Method::GET, "/health", None).await
}

async fn list_sessions_handler(State(state): State<AppState>) -> Result<Response, ApiError> {
    proxy_to_upstream(&state, Method::GET, "/api/sessions", None).await
}

async fn create_session_handler(
    State(state): State<AppState>,
    Json(body): Json<Value>,
) -> Result<Response, ApiError> {
    proxy_to_upstream(&state, Method::POST, "/api/sessions", Some(body)).await
}

async fn get_session_handler(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Response, ApiError> {
    proxy_to_upstream(&state, Method::GET, &format!("/api/sessions/{id}"), None).await
}

async fn write_input_handler(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
    Json(body): Json<Value>,
) -> Result<Response, ApiError> {
    proxy_to_upstream(
        &state,
        Method::POST,
        &format!("/api/sessions/{id}/input"),
        Some(body),
    )
    .await
}

async fn delete_session_handler(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
) -> Result<Response, ApiError> {
    proxy_to_upstream(&state, Method::DELETE, &format!("/api/sessions/{id}"), None).await
}

async fn output_events_handler(
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
    Query(query): Query<EventQuery>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, ApiError> {
    let mut cursor = query.since.unwrap_or_default();
    let stream = async_stream::stream! {
        loop {
            match read_upstream_output(&state, id, cursor).await {
                Ok(output) => {
                    cursor = output.next;
                    if output.dropped || !output.data.is_empty() || !output.alive {
                        match serde_json::to_string(&output) {
                            Ok(json) => yield Ok(Event::default().event("output").data(json)),
                            Err(error) => {
                                let event = Event::default()
                                    .event("error")
                                    .data(error.to_string());
                                yield Ok(event);
                                break;
                            }
                        }
                    }
                    if !output.alive {
                        break;
                    }
                }
                Err(error) => {
                    let event = Event::default().event("error").data(error.message);
                    yield Ok(event);
                    break;
                }
            }
            tokio::time::sleep(state.poll).await;
        }
    };

    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}

async fn proxy_to_upstream(
    state: &AppState,
    method: Method,
    path: &str,
    body: Option<Value>,
) -> Result<Response, ApiError> {
    let mut request = state.upstream_auth(state.client.request(method, state.upstream_url(path)));
    if let Some(body) = body {
        request = request.json(&body);
    }

    let response = request.send().await.map_err(ApiError::upstream)?;
    let status =
        StatusCode::from_u16(response.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
    let content_type = response
        .headers()
        .get(reqwest::header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| HeaderValue::from_str(value).ok());
    let text = response.text().await.map_err(ApiError::upstream)?;
    let mut response = (status, text).into_response();
    if let Some(content_type) = content_type {
        response
            .headers_mut()
            .insert(header::CONTENT_TYPE, content_type);
    }
    Ok(response)
}

async fn read_upstream_output(
    state: &AppState,
    id: Uuid,
    since: u64,
) -> Result<UpstreamOutput, ApiError> {
    let response = state
        .upstream_auth(
            state
                .client
                .get(state.upstream_url(&format!("/api/sessions/{id}/output?since={since}"))),
        )
        .send()
        .await
        .map_err(ApiError::upstream)?;
    if !response.status().is_success() {
        return Err(ApiError::bad_gateway(format!(
            "upstream output read failed: {}",
            response.status()
        )));
    }
    response
        .json::<UpstreamOutput>()
        .await
        .map_err(ApiError::upstream)
}

async fn require_token(State(token): State<Arc<str>>, req: Request<Body>, next: Next) -> Response {
    if request_token_matches(req.headers(), req.uri(), token.as_ref()) {
        return next.run(req).await;
    }
    (
        StatusCode::UNAUTHORIZED,
        Json(ErrorResponse {
            error: "missing or invalid bridge token".to_string(),
        }),
    )
        .into_response()
}

pub fn request_token_matches(headers: &HeaderMap, uri: &Uri, token: &str) -> bool {
    bearer_token_matches(headers, token) || query_token_matches(uri, token)
}

pub fn bearer_token_matches(headers: &HeaderMap, token: &str) -> bool {
    let Some(value) = headers
        .get(header::AUTHORIZATION)
        .and_then(|value| value.to_str().ok())
    else {
        return false;
    };
    let Some(candidate) = value.strip_prefix("Bearer ") else {
        return false;
    };
    constant_time_eq(candidate.as_bytes(), token.as_bytes())
}

fn query_token_matches(uri: &Uri, token: &str) -> bool {
    uri.query()
        .and_then(|query| {
            query.split('&').find_map(|part| {
                let (key, value) = part.split_once('=')?;
                (key == "token").then_some(value)
            })
        })
        .map(|candidate| constant_time_eq(candidate.as_bytes(), token.as_bytes()))
        .unwrap_or(false)
}

fn normalize_token(token: Option<String>) -> String {
    token
        .map(|token| token.trim().to_string())
        .filter(|token| !token.is_empty())
        .unwrap_or_else(|| Uuid::now_v7().to_string())
}

pub fn normalize_upstream(upstream: &str) -> String {
    upstream.trim().trim_end_matches('/').to_string()
}

fn phone_url(config: &ServerConfig) -> String {
    if let Some(public_url) = &config.public_url {
        return public_url.clone();
    }
    if config.bind.ip().is_unspecified() {
        if let Some(url) = lan_url(config.bind.port()) {
            return url;
        }
    }
    format!("http://{}", config.bind)
}

fn lan_url(port: u16) -> Option<String> {
    let socket = UdpSocket::bind("0.0.0.0:0").ok()?;
    socket.connect("8.8.8.8:80").ok()?;
    let local_addr = socket.local_addr().ok()?;
    Some(format!("http://{}:{port}", local_addr.ip()))
}

fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
    if left.len() != right.len() {
        return false;
    }
    left.iter()
        .zip(right)
        .fold(0_u8, |diff, (left, right)| diff | (left ^ right))
        == 0
}

async fn shutdown_signal() {
    let _ = tokio::signal::ctrl_c().await;
}

const INDEX_HTML: &str = r#"<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>aev-bridge</title>
  <style>
    :root { color-scheme: dark; font-family: Inter, ui-sans-serif, system-ui, sans-serif; background: #07090f; color: #f5f7fb; }
    body { margin: 0; min-height: 100vh; background: linear-gradient(135deg, #07111f, #140b24 55%, #05070b); }
    main { min-height: 100vh; display: grid; grid-template-rows: auto auto 1fr auto; }
    header { padding: 22px 18px 14px; display: flex; justify-content: space-between; gap: 12px; align-items: end; }
    h1 { margin: 0; font-size: clamp(28px, 10vw, 58px); letter-spacing: -0.06em; }
    .tag { color: #9da7bd; font-size: 13px; }
    .bar { display: grid; grid-template-columns: 1fr auto; gap: 9px; padding: 0 14px 12px; }
    input, textarea, button { border: 1px solid #2f3a50; border-radius: 14px; background: #0e1420; color: #f5f7fb; font: inherit; }
    input, textarea { padding: 13px 14px; min-width: 0; }
    button { padding: 13px 15px; background: #6d5dfc; border-color: #877cff; font-weight: 800; }
    button.secondary { background: #182033; border-color: #2f3a50; }
    pre { margin: 0 14px; border: 1px solid #253047; border-radius: 18px; background: #02040a; padding: 14px; overflow: auto; white-space: pre-wrap; word-break: break-word; font: 13px/1.45 ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; color: #cbffd8; }
    footer { padding: 12px 14px 16px; display: grid; grid-template-columns: 1fr auto auto; gap: 9px; }
    textarea { min-height: 46px; resize: vertical; }
    @media (max-width: 620px) { footer, .bar { grid-template-columns: 1fr; } header { display: block; } }
  </style>
</head>
<body>
  <main>
    <header>
      <div>
        <h1>aev-bridge</h1>
        <div class="tag">HTTP + SSE bridge to aev-web</div>
      </div>
      <div id="status" class="tag">offline</div>
    </header>
    <section class="bar">
      <input id="token" type="password" autocomplete="off" placeholder="Bridge token">
      <button id="start">Start Session</button>
    </section>
    <pre id="screen"></pre>
    <footer>
      <textarea id="input" placeholder="Command, then Enter"></textarea>
      <button id="send">Send</button>
      <button id="ctrlc" class="secondary">Ctrl-C</button>
    </footer>
  </main>
  <script>
    const statusEl = document.querySelector('#status');
    const screenEl = document.querySelector('#screen');
    const tokenEl = document.querySelector('#token');
    const inputEl = document.querySelector('#input');
    let sessionId = null;
    let events = null;

    tokenEl.value = new URLSearchParams(location.search).get('token') || localStorage.getItem('aevBridgeToken') || '';

    function token() {
      const value = tokenEl.value.trim();
      if (value) localStorage.setItem('aevBridgeToken', value);
      return value;
    }

    async function api(path, options = {}) {
      const headers = Object.assign({ 'Authorization': `Bearer ${token()}` }, options.headers || {});
      if (options.body) headers['Content-Type'] = 'application/json';
      const response = await fetch(`/api${path}`, Object.assign({}, options, { headers }));
      const body = await response.json().catch(() => ({}));
      if (!response.ok) throw new Error(body.error || response.statusText);
      return body;
    }

    function connectEvents() {
      if (events) events.close();
      events = new EventSource(`/api/sessions/${sessionId}/events?token=${encodeURIComponent(token())}`);
      events.addEventListener('output', event => {
        const output = JSON.parse(event.data);
        if (output.dropped) screenEl.textContent += '\n[bridge: output truncated]\n';
        if (output.data) {
          screenEl.textContent += output.data;
          screenEl.scrollTop = screenEl.scrollHeight;
        }
        if (!output.alive) statusEl.textContent = `session ${sessionId} exited`;
      });
      events.addEventListener('error', () => { statusEl.textContent = 'event stream disconnected'; });
    }

    async function start() {
      if (!token()) {
        statusEl.textContent = 'token required';
        return;
      }
      const result = await api('/sessions', { method: 'POST', body: JSON.stringify({}) });
      sessionId = result.session.id;
      screenEl.textContent = '';
      statusEl.textContent = `session ${sessionId}`;
      connectEvents();
    }

    async function send(data) {
      if (!sessionId) await start();
      if (!sessionId || !data) return;
      await api(`/sessions/${sessionId}/input`, { method: 'POST', body: JSON.stringify({ data }) });
    }

    document.querySelector('#start').addEventListener('click', () => start().catch(error => statusEl.textContent = error.message));
    document.querySelector('#send').addEventListener('click', () => {
      const data = inputEl.value;
      inputEl.value = '';
      send(data).catch(error => statusEl.textContent = error.message);
    });
    document.querySelector('#ctrlc').addEventListener('click', () => send('\u0003').catch(error => statusEl.textContent = error.message));
    inputEl.addEventListener('keydown', event => {
      if (event.key === 'Enter' && !event.shiftKey) {
        event.preventDefault();
        const data = `${inputEl.value}\r`;
        inputEl.value = '';
        send(data).catch(error => statusEl.textContent = error.message);
      }
    });
  </script>
</body>
</html>"#;

#[cfg(test)]
mod tests {
    use super::*;
    use axum::http::HeaderValue;
    use std::ffi::CStr;

    #[test]
    fn manifest_matches_plugin_contract() {
        let manifest = plugin_manifest();

        assert_eq!(manifest.id, "aev-bridge");
        assert_eq!(manifest.package, "aev-bridge");
        assert_eq!(manifest.entrypoint, "aev_plugin_init");
        assert_eq!(manifest.protocol, "aev.plugin.v1");
        assert!(manifest.capabilities.contains(&"sse"));
    }

    #[test]
    fn ffi_entrypoint_returns_static_manifest_json() {
        assert!(!aev_plugin_init().is_null());
        let json = unsafe { CStr::from_ptr(aev_plugin_init()) }
            .to_str()
            .expect("plugin manifest is utf-8");
        assert_eq!(json, PLUGIN_INIT_JSON);
    }

    #[test]
    fn auth_accepts_bearer_or_query_token() {
        let mut headers = HeaderMap::new();
        let uri = Uri::from_static("/api/sessions?token=secret");
        assert!(request_token_matches(&headers, &uri, "secret"));

        headers.insert(
            header::AUTHORIZATION,
            HeaderValue::from_static("Bearer secret"),
        );
        let uri = Uri::from_static("/api/sessions");
        assert!(request_token_matches(&headers, &uri, "secret"));
        assert!(!request_token_matches(&headers, &uri, "different"));
    }

    #[test]
    fn upstream_base_url_is_normalized() {
        assert_eq!(
            normalize_upstream(" http://127.0.0.1:8765/// "),
            "http://127.0.0.1:8765"
        );
    }

    #[test]
    fn router_constructs_with_capture_syntax() {
        let config = ServerConfig {
            bind: DEFAULT_BIND.parse().expect("bind parses"),
            token: "bridge".to_string(),
            upstream: DEFAULT_UPSTREAM.to_string(),
            upstream_token: "upstream".to_string(),
            public_url: None,
            poll_ms: 300,
        };
        let state = AppState::new(config).expect("state builds");
        let _ = router(state);
    }
}