PiCloud/crates/picloud/tests/invoke_e2e.rs

//! v1.1.9 invoke end-to-end tests. invoke()/invoke_async() exercised
//! against the full all-in-one app via build_app, so the invoke service
//! + Engine back-reference + InvokeServiceImpl all participate.
//!
//! The caller scripts run via the admin bypass `POST /api/v1/execute/{id}`
//! (same pattern dispatcher_e2e uses) — sidesteps the per-app domain /
//! route matcher entirely.
//!
//! Skips when DATABASE_URL is unset.

#![allow(clippy::needless_pass_by_value)]

use std::time::Duration;

use axum_test::TestServer;
use serde_json::{json, Value};
use sqlx::postgres::PgPoolOptions;
use sqlx::PgPool;
use uuid::Uuid;

async fn pool_or_skip() -> Option<PgPool> {
    let Ok(url) = std::env::var("DATABASE_URL") else {
        eprintln!("invoke_e2e: DATABASE_URL unset — skipping");
        return None;
    };
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect(&url)
        .await
        .expect("connect to DATABASE_URL");
    sqlx::migrate!("../manager-core/migrations")
        .run(&pool)
        .await
        .expect("apply migrations");
    Some(pool)
}

async fn server_for(pool: PgPool, suffix: &str) -> (TestServer, String) {
    use picloud_manager_core::auth::hash_password;
    use picloud_shared::InstanceRole;

    let unique = format!("{suffix}-{}", Uuid::new_v4().simple());
    let auth = picloud::AuthDeps::from_pool(pool.clone());
    let username = format!("e2e-{unique}");
    let hash = hash_password("pw").expect("hash");
    auth.users
        .create(&username, &hash, InstanceRole::Owner, None)
        .await
        .expect("seed admin");

    let app = picloud::build_app(
        pool,
        auth,
        picloud_shared::MasterKey::from_bytes([0x42u8; 32]),
    )
    .await
    .expect("build_app");
    let mut server = TestServer::new(app).expect("TestServer");
    let resp = server
        .post("/api/v1/admin/auth/login")
        .json(&json!({ "username": username, "password": "pw" }))
        .await;
    resp.assert_status_ok();
    let token = resp.json::<Value>()["token"]
        .as_str()
        .expect("login token")
        .to_string();
    server.add_header("authorization", format!("Bearer {token}"));

    let slug = format!("e2e-{unique}");
    let created: Value = server
        .post("/api/v1/admin/apps")
        .json(&json!({ "slug": slug, "name": slug }))
        .await
        .json();
    let app_id = created["id"].as_str().expect("app id").to_string();
    (server, app_id)
}

async fn create_script(server: &TestServer, app_id: &str, name: &str, source: &str) -> String {
    let created: Value = server
        .post("/api/v1/admin/scripts")
        .json(&json!({ "app_id": app_id, "name": name, "source": source }))
        .await
        .json();
    created["id"].as_str().expect("script id").to_string()
}

/// Direct executor bypass — same as dispatcher_e2e::execute. The script
/// runs sync with an empty body; sidesteps the route/host matcher.
async fn execute(server: &TestServer, script_id: &str) -> Value {
    server
        .post(&format!("/api/v1/execute/{script_id}"))
        .json(&json!({}))
        .await
        .json()
}

async fn poll_marker(pool: &PgPool, app_id: &str) -> Option<Value> {
    for _ in 0..150 {
        let row: Option<(Value,)> = sqlx::query_as(
            "SELECT value FROM kv_entries WHERE app_id = $1 \
             AND collection = 'e2e_markers' AND key = 'marker'",
        )
        .bind(Uuid::parse_str(app_id).expect("uuid"))
        .fetch_optional(pool)
        .await
        .expect("kv read");
        if let Some((v,)) = row {
            return Some(v);
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    None
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn invoke_by_name_same_app_returns_value() {
    let Some(pool) = pool_or_skip().await else { return };
    let (server, app_id) = server_for(pool.clone(), "invoke-name").await;

    // Callee returns 42 — the simplest possible round-trip.
    create_script(
        &server,
        &app_id,
        "callee",
        r#"#{ statusCode: 200, body: 42 }"#,
    )
    .await;

    // Caller invokes by name and surfaces the result via the response body.
    let caller = create_script(
        &server,
        &app_id,
        "caller",
        r#"
            let r = invoke("callee", #{});
            #{ statusCode: 200, body: r }
        "#,
    )
    .await;

    let body = execute(&server, &caller).await;
    assert_eq!(body, json!(42), "callee return value should round-trip");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn invoke_cross_app_rejects() {
    let Some(pool) = pool_or_skip().await else { return };
    let (server, app_a) = server_for(pool.clone(), "invoke-xa").await;
    // Create a second app via the SAME server / admin (Owner can manage
    // every app) so the script-create calls don't run into role gates.
    let slug_b = format!("e2e-invoke-xb-{}", Uuid::new_v4().simple());
    let created_b: Value = server
        .post("/api/v1/admin/apps")
        .json(&json!({ "slug": slug_b, "name": slug_b }))
        .await
        .json();
    let app_b = created_b["id"].as_str().expect("app_b id").to_string();

    // Callee lives in app_b — caller in app_a must not be able to reach it.
    let callee_b = create_script(
        &server,
        &app_b,
        "callee_b",
        r#"#{ statusCode: 200, body: 0 }"#,
    )
    .await;

    // Caller in app_a tries to invoke callee_b by id; the error must
    // surface to the response body.
    let caller_src = format!(
        r#"
            let err = "no error";
            try {{
                invoke("{callee_b}", #{{}});
            }} catch(e) {{
                err = e;
            }}
            #{{ statusCode: 200, body: err }}
        "#
    );
    let caller_a = create_script(&server, &app_a, "caller_a", &caller_src).await;
    let body = execute(&server, &caller_a).await;
    let err_str = body.as_str().unwrap_or_default();
    assert!(
        err_str.contains("different app")
            || err_str.contains("CrossApp")
            || err_str.contains("not found"),
        "expected cross-app or not-found rejection, got: {err_str}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn invoke_depth_limit_exceeds_cleanly() {
    let Some(pool) = pool_or_skip().await else { return };
    let (server, app_id) = server_for(pool.clone(), "invoke-depth").await;

    // The recurser propagates any throw (no try-catch). At the depth
    // ceiling Limits::trigger_depth_max=8 the innermost invoke() throws
    // "invoke: depth limit exceeded (max 8)"; each nested invoke() then
    // re-throws up the stack until the OUTER caller's try-catch surfaces
    // it.
    create_script(
        &server,
        &app_id,
        "recurser",
        r#"invoke("recurser", #{}); #{ statusCode: 200 }"#,
    )
    .await;

    let caller = create_script(
        &server,
        &app_id,
        "depth_caller",
        r#"
            let err = "ok";
            try {
                invoke("recurser", #{});
            } catch(e) {
                err = e;
            }
            #{ statusCode: 200, body: err }
        "#,
    )
    .await;

    let body = execute(&server, &caller).await;
    let err = body.as_str().unwrap_or_default();
    assert!(
        err.contains("depth"),
        "expected depth-limit error, got: {err}"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn invoke_async_enqueues_outbox_row() {
    let Some(pool) = pool_or_skip().await else { return };
    let (server, app_id) = server_for(pool.clone(), "invoke-async").await;

    // Callee writes a marker the test polls — the dispatcher must fire
    // the OutboxSourceKind::Invoke row.
    create_script(
        &server,
        &app_id,
        "async_callee",
        r#"
            kv::collection("e2e_markers").set("marker", #{ from: "async" });
            #{ statusCode: 200 }
        "#,
    )
    .await;

    let caller = create_script(
        &server,
        &app_id,
        "async_caller",
        r#"
            let id = invoke_async("async_callee", #{});
            #{ statusCode: 200, body: id }
        "#,
    )
    .await;
    execute(&server, &caller).await;

    let marker = poll_marker(&pool, &app_id)
        .await
        .expect("dispatcher should have fired the invoke_async row");
    assert_eq!(marker["from"], "async");
}