PiCloud/crates/executor-core/tests/modules.rs

//! v1.1.3 — `PicloudModuleResolver` integration tests.
#![allow(clippy::needless_raw_string_hashes)] // r#""# is more uniform when many tests embed Rhai sources
//!
//! Each test wires an `Engine` with a `CountingModuleSource` (an
//! in-memory fake), a `Services` bundle, and an `ExecRequest` whose
//! `app_id` controls the cross-app boundary. The resolver is
//! exercised end-to-end through `Engine::execute`, so these tests
//! verify the same code path the `picloud` binary runs at request
//! time.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use picloud_executor_core::{Engine, ExecRequest, InvocationType, Limits};
use picloud_shared::{
    AppId, ExecutionId, GroupId, ModuleScript, ModuleSource, ModuleSourceError,
    NoopDeadLetterService, NoopDocsService, NoopEventEmitter, NoopHttpService, NoopKvService,
    RequestId, ScriptId, ScriptOwner, ScriptSandbox, Services,
};
use tokio::sync::Mutex;

/// In-memory `ModuleSource` backed by a `HashMap<(AppId, name)>`.
/// Tracks total lookup count so tests can assert cache hit/miss.
#[derive(Default)]
struct CountingModuleSource {
    table: Mutex<HashMap<(AppId, String), ModuleScript>>,
    lookups: AtomicUsize,
    /// When `Some`, every lookup returns this error instead of the
    /// table — used by the backend-error test.
    fail_with: Mutex<Option<String>>,
}

impl CountingModuleSource {
    fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    async fn put(self: &Arc<Self>, app_id: AppId, name: &str, source: &str) -> ScriptId {
        self.put_with_updated_at(app_id, name, source, Utc::now())
            .await
    }

    async fn put_with_updated_at(
        self: &Arc<Self>,
        app_id: AppId,
        name: &str,
        source: &str,
        updated_at: DateTime<Utc>,
    ) -> ScriptId {
        let script_id = ScriptId::new();
        self.table.lock().await.insert(
            (app_id, name.to_string()),
            ModuleScript {
                script_id,
                app_id: Some(app_id),
                group_id: None,
                name: name.to_string(),
                source: source.to_string(),
                updated_at,
            },
        );
        script_id
    }

    fn lookup_count(&self) -> usize {
        self.lookups.load(Ordering::SeqCst)
    }
}

#[async_trait]
impl ModuleSource for CountingModuleSource {
    async fn resolve(
        &self,
        origin: ScriptOwner,
        name: &str,
    ) -> Result<Option<ModuleScript>, ModuleSourceError> {
        self.lookups.fetch_add(1, Ordering::SeqCst);
        if let Some(err) = self.fail_with.lock().await.as_ref() {
            return Err(ModuleSourceError::Backend(err.clone()));
        }
        // This fake is flat/app-scoped — the inheritance + lexical
        // resolution semantics are covered by the CLI journey tests
        // against real Postgres. A group origin has no entries here.
        let app_id = match origin {
            ScriptOwner::App(a) => a,
            ScriptOwner::Group(_) => return Ok(None),
        };
        Ok(self
            .table
            .lock()
            .await
            .get(&(app_id, name.to_string()))
            .cloned())
    }
}

fn services_with(modules: Arc<dyn ModuleSource>) -> Services {
    Services::new(
        Arc::new(NoopKvService),
        Arc::new(NoopDocsService),
        Arc::new(NoopDeadLetterService),
        Arc::new(NoopEventEmitter),
        modules,
        Arc::new(NoopHttpService),
        Arc::new(picloud_shared::NoopFilesService),
        Arc::new(picloud_shared::NoopPubsubService),
        Arc::new(picloud_shared::NoopSecretsService),
        Arc::new(picloud_shared::NoopEmailService),
        Arc::new(picloud_shared::NoopUsersService),
        Arc::new(picloud_shared::NoopQueueService),
        Arc::new(picloud_shared::NoopInvokeService),
        Arc::new(picloud_shared::NoopVarsService),
    )
}

fn engine_with(modules: Arc<dyn ModuleSource>) -> Engine {
    Engine::new(Limits::default(), services_with(modules))
}

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

#[tokio::test(flavor = "multi_thread")]
async fn resolver_loads_simple_module() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    source.put(app_id, "math", "fn add(a, b) { a + b }").await;

    let engine = engine_with(source.clone());
    let resp = engine
        .execute(r#"import "math" as m; m::add(2, 3)"#, req(app_id))
        .expect("should execute");
    assert_eq!(resp.status_code, 200);
    assert_eq!(resp.body, serde_json::json!(5));
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_cross_app_blocked() {
    let source = CountingModuleSource::new();
    let app_a = AppId::new();
    let app_b = AppId::new();
    source
        .put(app_a, "secrets", "fn token() { \"A-token\" }")
        .await;
    source
        .put(app_b, "secrets", "fn token() { \"B-token\" }")
        .await;

    let engine = engine_with(source.clone());

    // App A sees A's module.
    let resp = engine
        .execute(r#"import "secrets" as s; s::token()"#, req(app_a))
        .unwrap();
    assert_eq!(resp.body, serde_json::json!("A-token"));

    // App B sees B's module — same name, completely separate value.
    let resp = engine
        .execute(r#"import "secrets" as s; s::token()"#, req(app_b))
        .unwrap();
    assert_eq!(resp.body, serde_json::json!("B-token"));
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_cross_app_module_not_found() {
    let source = CountingModuleSource::new();
    let app_a = AppId::new();
    let app_b = AppId::new();
    // Only app A has the module.
    source.put(app_a, "lonely", "fn ping() { \"pong\" }").await;

    // App B's lookup should return None → resolver surfaces
    // ErrorModuleNotFound.
    let engine = engine_with(source.clone());
    let err = engine
        .execute(r#"import "lonely" as l; l::ping()"#, req(app_b))
        .expect_err("cross-app import should fail");
    let msg = format!("{err:?}");
    assert!(
        msg.to_lowercase().contains("module")
            || msg.to_lowercase().contains("not found")
            || msg.to_lowercase().contains("lonely"),
        "expected module-not-found-flavoured error, got {msg}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_module_not_found() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    let engine = engine_with(source);

    let err = engine
        .execute(r#"import "doesnotexist" as x; 1"#, req(app_id))
        .expect_err("unknown module should fail");
    let msg = format!("{err:?}").to_lowercase();
    assert!(
        msg.contains("doesnotexist") || msg.contains("not found"),
        "expected ErrorModuleNotFound-flavoured error, got {msg}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_self_import_detected() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    // a imports itself
    source
        .put(app_id, "a", r#"import "a" as a; fn nope() { 0 }"#)
        .await;
    let engine = engine_with(source);

    let err = engine
        .execute(r#"import "a" as a; a::nope()"#, req(app_id))
        .expect_err("self-import should detect cycle");
    let msg = format!("{err:?}").to_lowercase();
    assert!(
        msg.contains("circular") || msg.contains("cycle"),
        "expected circular-import error, got {msg}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_circular_detected() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    // a imports b; b imports a; both then declare a fn.
    source
        .put(app_id, "a", r#"import "b" as b; fn x() { 0 }"#)
        .await;
    source
        .put(app_id, "b", r#"import "a" as a; fn y() { 0 }"#)
        .await;
    let engine = engine_with(source);

    let err = engine
        .execute(r#"import "a" as a; a::x()"#, req(app_id))
        .expect_err("circular import should fail");
    let msg = format!("{err:?}").to_lowercase();
    assert!(
        msg.contains("circular") || msg.contains("cycle"),
        "expected circular-import error, got {msg}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_depth_limit_enforced() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    // Chain `m0 -> m1 -> ... -> m9` (10 levels). Default depth limit is 8.
    for i in 0..9 {
        let next = format!("m{}", i + 1);
        source
            .put(
                app_id,
                &format!("m{i}"),
                &format!(r#"import "{next}" as nxt; fn x() {{ 0 }}"#),
            )
            .await;
    }
    source.put(app_id, "m9", "fn x() { 0 }").await;

    let engine = engine_with(source);
    let err = engine
        .execute(r#"import "m0" as m0; m0::x()"#, req(app_id))
        .expect_err("chain exceeding depth limit should fail");
    let msg = format!("{err:?}").to_lowercase();
    assert!(
        msg.contains("depth"),
        "expected depth-exceeded error, got {msg}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_depth_limit_just_under_succeeds() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    // Chain depth 7 (under default 8). m0 -> m1 -> ... -> m6 (terminal).
    for i in 0..6 {
        let next = format!("m{}", i + 1);
        source
            .put(
                app_id,
                &format!("m{i}"),
                &format!(r#"import "{next}" as nxt; fn x() {{ nxt::x() }}"#),
            )
            .await;
    }
    source.put(app_id, "m6", "fn x() { 42 }").await;

    let engine = engine_with(source);
    let resp = engine
        .execute(r#"import "m0" as m0; m0::x()"#, req(app_id))
        .expect("chain under depth limit should succeed");
    assert_eq!(resp.body, serde_json::json!(42));
}

#[tokio::test(flavor = "multi_thread")]
async fn resolver_runtime_validation_rejects_top_level_expr() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    // Module has a top-level expression — bypassed the admin gate,
    // but the resolver re-validates and rejects.
    source.put(app_id, "bad", r#"42; fn x() { 1 }"#).await;
    let engine = engine_with(source);

    let err = engine
        .execute(r#"import "bad" as b; b::x()"#, req(app_id))
        .expect_err("top-level expr in module should be rejected at resolve");
    let msg = format!("{err:?}").to_lowercase();
    assert!(
        msg.contains("top-level") || msg.contains("module"),
        "expected module-shape error, got {msg}"
    );
}

/// v1.1.4 §10a regression: the backend error must be REDACTED before
/// it reaches a script. The verbatim message (which can leak internal
/// infrastructure shape, e.g. "connection refused") must not appear;
/// the script sees only a stable generic.
#[tokio::test(flavor = "multi_thread")]
async fn resolver_backend_error_is_redacted_from_script() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    *source.fail_with.lock().await = Some("connection refused to 10.1.2.3:5432".into());
    let engine = engine_with(source);

    let err = engine
        .execute(r#"import "x" as x; 1"#, req(app_id))
        .expect_err("backend error should propagate");
    let msg = format!("{err:?}");
    assert!(
        msg.contains("module backend unavailable"),
        "expected redacted generic message, got {msg}"
    );
    assert!(
        !msg.contains("connection refused") && !msg.contains("10.1.2.3"),
        "redacted message must not leak the backend error, got {msg}"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn module_cache_hit_reuses_compiled_module() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    source.put(app_id, "u", "fn ping() { 1 }").await;

    let engine = engine_with(source.clone());

    // First execution compiles and caches.
    engine
        .execute(r#"import "u" as u; u::ping()"#, req(app_id))
        .unwrap();
    let lookups_after_first = source.lookup_count();
    assert_eq!(
        lookups_after_first, 1,
        "first invocation should look up once"
    );

    // Second execution should re-lookup (to compare updated_at) but
    // serve from cache without recompiling. We can't directly observe
    // compile-vs-cache here, but we can assert lookup count grew by
    // one (no spurious extra calls).
    engine
        .execute(r#"import "u" as u; u::ping()"#, req(app_id))
        .unwrap();
    assert_eq!(source.lookup_count(), 2);
}

#[tokio::test(flavor = "multi_thread")]
async fn module_cache_stale_invalidated_on_updated_at_change() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    let t0 = Utc::now() - chrono::Duration::seconds(10);
    source
        .put_with_updated_at(app_id, "u", r#"fn v() { 1 }"#, t0)
        .await;

    let engine = engine_with(source.clone());

    let resp = engine
        .execute(r#"import "u" as u; u::v()"#, req(app_id))
        .unwrap();
    assert_eq!(resp.body, serde_json::json!(1));

    // Replace with newer updated_at — cache should refresh.
    let t1 = Utc::now();
    source
        .put_with_updated_at(app_id, "u", r#"fn v() { 99 }"#, t1)
        .await;

    let resp = engine
        .execute(r#"import "u" as u; u::v()"#, req(app_id))
        .unwrap();
    assert_eq!(
        resp.body,
        serde_json::json!(99),
        "edited module should be visible on next invocation"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn module_cache_keyed_by_app() {
    let source = CountingModuleSource::new();
    let app_a = AppId::new();
    let app_b = AppId::new();
    source.put(app_a, "u", "fn id() { 1 }").await;
    source.put(app_b, "u", "fn id() { 2 }").await;

    let engine = engine_with(source.clone());

    // Both apps should compile + cache independently; neither sees
    // the other's compiled module.
    let resp = engine
        .execute(r#"import "u" as u; u::id()"#, req(app_a))
        .unwrap();
    assert_eq!(resp.body, serde_json::json!(1));
    let resp = engine
        .execute(r#"import "u" as u; u::id()"#, req(app_b))
        .unwrap();
    assert_eq!(resp.body, serde_json::json!(2));
}

#[tokio::test(flavor = "multi_thread")]
async fn module_cache_lru_evicts_when_capacity_exceeded() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    source.put(app_id, "a", "fn v() { 1 }").await;
    source.put(app_id, "b", "fn v() { 2 }").await;
    source.put(app_id, "c", "fn v() { 3 }").await;

    // Capacity 1 — only the most recently used entry stays cached.
    let engine =
        Engine::with_module_cache_capacity(Limits::default(), services_with(source.clone()), 1);

    engine
        .execute(r#"import "a" as m; m::v()"#, req(app_id))
        .unwrap();
    engine
        .execute(r#"import "b" as m; m::v()"#, req(app_id))
        .unwrap();
    engine
        .execute(r#"import "c" as m; m::v()"#, req(app_id))
        .unwrap();

    // Cache should hold at most one entry.
    let cache = engine.module_cache().lock().unwrap();
    assert!(
        cache.len() <= 1,
        "cache size {} exceeded capacity 1",
        cache.len()
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn endpoint_can_import_module() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    source
        .put(app_id, "helpers", r#"fn greet(name) { `hello, ${name}` }"#)
        .await;

    let engine = engine_with(source);
    let resp = engine
        .execute(
            r#"import "helpers" as h; #{ statusCode: 200, body: h::greet("world") }"#,
            req(app_id),
        )
        .unwrap();
    assert_eq!(resp.status_code, 200);
    assert_eq!(resp.body, serde_json::json!("hello, world"));
}

#[tokio::test(flavor = "multi_thread")]
async fn module_can_import_module() {
    let source = CountingModuleSource::new();
    let app_id = AppId::new();
    source.put(app_id, "inner", "fn three() { 3 }").await;
    source
        .put(
            app_id,
            "outer",
            r#"import "inner" as i; fn nine() { i::three() * 3 }"#,
        )
        .await;
    let engine = engine_with(source);

    let resp = engine
        .execute(r#"import "outer" as o; o::nine()"#, req(app_id))
        .unwrap();
    assert_eq!(resp.body, serde_json::json!(9));
}

#[test]
fn validate_module_accepts_fn_const_import_only() {
    let engine = Engine::new(Limits::default(), Services::default());
    let valid = r#"
        const PI = 3.14;
        import "other" as o;
        fn area(r) { PI * r * r }
    "#;
    let v = engine.validate_module(valid).expect("valid module body");
    assert_eq!(v.imports, vec!["other".to_string()]);
}

#[test]
fn validate_module_rejects_top_level_let() {
    let engine = Engine::new(Limits::default(), Services::default());
    let bad = "let x = 1; fn f() { x }";
    let err = engine
        .validate_module(bad)
        .expect_err("top-level let should be rejected");
    let msg = format!("{err:?}").to_lowercase();
    assert!(msg.contains("top-level") || msg.contains("module"));
}

#[test]
fn validate_module_rejects_top_level_expr() {
    let engine = Engine::new(Limits::default(), Services::default());
    let bad = "42";
    let err = engine
        .validate_module(bad)
        .expect_err("top-level expr should be rejected");
    let msg = format!("{err:?}").to_lowercase();
    assert!(msg.contains("top-level") || msg.contains("module"));
}

#[test]
fn validate_module_rejects_top_level_while() {
    // Avoid `if true { ... }` — Rhai folds constant-condition `if`s
    // at optimize time, leaving an empty statement list that passes
    // module-shape validation vacuously. A `while` with a variable
    // condition isn't folded.
    let engine = Engine::new(Limits::default(), Services::default());
    let bad = r#"let i = 0; while i < 1 { i += 1; }"#;
    let err = engine
        .validate_module(bad)
        .expect_err("top-level loop should be rejected");
    let msg = format!("{err:?}").to_lowercase();
    assert!(msg.contains("top-level") || msg.contains("module"));
}

#[test]
fn validate_endpoint_extracts_literal_imports() {
    let engine = Engine::new(Limits::default(), Services::default());
    let src = r#"
        import "a" as a;
        import "b" as b;
        a::run() + b::run()
    "#;
    let v = engine
        .validate(src)
        .expect("endpoint with imports should parse");
    assert_eq!(v.imports, vec!["a".to_string(), "b".to_string()]);
}

#[test]
fn validate_endpoint_top_level_expr_still_allowed() {
    // Endpoints can have arbitrary top-level statements — only
    // modules are restricted. Confirm v1.1.3 didn't tighten endpoints.
    let engine = Engine::new(Limits::default(), Services::default());
    let src = r#"let x = 1; #{ statusCode: 200, body: x }"#;
    engine
        .validate(src)
        .expect("endpoints may have top-level statements");
}

#[test]
fn validate_endpoint_skips_dynamic_imports_in_imports_list() {
    // `import some_var as y;` parses but is not a literal-path
    // import — the dep graph cannot track it. The imports list
    // should be empty for such a script.
    let engine = Engine::new(Limits::default(), Services::default());
    let src = r#"
        let name = "x";
        import name as y;
        y::run()
    "#;
    let v = engine.validate(src).expect("dynamic import should parse");
    assert!(
        v.imports.is_empty(),
        "dynamic imports should not appear in the dep-graph imports list, got {:?}",
        v.imports
    );
}

// ---------------------------------------------------------------------------
// Phase 4b — lexical (origin-aware) resolution.
//
// `LexicalModuleSource` keys modules by their *exact* defining node, with no
// chain walk. That's enough to prove the resolver passes the RIGHT origin:
// `default_origin` for the entry AST, and each module's own owner (decoded
// from Rhai's `_source`) for its nested imports. The chain-walk + group-tree
// semantics are covered end-to-end by the CLI journey tests against Postgres.
// ---------------------------------------------------------------------------

#[derive(Default)]
struct LexicalModuleSource {
    table: Mutex<HashMap<(ScriptOwner, String), ModuleScript>>,
}

impl LexicalModuleSource {
    fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    async fn put(self: &Arc<Self>, owner: ScriptOwner, name: &str, source: &str) {
        let (app_id, group_id) = match owner {
            ScriptOwner::App(a) => (Some(a), None),
            ScriptOwner::Group(g) => (None, Some(g)),
        };
        self.table.lock().await.insert(
            (owner, name.to_string()),
            ModuleScript {
                script_id: ScriptId::new(),
                app_id,
                group_id,
                name: name.to_string(),
                source: source.to_string(),
                updated_at: Utc::now(),
            },
        );
    }
}

#[async_trait]
impl ModuleSource for LexicalModuleSource {
    async fn resolve(
        &self,
        origin: ScriptOwner,
        name: &str,
    ) -> Result<Option<ModuleScript>, ModuleSourceError> {
        Ok(self
            .table
            .lock()
            .await
            .get(&(origin, name.to_string()))
            .cloned())
    }
}

fn req_with_owner(app_id: AppId, owner: ScriptOwner) -> ExecRequest {
    let mut r = req(app_id);
    r.script_owner = Some(owner);
    r
}

/// A group module's `import` resolves from the GROUP (its own defining
/// node), not the inheriting app — even when an app-owned module of the
/// same name exists. This is the §5.5 trust boundary: a leaf can't shadow
/// a module an inherited group script depends on.
#[tokio::test(flavor = "multi_thread")]
async fn lexical_nested_import_seals_to_module_owner() {
    let source = LexicalModuleSource::new();
    let app = AppId::new();
    let group = GroupId::new();

    // Two "inner" modules: the group's (real) and the app's (a trap).
    source
        .put(ScriptOwner::Group(group), "inner", "fn v() { 42 }")
        .await;
    source
        .put(ScriptOwner::App(app), "inner", "fn v() { 999 }")
        .await;
    // The group's "outer" imports "inner" — must bind the group's inner.
    source
        .put(
            ScriptOwner::Group(group),
            "outer",
            r#"import "inner" as i; fn val() { i::v() }"#,
        )
        .await;

    let engine = engine_with(source.clone());
    // Entry script runs as the inherited GROUP script (default_origin = group).
    let resp = engine
        .execute(
            r#"import "outer" as o; o::val()"#,
            req_with_owner(app, ScriptOwner::Group(group)),
        )
        .expect("should execute");
    assert_eq!(
        resp.body,
        serde_json::json!(42),
        "group module's import must seal to the group's `inner`, not the app's trap"
    );
}

/// The same registry, entered as an APP-owned script, reaches the app's
/// module — proving the fake distinguishes origins and the entry uses
/// `default_origin`.
#[tokio::test(flavor = "multi_thread")]
async fn lexical_entry_origin_selects_app_module() {
    let source = LexicalModuleSource::new();
    let app = AppId::new();
    let group = GroupId::new();
    source
        .put(ScriptOwner::Group(group), "inner", "fn v() { 42 }")
        .await;
    source
        .put(ScriptOwner::App(app), "inner", "fn v() { 999 }")
        .await;

    let engine = engine_with(source.clone());
    let resp = engine
        .execute(
            r#"import "inner" as i; i::v()"#,
            req_with_owner(app, ScriptOwner::App(app)),
        )
        .expect("should execute");
    assert_eq!(resp.body, serde_json::json!(999));
}

// ---------------------------------------------------------------------------
// §5.5 extension points — resolver handling of the policy outcomes.
//
// `PolicyModuleSource` is a flat fake that overrides `resolve_policy`: an EP
// name resolves dynamically against the inheriting app; a non-EP name resolves
// lexically from the importing origin. This verifies the resolver maps
// Module/NoProvider/NotFound correctly. The full chain semantics (default body
// up-chain, nearest-declaration tie) are covered by the CLI journey tests.
// ---------------------------------------------------------------------------

#[derive(Default)]
struct PolicyModuleSource {
    modules: Mutex<HashMap<(ScriptOwner, String), ModuleScript>>,
    eps: Mutex<HashSet<String>>,
}

impl PolicyModuleSource {
    fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }
    async fn put(self: &Arc<Self>, owner: ScriptOwner, name: &str, source: &str) {
        let (app_id, group_id) = match owner {
            ScriptOwner::App(a) => (Some(a), None),
            ScriptOwner::Group(g) => (None, Some(g)),
        };
        self.modules.lock().await.insert(
            (owner, name.to_string()),
            ModuleScript {
                script_id: ScriptId::new(),
                app_id,
                group_id,
                name: name.to_string(),
                source: source.to_string(),
                updated_at: Utc::now(),
            },
        );
    }
    async fn mark_ep(self: &Arc<Self>, name: &str) {
        self.eps.lock().await.insert(name.to_string());
    }
}

#[async_trait]
impl ModuleSource for PolicyModuleSource {
    async fn resolve(
        &self,
        origin: ScriptOwner,
        name: &str,
    ) -> Result<Option<ModuleScript>, ModuleSourceError> {
        Ok(self
            .modules
            .lock()
            .await
            .get(&(origin, name.to_string()))
            .cloned())
    }

    async fn resolve_policy(
        &self,
        origin: ScriptOwner,
        inheriting_app: AppId,
        name: &str,
    ) -> Result<picloud_shared::ModuleResolution, ModuleSourceError> {
        use picloud_shared::ModuleResolution;
        if self.eps.lock().await.contains(name) {
            // Extension point → dynamic, resolved against the inheriting app.
            return Ok(
                match self.resolve(ScriptOwner::App(inheriting_app), name).await? {
                    Some(m) => ModuleResolution::Module(m),
                    None => ModuleResolution::NoProvider,
                },
            );
        }
        Ok(match self.resolve(origin, name).await? {
            Some(m) => ModuleResolution::Module(m),
            None => ModuleResolution::NotFound,
        })
    }
}

/// An extension-point import binds the inheriting APP's module even though the
/// importing script's defining node is the group (dynamic override — the
/// inverse of the Phase 4b sealed/lexical import).
#[tokio::test(flavor = "multi_thread")]
async fn extension_point_resolves_app_override() {
    let source = PolicyModuleSource::new();
    let app = AppId::new();
    let group = GroupId::new();
    source.mark_ep("theme").await;
    // App provides its own `theme`; group has none.
    source
        .put(ScriptOwner::App(app), "theme", r#"fn color() { "red" }"#)
        .await;

    let engine = engine_with(source.clone());
    // Entry runs as the inherited GROUP endpoint importing the EP `theme`.
    let resp = engine
        .execute(
            r#"import "theme" as t; t::color()"#,
            req_with_owner(app, ScriptOwner::Group(group)),
        )
        .expect("should execute");
    assert_eq!(resp.body, serde_json::json!("red"));
}

/// An extension point with no provider for the app is a hard error.
#[tokio::test(flavor = "multi_thread")]
async fn extension_point_without_provider_errors() {
    let source = PolicyModuleSource::new();
    let app = AppId::new();
    let group = GroupId::new();
    source.mark_ep("theme").await; // declared, but no module anywhere.

    let engine = engine_with(source.clone());
    let err = engine
        .execute(
            r#"import "theme" as t; t::color()"#,
            req_with_owner(app, ScriptOwner::Group(group)),
        )
        .expect_err("missing provider must error");
    let msg = format!("{err:?}").to_lowercase();
    assert!(
        msg.contains("no provider") || msg.contains("extension point"),
        "expected a no-provider error, got {msg}"
    );
}

/// A non-extension-point name still resolves lexically from the origin.
#[tokio::test(flavor = "multi_thread")]
async fn non_extension_point_stays_lexical() {
    let source = PolicyModuleSource::new();
    let app = AppId::new();
    let group = GroupId::new();
    source
        .put(ScriptOwner::Group(group), "util", r#"fn v() { 7 }"#)
        .await;

    let engine = engine_with(source.clone());
    let resp = engine
        .execute(
            r#"import "util" as u; u::v()"#,
            req_with_owner(app, ScriptOwner::Group(group)),
        )
        .expect("should execute");
    assert_eq!(resp.body, serde_json::json!(7));
}