//! Bridge integration for the `http::*` SDK (v1.1.4).
//!
//! Runs a real Rhai engine under `spawn_blocking` against an in-memory
//! `HttpService` fake that records the last request and returns a
//! configured response (or error). This exercises the full bridge:
//! option parsing, body dispatch, response→map projection, the
//! throw-on-network-error / no-throw-on-non-2xx convention, and that
//! `cx.app_id` / `cx.script_id` are forwarded for attribution.

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use picloud_executor_core::{Engine, ExecRequest, InvocationType, Limits};
use picloud_shared::{
    AppId, ExecutionId, HttpError, HttpRequest, HttpResponse, HttpService, NoopDeadLetterService,
    NoopDocsService, NoopEventEmitter, NoopKvService, NoopModuleSource, RequestId, ScriptId,
    ScriptSandbox, Services,
};
use serde_json::{json, Value};

/// What the fake returns. Either a canned response or an error.
#[derive(Clone)]
enum Behavior {
    Respond(HttpResponse),
    Fail(String), // becomes HttpError::Network
}

#[derive(Default)]
struct Recorded {
    last: Option<HttpRequest>,
    last_app: Option<AppId>,
    last_script: Option<String>,
}

struct FakeHttp {
    behavior: Behavior,
    recorded: Mutex<Recorded>,
}

impl FakeHttp {
    fn responding(status: u16, content_type: &str, body: &str) -> Arc<Self> {
        let mut headers = BTreeMap::new();
        headers.insert("content-type".into(), content_type.into());
        Arc::new(Self {
            behavior: Behavior::Respond(HttpResponse {
                status,
                headers,
                body_raw: body.into(),
            }),
            recorded: Mutex::new(Recorded::default()),
        })
    }

    fn failing(msg: &str) -> Arc<Self> {
        Arc::new(Self {
            behavior: Behavior::Fail(msg.into()),
            recorded: Mutex::new(Recorded::default()),
        })
    }
}

#[async_trait]
impl HttpService for FakeHttp {
    async fn request(
        &self,
        cx: &picloud_shared::SdkCallCx,
        req: HttpRequest,
    ) -> Result<HttpResponse, HttpError> {
        {
            let mut r = self.recorded.lock().unwrap();
            r.last = Some(req.clone());
            r.last_app = Some(cx.app_id);
            r.last_script = Some(cx.script_id.to_string());
        }
        match &self.behavior {
            Behavior::Respond(resp) => Ok(resp.clone()),
            Behavior::Fail(msg) => Err(HttpError::Network(msg.clone())),
        }
    }
}

fn engine_with(http: Arc<dyn HttpService>) -> Arc<Engine> {
    let services = Services::new(
        Arc::new(NoopKvService),
        Arc::new(NoopDocsService),
        Arc::new(NoopDeadLetterService),
        Arc::new(NoopEventEmitter),
        Arc::new(NoopModuleSource),
        http,
        Arc::new(picloud_shared::NoopFilesService),
        Arc::new(picloud_shared::NoopPubsubService),
        Arc::new(picloud_shared::NoopSecretsService),
    );
    Arc::new(Engine::new(Limits::default(), services))
}

fn baseline_request(app_id: AppId, script_id: ScriptId) -> ExecRequest {
    let execution_id = ExecutionId::new();
    ExecRequest {
        execution_id,
        request_id: RequestId::new(),
        script_id,
        script_name: "http-test".into(),
        invocation_type: InvocationType::Http,
        path: "/http-test".into(),
        headers: BTreeMap::new(),
        body: Value::Null,
        params: BTreeMap::new(),
        query: BTreeMap::new(),
        rest: String::new(),
        sandbox_overrides: ScriptSandbox::default(),
        app_id,
        principal: None,
        trigger_depth: 0,
        root_execution_id: execution_id,
        is_dead_letter_handler: false,
        event: None,
    }
}

async fn run(engine: Arc<Engine>, src: &str, req: ExecRequest) -> Value {
    let src = src.to_string();
    tokio::task::spawn_blocking(move || engine.execute(&src, req))
        .await
        .expect("spawn_blocking should not panic")
        .expect("script execution should succeed")
        .body
}

async fn run_err(engine: Arc<Engine>, src: &str, req: ExecRequest) -> String {
    let src = src.to_string();
    let err = tokio::task::spawn_blocking(move || engine.execute(&src, req))
        .await
        .unwrap()
        .expect_err("script should throw");
    format!("{err:?}")
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn get_returns_status_and_json_body() {
    let http = FakeHttp::responding(200, "application/json", r#"{"ok":true,"n":7}"#);
    let engine = engine_with(http.clone());
    let src = r#"
        let r = http::get("https://api.example.com/x");
        #{ status: r.status, ok: r.body.ok, n: r.body.n }
    "#;
    let body = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert_eq!(body, json!({ "status": 200, "ok": true, "n": 7 }));
    // GET carries no body.
    assert!(http
        .recorded
        .lock()
        .unwrap()
        .last
        .as_ref()
        .unwrap()
        .body
        .is_none());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn non_json_body_stays_string() {
    let http = FakeHttp::responding(200, "text/plain", "plain text");
    let engine = engine_with(http);
    let src = r#"http::get("https://x/").body"#;
    let body = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert_eq!(body, json!("plain text"));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn empty_body_is_unit() {
    let http = FakeHttp::responding(204, "text/plain", "");
    let engine = engine_with(http);
    let src = r#"
        let r = http::get("https://x/");
        #{ is_unit: r.body == (), raw: r.body_raw }
    "#;
    let body = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert_eq!(body, json!({ "is_unit": true, "raw": "" }));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_map_body_is_json_encoded() {
    let http = FakeHttp::responding(200, "application/json", "{}");
    let engine = engine_with(http.clone());
    let src = r#"http::post("https://hooks/x", #{ text: "hello", n: 3 }).status"#;
    let _ = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    let rec = http.recorded.lock().unwrap();
    let req = rec.last.as_ref().unwrap();
    assert_eq!(req.method, "POST");
    assert_eq!(req.content_type.as_deref(), Some("application/json"));
    let sent: Value = serde_json::from_slice(req.body.as_ref().unwrap()).unwrap();
    assert_eq!(sent, json!({ "text": "hello", "n": 3 }));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_string_body_is_text_plain() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http.clone());
    let src = r#"http::post("https://x/", "raw payload").status"#;
    let _ = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    let rec = http.recorded.lock().unwrap();
    let req = rec.last.as_ref().unwrap();
    assert_eq!(req.content_type.as_deref(), Some("text/plain"));
    assert_eq!(req.body.as_deref(), Some(&b"raw payload"[..]));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_unit_body_sends_nothing() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http.clone());
    let src = r#"http::post("https://x/", ()).status"#;
    let _ = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert!(http
        .recorded
        .lock()
        .unwrap()
        .last
        .as_ref()
        .unwrap()
        .body
        .is_none());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn custom_headers_and_timeout_forwarded() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http.clone());
    let src = r#"
        http::get("https://x/", #{
            headers: #{ "Authorization": "Bearer t0ken" },
            timeout_ms: 4200,
        }).status
    "#;
    let _ = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    let rec = http.recorded.lock().unwrap();
    let req = rec.last.as_ref().unwrap();
    assert_eq!(
        req.headers.get("Authorization").map(String::as_str),
        Some("Bearer t0ken")
    );
    assert_eq!(req.timeout_ms, 4200);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unknown_option_key_throws() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http);
    let src = r#"http::get("https://x/", #{ timeoutms: 1000 })"#; // typo
    let err = run_err(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert!(err.contains("unknown option key"), "got {err}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn timeout_above_max_throws() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http);
    let src = r#"http::get("https://x/", #{ timeout_ms: 99999 })"#;
    let err = run_err(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert!(err.contains("maximum"), "got {err}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn non_2xx_does_not_throw() {
    let http = FakeHttp::responding(503, "text/plain", "down");
    let engine = engine_with(http);
    let src = r#"http::get("https://x/").status"#;
    let body = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert_eq!(body, json!(503));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn network_error_throws_with_http_prefix() {
    let http = FakeHttp::failing("connection refused");
    let engine = engine_with(http);
    let src = r#"http::get("https://x/")"#;
    let err = run_err(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert!(err.contains("http:"), "expected http: prefix, got {err}");
    assert!(err.contains("connection refused"), "got {err}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_form_url_encodes() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http.clone());
    let src = r#"http::post_form("https://x/login", #{ user: "alice", pw: "p@ss word" }).status"#;
    let _ = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    let rec = http.recorded.lock().unwrap();
    let req = rec.last.as_ref().unwrap();
    assert_eq!(
        req.content_type.as_deref(),
        Some("application/x-www-form-urlencoded")
    );
    let body = String::from_utf8(req.body.clone().unwrap()).unwrap();
    // order is map iteration order; assert both pairs present, encoded.
    assert!(body.contains("user=alice"), "got {body}");
    assert!(body.contains("pw=p%40ss+word"), "got {body}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn request_escape_hatch_arbitrary_method() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http.clone());
    let src = r#"http::request("OPTIONS", "https://x/").status"#;
    let _ = run(engine, src, baseline_request(AppId::new(), ScriptId::new())).await;
    assert_eq!(
        http.recorded.lock().unwrap().last.as_ref().unwrap().method,
        "OPTIONS"
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn default_user_agent_carries_script_id() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http.clone());
    let script_id = ScriptId::new();
    let src = r#"http::get("https://x/").status"#;
    let _ = run(engine, src, baseline_request(AppId::new(), script_id)).await;
    let rec = http.recorded.lock().unwrap();
    // The bridge forwards script_id on the request; the manager-core
    // impl turns it into the User-Agent. Here we assert the forward.
    assert_eq!(
        rec.last.as_ref().unwrap().script_id.as_deref(),
        Some(script_id.to_string().as_str())
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cx_app_id_forwarded_for_attribution() {
    let http = FakeHttp::responding(200, "text/plain", "ok");
    let engine = engine_with(http.clone());
    let app = AppId::new();
    let src = r#"http::get("https://x/").status"#;
    let _ = run(engine, src, baseline_request(app, ScriptId::new())).await;
    assert_eq!(http.recorded.lock().unwrap().last_app, Some(app));
}