fraiseql-server 2.16.0

use std::{collections::HashMap, sync::Arc};

use fraiseql_core::{runtime::CommittedMutation, schema::MutationOperation};
use fraiseql_functions::{
    FunctionDefinition, FunctionModule, FunctionObserver, RuntimeType, TriggerRegistry,
};
use serde_json::json;

use super::*;

fn module(name: &str) -> FunctionModule {
    FunctionModule {
        name:        name.to_string(),
        source_hash: "test".to_string(),
        bytecode:    bytes::Bytes::new(),
        runtime:     RuntimeType::Wasm,
    }
}

fn hooks(triggers: &[(&str, &str)], modules: &[&str]) -> BeforeMutationHooks {
    let definitions: Vec<FunctionDefinition> = triggers
        .iter()
        .map(|(name, trigger)| FunctionDefinition {
            name:        (*name).to_string(),
            trigger:     (*trigger).to_string(),
            runtime:     RuntimeType::Wasm,
            timeout_ms:  None,
            run_as:      None,
            when:        Vec::new(),
            re_runnable: false,
            retry:       None,
        })
        .collect();
    let trigger_registry =
        TriggerRegistry::load_from_definitions(&definitions).expect("valid trigger definitions");
    let module_registry: HashMap<String, FunctionModule> =
        modules.iter().map(|name| ((*name).to_string(), module(name))).collect();
    BeforeMutationHooks::new(trigger_registry, module_registry, Arc::new(FunctionObserver::new()))
}

/// Plan the dispatch for a committed `entity_type` write, as the engine would report it.
fn plan(
    hooks: &BeforeMutationHooks,
    entity_type: &str,
    operation: &MutationOperation,
    entity: &serde_json::Value,
) -> Vec<AfterMutationDispatch> {
    plan_after_mutation_dispatch(hooks, &committed(entity_type, operation, entity, 0))
}

fn committed<'a>(
    entity_type: &'a str,
    operation: &'a MutationOperation,
    entity: &'a serde_json::Value,
    dispatch_depth: u8,
) -> CommittedMutation<'a> {
    CommittedMutation {
        mutation_name: "aMutation",
        entity_type,
        operation,
        entity,
        entity_id: None,
        security_ctx: None,
        dispatch_depth,
    }
}

fn insert(table: &str) -> MutationOperation {
    MutationOperation::Insert {
        table: table.to_string(),
    }
}

#[test]
fn event_kind_maps_dml_verbs_and_skips_custom() {
    assert_eq!(event_kind_for(&insert("t")), Some(EventKind::Insert));
    assert_eq!(
        event_kind_for(&MutationOperation::Update {
            table: "t".to_string(),
        }),
        Some(EventKind::Update)
    );
    assert_eq!(
        event_kind_for(&MutationOperation::Delete {
            table: "t".to_string(),
        }),
        Some(EventKind::Delete)
    );
    assert_eq!(event_kind_for(&MutationOperation::Custom), None);
}

#[test]
fn plans_dispatch_for_matching_insert_trigger() {
    let hooks = hooks(&[("onUserCreated", "after:mutation:User:insert")], &["onUserCreated"]);
    let entity = json!({ "id": "u1", "name": "Ada" });

    let plans = plan(&hooks, "User", &insert("tb_user"), &entity);

    assert_eq!(plans.len(), 1);
    assert_eq!(plans[0].module.name, "onUserCreated");
    // The payload carries the new entity under `data.new`.
    let data = &plans[0].payload.data;
    assert_eq!(data["event_kind"], "insert");
    assert_eq!(data["new"]["name"], "Ada");
    assert!(data["old"].is_null());
}

#[test]
fn delete_reports_entity_as_old_not_new() {
    let hooks = hooks(&[("onUserDeleted", "after:mutation:User:delete")], &["onUserDeleted"]);
    let delete = MutationOperation::Delete {
        table: "tb_user".to_string(),
    };

    let plans = plan(&hooks, "User", &delete, &json!({ "id": "u1" }));

    assert_eq!(plans.len(), 1);
    let data = &plans[0].payload.data;
    assert_eq!(data["event_kind"], "delete");
    assert_eq!(data["old"]["id"], "u1");
    assert!(data["new"].is_null());
}

#[test]
fn custom_mutation_emits_no_dispatch() {
    // A trigger keyed on the entity exists, but a Custom op has no event kind.
    let hooks = hooks(&[("onAnything", "after:mutation:Report")], &["onAnything"]);

    let plans = plan(&hooks, "Report", &MutationOperation::Custom, &json!({ "id": "r1" }));

    assert!(plans.is_empty());
}

#[test]
fn non_matching_entity_emits_no_dispatch() {
    // Trigger is for Post, but the write produced a User.
    let hooks = hooks(&[("onPostCreated", "after:mutation:Post:insert")], &["onPostCreated"]);

    let plans = plan(&hooks, "User", &insert("tb_user"), &json!({ "id": "u1" }));

    assert!(plans.is_empty());
}

#[test]
fn wrong_event_kind_filter_emits_no_dispatch() {
    // Trigger only fires on update; the mutation is an insert.
    let hooks = hooks(&[("onUserUpdated", "after:mutation:User:update")], &["onUserUpdated"]);

    let plans = plan(&hooks, "User", &insert("tb_user"), &json!({ "id": "u1" }));

    assert!(plans.is_empty());
}

#[test]
fn all_kinds_trigger_matches_insert() {
    // No operation filter → matches every event kind for the entity.
    let hooks = hooks(&[("onUserChange", "after:mutation:User")], &["onUserChange"]);

    let plans = plan(&hooks, "User", &insert("tb_user"), &json!({ "id": "u1" }));

    assert_eq!(plans.len(), 1);
    assert_eq!(plans[0].module.name, "onUserChange");
}

#[test]
fn trigger_without_loaded_module_is_skipped() {
    // Trigger is registered but its module never loaded → dropped, not panicked.
    let hooks = hooks(&[("ghost", "after:mutation:User:insert")], &[]);

    let plans = plan(&hooks, "User", &insert("tb_user"), &json!({ "id": "u1" }));

    assert!(plans.is_empty());
}

#[test]
fn null_entity_yields_empty_new_but_still_dispatches() {
    let hooks = hooks(&[("onUserCreated", "after:mutation:User:insert")], &["onUserCreated"]);

    let plans = plan(&hooks, "User", &insert("tb_user"), &serde_json::Value::Null);

    assert_eq!(plans.len(), 1);
    assert!(plans[0].payload.data["new"].is_null());
}

/// M-bridge: a write a dispatched function or a scheduled source made through its query
/// bridge never dispatches, so a function writing the entity it is triggered by cannot
/// trigger itself. The same write at depth 0 (a request's) does.
#[cfg(feature = "functions-runtime")]
#[test]
fn a_bridge_write_is_not_dispatched() {
    let hooks = Arc::new(hooks(&[("onOrder", "after:mutation:Order")], &["onOrder"]));
    let observer = FunctionDispatchObserver::new(hooks);
    let update = MutationOperation::Update {
        table: "tb_order".to_string(),
    };
    let entity = json!({ "id": "o1" });

    assert_eq!(observer.plans_for(&committed("Order", &update, &entity, 0)).len(), 1);
    assert!(observer.plans_for(&committed("Order", &update, &entity, 1)).is_empty());
}

// ── after:ingest planning ───────────────────────────────────────────────────

#[test]
fn plan_after_ingest_matches_source_and_builds_payload() {
    use fraiseql_functions::{InboundMessage, IngestSource};

    let hooks = hooks(
        &[
            ("onStripe", "after:ingest:webhook:stripe"),
            ("onEmail", "after:ingest:email"),
        ],
        &["onStripe", "onEmail"],
    );
    let message = InboundMessage::new(
        IngestSource::Webhook {
            provider: "stripe".to_string(),
        },
        "evt_1",
        chrono::Utc::now(),
    );

    let plans = plan_after_ingest_dispatch(&hooks, &message);

    assert_eq!(plans.len(), 1);
    assert_eq!(plans[0].module.name, "onStripe");
    assert_eq!(plans[0].payload.trigger_type, "after:ingest:webhook:stripe");
}

#[test]
fn plan_after_ingest_skips_when_no_trigger_matches() {
    use fraiseql_functions::{InboundMessage, IngestSource};

    let hooks = hooks(&[("onEmail", "after:ingest:email")], &["onEmail"]);
    let message = InboundMessage::new(
        IngestSource::Webhook {
            provider: "stripe".to_string(),
        },
        "evt_1",
        chrono::Utc::now(),
    );

    assert!(plan_after_ingest_dispatch(&hooks, &message).is_empty());
}

// ── Durable dispatch: retry + DLQ + re_runnable opt-out ─────────────────────

#[cfg(feature = "functions-runtime")]
mod durable_dispatch {
    use fraiseql_functions::{EventPayload, FunctionObserver, ResourceLimits};
    use fraiseql_observers::{
        BackoffStrategy, DeadLetterQueue, DispatchPolicy, FailurePolicy, RetryConfig,
    };

    use super::{
        super::{DurableDispatcher, FunctionDispatchSetting, host_context_config},
        module,
    };
    use crate::observers::runtime::InMemoryDlq;

    /// A dispatcher whose observer has no registered runtime, so every
    /// `invoke_with_context` returns a permanent-less `Unsupported` error
    /// (`501` → not a client error → transient), letting durable dispatch retry.
    fn failing_dispatcher(dlq: std::sync::Arc<dyn DeadLetterQueue>) -> DurableDispatcher {
        keyed_failing_dispatcher(dlq, None)
    }

    fn keyed_failing_dispatcher(
        dlq: std::sync::Arc<dyn DeadLetterQueue>,
        idempotency_key: Option<std::sync::Arc<[u8]>>,
    ) -> DurableDispatcher {
        DurableDispatcher {
            observer: std::sync::Arc::new(FunctionObserver::new()),
            host_config: host_context_config(),
            limits: ResourceLimits::default(),
            dlq,
            source: fraiseql_observers::DispatchSource::AfterMutation,
            sender_resolver: None,
            email_transport: None,
            idempotency_key,
            query_executor_factory: None,
            run_as: None,
            caller: None,
        }
    }

    fn payload() -> EventPayload {
        EventPayload {
            trigger_type: "after:mutation:onUserCreated".to_string(),
            entity:       "User".to_string(),
            event_kind:   "insert".to_string(),
            data:         serde_json::json!({ "new": { "id": "u1" } }),
            timestamp:    chrono::Utc::now(),
        }
    }

    /// Zero-delay policy so retry tests run without real backoff waits.
    fn zero_delay_policy(max_attempts: u32) -> DispatchPolicy {
        DispatchPolicy::new(
            RetryConfig {
                max_attempts,
                initial_delay_ms: 0,
                max_delay_ms: 0,
                backoff_strategy: BackoffStrategy::Fixed,
            },
            FailurePolicy::Dlq,
        )
    }

    #[tokio::test]
    async fn durable_dispatch_dead_letters_after_exhausting_retries() {
        let dlq = std::sync::Arc::new(InMemoryDlq::new_with_max(None));
        let dispatcher = failing_dispatcher(dlq.clone());
        let setting = FunctionDispatchSetting {
            re_runnable: false,
            policy:      zero_delay_policy(3),
        };

        dispatcher.dispatch(module("onUserCreated"), payload(), &setting).await;

        assert_eq!(dlq.function_count(), 1, "an exhausted durable dispatch lands one DLQ row");
        let pending = dlq.get_pending_functions(10).await.expect("list pending function DLQ");
        assert_eq!(pending[0].function_name, "onUserCreated");
        assert_eq!(pending[0].attempts, 3, "the record captures every attempt made");
        assert_eq!(pending[0].trigger_type, "after:mutation:onUserCreated");
    }

    #[tokio::test]
    async fn dead_letter_carries_the_per_dispatch_idempotency_token() {
        let dlq = std::sync::Arc::new(InMemoryDlq::new_with_max(None));
        let dispatcher = failing_dispatcher(dlq.clone());
        let setting = FunctionDispatchSetting {
            re_runnable: false,
            policy:      zero_delay_policy(2),
        };
        let event = payload();

        dispatcher.dispatch(module("onUserCreated"), event.clone(), &setting).await;

        // The dispatcher derives the token ONCE and passes it to every attempt; it
        // is recorded on the dead-letter, and equals the pure derivation from the
        // dispatch's stable identity (source + function + trigger + payload data).
        // Same-token-every-attempt therefore holds by construction, and the derived
        // value is exactly what the guest's `fraiseql_idempotency_token()` returns.
        let expected = fraiseql_observers::derive_idempotency_token(
            None,
            fraiseql_observers::DispatchSource::AfterMutation,
            "onUserCreated",
            "after:mutation:onUserCreated:User:insert",
            &event.data,
        );
        let pending = dlq.get_pending_functions(10).await.expect("list pending function DLQ");
        assert_eq!(
            pending[0].idempotency_token, expected,
            "the dead-letter carries the derived per-dispatch token"
        );
        assert_eq!(expected.len(), 32, "the token is a 32-char hex send-id");
    }

    #[tokio::test]
    async fn keyed_dispatch_signs_the_idempotency_token() {
        let dlq = std::sync::Arc::new(InMemoryDlq::new_with_max(None));
        let key: std::sync::Arc<[u8]> = std::sync::Arc::from(
            fraiseql_observers::derive_idempotency_subkey(b"root-secret").as_slice(),
        );
        let dispatcher = keyed_failing_dispatcher(dlq.clone(), Some(key.clone()));
        let setting = FunctionDispatchSetting {
            re_runnable: false,
            policy:      zero_delay_policy(2),
        };
        let event = payload();

        dispatcher.dispatch(module("onUserCreated"), event.clone(), &setting).await;

        // With a key configured, the dispatched (and dead-lettered) token is the
        // HMAC-keyed derivation — and differs from the unsigned digest of the same
        // identity, proving the secret is actually applied end-to-end.
        let signed = fraiseql_observers::derive_idempotency_token(
            Some(&key[..]),
            fraiseql_observers::DispatchSource::AfterMutation,
            "onUserCreated",
            "after:mutation:onUserCreated:User:insert",
            &event.data,
        );
        let unsigned = fraiseql_observers::derive_idempotency_token(
            None,
            fraiseql_observers::DispatchSource::AfterMutation,
            "onUserCreated",
            "after:mutation:onUserCreated:User:insert",
            &event.data,
        );
        let pending = dlq.get_pending_functions(10).await.expect("list pending function DLQ");
        assert_eq!(pending[0].idempotency_token, signed, "the dispatch signs the token");
        assert_ne!(pending[0].idempotency_token, unsigned, "signed ≠ unsigned digest");
    }

    #[tokio::test]
    async fn re_runnable_dispatch_does_not_dead_letter() {
        let dlq = std::sync::Arc::new(InMemoryDlq::new_with_max(None));
        let dispatcher = failing_dispatcher(dlq.clone());
        let setting = FunctionDispatchSetting {
            re_runnable: true,
            policy:      zero_delay_policy(3),
        };

        dispatcher.dispatch(module("scoreDeal"), payload(), &setting).await;

        assert_eq!(
            dlq.function_count(),
            0,
            "a re-runnable dispatch is fire-and-forget: no retry, no DLQ"
        );
    }

    #[test]
    fn default_setting_is_durable() {
        // ADR 0015: dispatch is durable by default; re_runnable is the opt-out.
        assert!(!FunctionDispatchSetting::default().re_runnable);
    }
}

// ── Config surface: env overrides + per-function resolution ─────────────────

#[cfg(feature = "functions-runtime")]
mod dispatch_config {
    use std::collections::HashMap;

    use fraiseql_functions::{FunctionDefinition, RuntimeType};
    use fraiseql_observers::RetryConfig;

    use super::super::{DispatchDefaults, DlqStoreKind, resolve_dispatch_settings};

    #[test]
    fn dlq_store_resolves_from_compiled_value_and_env_override() {
        let no_env = |_: &str| None;

        // Compiled value, no env: honoured (case/space-insensitive).
        assert_eq!(DlqStoreKind::resolve(Some("postgres"), no_env), DlqStoreKind::Postgres);
        assert_eq!(DlqStoreKind::resolve(Some("  Postgres "), no_env), DlqStoreKind::Postgres);
        assert_eq!(DlqStoreKind::resolve(Some("memory"), no_env), DlqStoreKind::Memory);

        // Absent everywhere → the in-memory default.
        assert_eq!(DlqStoreKind::resolve(None, no_env), DlqStoreKind::Memory);

        // Env overrides the compiled value (production tuning without recompiling).
        let env_pg =
            |key: &str| (key == "FRAISEQL_FUNCTIONS_DLQ_STORE").then(|| "postgres".to_string());
        assert_eq!(DlqStoreKind::resolve(Some("memory"), env_pg), DlqStoreKind::Postgres);
        let env_mem =
            |key: &str| (key == "FRAISEQL_FUNCTIONS_DLQ_STORE").then(|| "memory".to_string());
        assert_eq!(DlqStoreKind::resolve(Some("postgres"), env_mem), DlqStoreKind::Memory);

        // Unknown value → fail-safe to memory (never a startup failure).
        assert_eq!(DlqStoreKind::resolve(Some("redis"), no_env), DlqStoreKind::Memory);
    }

    fn definition(name: &str, re_runnable: bool, retry: Option<RetryConfig>) -> FunctionDefinition {
        FunctionDefinition {
            name: name.to_string(),
            trigger: "after:mutation:Entity:insert".to_string(),
            runtime: RuntimeType::Wasm,
            timeout_ms: None,
            run_as: None,
            when: Vec::new(),
            re_runnable,
            retry,
        }
    }

    #[test]
    fn env_overrides_layer_over_retry_defaults() {
        let env: HashMap<&str, &str> = [
            ("FRAISEQL_FUNCTIONS_RETRY_MAX_ATTEMPTS", "9"),
            ("FRAISEQL_FUNCTIONS_DLQ_MAX_SIZE", "500"),
        ]
        .into_iter()
        .collect();

        let defaults =
            DispatchDefaults::from_getter(|key| env.get(key).map(|value| (*value).to_string()));

        assert_eq!(defaults.retry.max_attempts, 9, "env overrides the default attempts");
        assert_eq!(defaults.dlq_max_size, Some(500), "env sets the DLQ cap");
        // An untouched knob keeps the library default.
        assert_eq!(defaults.retry.initial_delay_ms, RetryConfig::default().initial_delay_ms);
    }

    #[test]
    fn unset_env_uses_library_defaults() {
        let defaults = DispatchDefaults::from_getter(|_| None);
        assert_eq!(defaults.retry.max_attempts, RetryConfig::default().max_attempts);
        assert_eq!(defaults.dlq_max_size, None, "unbounded DLQ when unset");
    }

    #[test]
    fn resolution_maps_re_runnable_and_per_function_retry() {
        let defaults = DispatchDefaults::from_getter(|_| None); // library defaults
        let per_function = RetryConfig {
            max_attempts: 5,
            ..RetryConfig::default()
        };
        let definitions = vec![
            definition("scoreDeal", true, None),
            definition("chargeCard", false, Some(per_function)),
            definition("sendEmail", false, None),
        ];

        let settings = resolve_dispatch_settings(&definitions, &defaults);

        assert!(settings["scoreDeal"].re_runnable, "re_runnable carried through");
        assert!(!settings["chargeCard"].re_runnable);
        assert_eq!(
            settings["chargeCard"].policy.retry.max_attempts, 5,
            "explicit per-function retry is used"
        );
        assert_eq!(
            settings["sendEmail"].policy.retry.max_attempts, defaults.retry.max_attempts,
            "a function with no retry inherits the default"
        );
    }
}

// ── #594: the fraiseql_query bridge wiring on the dispatched host ────────────

#[cfg(feature = "functions-runtime")]
mod query_bridge_wiring {
    #![allow(clippy::unwrap_used)] // Reason: test module — mutex locks are infallible here

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

    use fraiseql_core::security::SecurityContext;
    use fraiseql_functions::{
        EventPayload, FunctionObserver, HostContext, ResourceLimits, RunAs,
        host::live::QueryExecutor,
    };

    use super::super::{DurableDispatcher, QueryExecutorFactory, host_context_config};
    use crate::observers::runtime::InMemoryDlq;

    /// A mock executor returning a canned result. The identity it runs under is
    /// captured by the factory closure below (that is where the `run_as` identity is
    /// resolved), so this executor itself is stateless.
    struct RecordingExecutor;

    impl QueryExecutor for RecordingExecutor {
        fn execute_query(
            &self,
            _query: &str,
            _variables: Option<&serde_json::Value>,
        ) -> std::pin::Pin<
            Box<
                dyn std::future::Future<Output = fraiseql_error::Result<serde_json::Value>>
                    + Send
                    + '_,
            >,
        > {
            Box::pin(async { Ok(serde_json::json!({ "data": { "ok": true } })) })
        }
    }

    /// Build a factory that records the identity each dispatched host runs under.
    fn recording_factory() -> (QueryExecutorFactory, Arc<Mutex<Option<SecurityContext>>>) {
        let captured: Arc<Mutex<Option<SecurityContext>>> = Arc::new(Mutex::new(None));
        let sink = Arc::clone(&captured);
        let factory: QueryExecutorFactory = Arc::new(move |identity: SecurityContext| {
            *sink.lock().unwrap() = Some(identity);
            Arc::new(RecordingExecutor) as Arc<dyn QueryExecutor>
        });
        (factory, captured)
    }

    fn dispatcher(
        query_executor_factory: Option<QueryExecutorFactory>,
        run_as: Option<RunAs>,
    ) -> DurableDispatcher {
        DurableDispatcher {
            observer: Arc::new(FunctionObserver::new()),
            host_config: host_context_config(),
            limits: ResourceLimits::default(),
            dlq: Arc::new(InMemoryDlq::new_with_max(None)),
            source: fraiseql_observers::DispatchSource::AfterMutation,
            sender_resolver: None,
            email_transport: None,
            idempotency_key: None,
            query_executor_factory,
            run_as,
            caller: None,
        }
    }

    fn payload() -> EventPayload {
        EventPayload {
            trigger_type: "after:mutation:recordApproval".to_string(),
            entity:       "Order".to_string(),
            event_kind:   "update".to_string(),
            data:         serde_json::json!({ "new": { "id": 1 } }),
            timestamp:    chrono::Utc::now(),
        }
    }

    #[tokio::test]
    async fn without_a_factory_the_bridge_is_unconfigured() {
        // No factory ⇒ the dispatched host has no query executor, so a function's
        // `fraiseql_query` fails "query executor not configured" — the pre-#594
        // behavior, preserved for a server with no request-path executor.
        let host = dispatcher(None, None).build_host("notify", payload(), "tok-1");
        let err = host
            .query("mutation { x }", serde_json::json!({}))
            .await
            .expect_err("no executor");
        assert!(err.to_string().contains("query executor not configured"));
    }

    #[tokio::test]
    async fn a_factory_wires_the_bridge_under_the_run_as_ceiling() {
        // #594: with a factory, the dispatched host CAN issue `fraiseql_query`, and
        // it runs under this function's `run_as` identity (audited as
        // `system_job:<function-name>`).
        let (factory, captured) = recording_factory();
        let run_as = RunAs {
            roles:  vec!["order_writer".to_string()],
            scopes: vec!["write:order".to_string()],
            tenant: Some("acme".to_string()),
        };
        let host = dispatcher(Some(factory), Some(run_as)).build_host(
            "recordApproval",
            payload(),
            "tok-2",
        );

        let value = host
            .query("mutation { recordApproval(id: 1) { id } }", serde_json::json!({}))
            .await
            .expect("the bridge is wired");
        assert_eq!(value, serde_json::json!({ "data": { "ok": true } }));

        // The write ran under the function's ceiling, audited as its system job.
        let identity = captured.lock().unwrap().clone().expect("identity captured");
        assert_eq!(identity.user_id.0, "system_job:recordApproval");
        assert!(identity.has_role("order_writer"));
        assert!(identity.has_scope("write:order"));
        assert_eq!(identity.request_id, "tok-2", "identity correlates the dispatch token");
    }

    /// #840: the after:mutation host config has an env-var allowlist producer —
    /// `FRAISEQL_FUNCTIONS_ALLOWED_ENV_VARS`, mirroring the sibling
    /// `FRAISEQL_FUNCTIONS_ALLOWED_DOMAINS` escape hatch.
    #[test]
    fn host_context_config_reads_the_env_var_allowlist() {
        use super::super::host_context_config_from;

        // No producer key set → deny-by-default (empty).
        let config = host_context_config_from(|_| None);
        assert!(config.allowed_env_vars.is_empty());

        // Comma-split and trimmed.
        let config = host_context_config_from(|key| {
            (key == "FRAISEQL_FUNCTIONS_ALLOWED_ENV_VARS")
                .then(|| " LLM_API_KEY, QONTO_API_KEY ".to_string())
        });
        assert!(config.allowed_env_vars.contains("LLM_API_KEY"));
        assert!(config.allowed_env_vars.contains("QONTO_API_KEY"));
        assert_eq!(config.allowed_env_vars.len(), 2);
    }

    // ── #803: the dispatched host's identity ─────────────────────────────────

    /// #803: with no triggering caller, the dispatched host's `auth_context`
    /// reflects the function's own `run_as` identity — never the pre-fix
    /// hard-coded `anonymous` placeholder.
    #[tokio::test]
    async fn dispatched_host_auth_context_reflects_the_run_as_identity() {
        let run_as = RunAs {
            roles:  vec!["order_writer".to_string()],
            scopes: vec!["write:order".to_string()],
            tenant: Some("acme".to_string()),
        };
        let host = dispatcher(None, Some(run_as)).build_host("recordApproval", payload(), "tok-3");
        let auth = host.auth_context().expect("auth context");
        assert_eq!(auth["sub"], "system_job:recordApproval");
        assert_eq!(auth["tenant_id"], "acme");
        assert_eq!(auth["roles"][0], "order_writer");
    }

    /// #803: an after:mutation dispatch triggered by an authenticated request
    /// surfaces THAT caller in `auth_context` (docs/architecture/functions.md:
    /// "the caller's authenticated context") — while the query bridge keeps the
    /// function's `run_as` ceiling.
    #[tokio::test]
    async fn dispatched_host_auth_context_prefers_the_triggering_caller() {
        let caller = SecurityContext {
            user_id:          fraiseql_core::types::UserId("user-42".to_string()),
            roles:            vec!["sales_rep".to_string()],
            scopes:           vec![],
            tenant_id:        Some(fraiseql_core::types::TenantId::new("tenant-7")),
            expires_at:       chrono::Utc::now() + chrono::Duration::hours(1),
            authenticated_at: chrono::Utc::now(),
            request_id:       "req-42".to_string(),
            ip_address:       None,
            attributes:       std::collections::HashMap::from([
                ("tenant_id".to_string(), serde_json::json!("tenant-7")),
                ("org_id".to_string(), serde_json::json!("tenant-7")),
            ]),
            issuer:           None,
            audience:         None,
            email:            Some("rep@outreach.example".to_string()),
            display_name:     Some("Sales Rep".to_string()),
        };
        let (factory, captured) = recording_factory();
        let run_as = RunAs {
            roles:  vec!["order_writer".to_string()],
            scopes: vec![],
            tenant: Some("acme".to_string()),
        };
        let mut dispatcher = dispatcher(Some(factory), Some(run_as));
        dispatcher.caller = Some(caller);
        let host = dispatcher.build_host("recordApproval", payload(), "tok-4");

        // The host identity is the triggering caller…
        let auth = host.auth_context().expect("auth context");
        assert_eq!(auth["sub"], "user-42");
        assert_eq!(auth["email"], "rep@outreach.example");
        assert_eq!(auth["tenant_id"], "tenant-7");

        // …while the query bridge stays under the function's run_as ceiling.
        let _ = host.query("mutation { x }", serde_json::json!({})).await;
        let identity = captured.lock().unwrap().clone().expect("identity captured");
        assert_eq!(identity.user_id.0, "system_job:recordApproval");
        assert!(identity.has_role("order_writer"));
    }

    /// #803: `send_email` succeeds on the dispatch path once the host carries
    /// the triggering caller's verified identity — pre-fix it refused
    /// unconditionally (the anonymous placeholder has no sending address), so
    /// the whole `build_send_email_wiring` was dead on arrival.
    #[tokio::test]
    async fn dispatched_host_send_email_succeeds_for_an_authenticated_caller() {
        use fraiseql_functions::{
            LoginEmailSender, SendContext, SendEmailRequest, SendEmailResponse, SenderIdentity,
        };

        /// Records the sender + tenant the host resolved, returns success.
        struct RecordingTransport {
            seen: Arc<Mutex<Option<(SenderIdentity, Option<String>)>>>,
        }
        impl fraiseql_functions::EmailTransport for RecordingTransport {
            fn send<'a>(
                &'a self,
                sender: &'a SenderIdentity,
                _request: &'a SendEmailRequest,
                context: SendContext<'a>,
            ) -> fraiseql_functions::outbound::BoxFuture<
                'a,
                fraiseql_error::Result<SendEmailResponse>,
            > {
                let seen = Arc::clone(&self.seen);
                let sender = sender.clone();
                let tenant = context.tenant.map(ToString::to_string);
                Box::pin(async move {
                    *seen.lock().unwrap() = Some((sender, tenant));
                    Ok(SendEmailResponse {
                        accepted:   true,
                        message_id: Some("msg-1".to_string()),
                    })
                })
            }
        }

        let caller = SecurityContext {
            user_id:          fraiseql_core::types::UserId("user-42".to_string()),
            roles:            vec![],
            scopes:           vec![],
            tenant_id:        Some(fraiseql_core::types::TenantId::new("tenant-7")),
            expires_at:       chrono::Utc::now() + chrono::Duration::hours(1),
            authenticated_at: chrono::Utc::now(),
            request_id:       "req-42".to_string(),
            ip_address:       None,
            attributes:       std::collections::HashMap::from([
                ("tenant_id".to_string(), serde_json::json!("tenant-7")),
                ("org_id".to_string(), serde_json::json!("tenant-7")),
            ]),
            issuer:           None,
            audience:         None,
            email:            Some("rep@outreach.example".to_string()),
            display_name:     Some("Sales Rep".to_string()),
        };

        let seen = Arc::new(Mutex::new(None));
        let mut dispatcher = dispatcher(None, None);
        dispatcher.sender_resolver = Some(Arc::new(LoginEmailSender));
        dispatcher.email_transport = Some(Arc::new(RecordingTransport {
            seen: Arc::clone(&seen),
        }));
        dispatcher.caller = Some(caller);
        let host = dispatcher.build_host("followUp", payload(), "tok-5");

        let response = host
            .send_email(&SendEmailRequest {
                to:       "prospect@example.com".to_string(),
                subject:  "hello".to_string(),
                text:     Some("world".to_string()),
                html:     None,
                reply_to: None,
            })
            .await
            .expect("send_email succeeds for an authenticated caller with a verified address");
        assert!(response.accepted);

        let (sender, tenant) = seen.lock().unwrap().clone().expect("transport reached");
        assert_eq!(sender.address, "rep@outreach.example", "the from is the caller's address");
        assert_eq!(
            tenant.as_deref(),
            Some("tenant-7"),
            "the send-status/suppression rows are scoped to the caller's tenant, not NULL"
        );
    }

    #[tokio::test]
    async fn after_ingest_bridge_runs_under_the_run_as_ceiling() {
        // #594 (ingest half): the same `run_as`→bridge wiring proven above for
        // after:mutation, but driven by an inbound-message-shaped payload — the
        // normalized `InboundMessage` a Source adapter puts on the durable spine and
        // that `spawn_after_ingest` hands to the dispatcher. This bypasses the
        // Postgres spine (`inbound::spine`, tested separately) and drives `build_host`
        // directly, exactly as the after:mutation case bypasses the HTTP route: the
        // dispatched host's `fraiseql_query` write must run under the after:ingest
        // function's `run_as` ceiling, audited as `system_job:<function-name>`.
        use fraiseql_functions::{InboundMessage, IngestSource, IngestTrigger};

        let (factory, captured) = recording_factory();
        let run_as = RunAs {
            roles:  vec!["ticket_writer".to_string()],
            scopes: vec!["write:ticket".to_string()],
            tenant: Some("acme".to_string()),
        };

        // The event payload an after:ingest function receives: the whole normalized
        // message rendered by the real `IngestTrigger::build_payload`, so its shape
        // (`after:ingest:<source>` trigger type, `event_kind = "ingest"`) matches what
        // the spine dispatches — not a hand-built mutation payload.
        let message = InboundMessage::new(
            IngestSource::Webhook {
                provider: "stripe".to_string(),
            },
            "evt_9f8",
            chrono::Utc::now(),
        );
        let payload = IngestTrigger {
            function_name: "ingestPayment".to_string(),
            source:        None,
        }
        .build_payload(&message);
        assert_eq!(payload.trigger_type, "after:ingest:webhook:stripe");
        assert_eq!(payload.event_kind, "ingest");

        // Tag the dispatcher AfterIngest (its DLQ/metrics identity) and drive it.
        let mut dispatcher = dispatcher(Some(factory), Some(run_as));
        dispatcher.source = fraiseql_observers::DispatchSource::AfterIngest;
        let host = dispatcher.build_host("ingestPayment", payload, "ingest-tok-1");

        let value = host
            .query("mutation { recordPayment(event: \"evt_9f8\") { id } }", serde_json::json!({}))
            .await
            .expect("the after:ingest bridge is wired");
        assert_eq!(value, serde_json::json!({ "data": { "ok": true } }));

        // The write ran under the after:ingest function's ceiling, audited as its
        // system job — identity carries the granted roles/scopes and the tenant.
        let identity = captured.lock().unwrap().clone().expect("identity captured");
        assert_eq!(identity.user_id.0, "system_job:ingestPayment");
        assert!(identity.has_role("ticket_writer"));
        assert!(identity.has_scope("write:ticket"));
        assert_eq!(
            identity.tenant_id.as_ref().map(|tenant| tenant.0.as_str()),
            Some("acme"),
            "the ceiling's tenant scopes the ingest write"
        );
        assert_eq!(identity.request_id, "ingest-tok-1", "identity correlates the dispatch token");
    }

    #[tokio::test]
    async fn a_function_without_run_as_is_fail_closed() {
        // A factory but no `run_as` ⇒ the bridge is wired but runs under an anonymous
        // system_job with no authority: RLS/field-authz deny writes.
        let (factory, captured) = recording_factory();
        let host = dispatcher(Some(factory), None).build_host("purge", payload(), "tok-3");
        let _ = host.query("query { me { id } }", serde_json::json!({})).await;

        let identity = captured.lock().unwrap().clone().expect("identity captured");
        assert_eq!(identity.user_id.0, "system_job:purge");
        assert!(identity.roles.is_empty(), "fail-closed: no roles");
        assert!(identity.scopes.is_empty(), "fail-closed: no scopes");
        assert!(identity.tenant_id.is_none(), "fail-closed: no tenant");
    }
}

// ── #597: `when` predicates gate the planner (no dispatch on non-match) ──────

#[test]
fn when_predicate_produces_no_dispatch_on_non_matching_update() {
    use fraiseql_functions::{RuntimeType, triggers::TriggerPredicate};

    // A function that only fires when `status` transitions to "approved".
    let mut def = FunctionDefinition::new(
        "notify_approved",
        "after:mutation:Order:update",
        RuntimeType::Wasm,
    );
    def.when = vec![TriggerPredicate {
        field:      "status".to_string(),
        eq:         None,
        changed_to: Some(json!("approved")),
    }];

    let registry = TriggerRegistry::load_from_definitions(&[def]).expect("valid when");
    let module_registry: HashMap<String, FunctionModule> =
        HashMap::from([("notify_approved".to_string(), module("notify_approved"))]);
    let hooks =
        BeforeMutationHooks::new(registry, module_registry, Arc::new(FunctionObserver::new()));

    let update = MutationOperation::Update {
        table: "tb_order".to_string(),
    };

    // The after:mutation path carries only the after-image (no pre-image), so
    // `changed_to` gates on `new.status == v`. An update whose result is NOT the
    // target value produces no dispatch record at all (predicate false).
    let other_value = json!({ "id": "o1", "status": "rejected" });
    assert!(
        plan_after_mutation_dispatch(&hooks, &committed("Order", &update, &other_value, 0))
            .is_empty(),
        "a predicate-false update produces no dispatch (no record, not a skipped dispatch)"
    );

    // An update to the target value fires. (The pre-image is absent here, so
    // `changed_to` cannot distinguish a transition from a re-save — full transition
    // detection needs the pre-image, i.e. the after:capture path with
    // `pre_image=True`. Documented in functions.md.) The row is the function's whole
    // entity, so the predicate reads `status` whether or not a client selected it.
    let to_target = json!({ "id": "o1", "status": "approved" });
    assert_eq!(
        plan_after_mutation_dispatch(&hooks, &committed("Order", &update, &to_target, 0)).len(),
        1,
        "an update to the target value fires the predicate function"
    );
}

// ── #366: after:capture dispatch (loop-safe, predicate-aware) ───────────────

#[cfg(feature = "functions-runtime")]
mod after_capture {
    #![allow(clippy::unwrap_used)] // Reason: test module

    use fraiseql_functions::RuntimeType;
    use fraiseql_observers::{
        DispatchSource, EntityEvent as ObserverEntityEvent, EventKind as ObserverEventKind,
    };

    use super::{
        super::{
            CAPTURED_WRITE_MARKER, dispatch_idempotency_token, observer_event_to_capture,
            plan_after_capture_dispatch,
        },
        *,
    };

    fn capture_hooks(name: &str, trigger: &str) -> BeforeMutationHooks {
        let def = FunctionDefinition::new(name, trigger, RuntimeType::Wasm);
        let registry =
            TriggerRegistry::load_from_definitions(&[def]).expect("valid capture trigger");
        let modules: HashMap<String, FunctionModule> =
            HashMap::from([(name.to_string(), module(name))]);
        BeforeMutationHooks::new(registry, modules, Arc::new(FunctionObserver::new()))
    }

    fn captured_insert(
        entity: &str,
        data: serde_json::Value,
        cdc: Option<&str>,
    ) -> ObserverEntityEvent {
        let mut e = ObserverEntityEvent::new(
            ObserverEventKind::Created,
            entity.to_string(),
            uuid::Uuid::new_v4(),
            data,
        );
        e.cdc_source = cdc.map(String::from);
        e
    }

    #[test]
    fn dispatches_only_genuinely_captured_writes() {
        let hooks = capture_hooks("onCapture", "after:capture:Order:insert");
        let data = json!({ "id": "o1", "status": "new" });

        // A captured write (cdc_source = fallback_trigger) → one after:capture plan.
        let captured = captured_insert("Order", data.clone(), Some(CAPTURED_WRITE_MARKER));
        let (event, cdc) = observer_event_to_capture(&captured).expect("insert converts");
        let plans = plan_after_capture_dispatch(&hooks, &event, cdc.as_deref());
        assert_eq!(plans.len(), 1, "a captured write drives after:capture");
        assert_eq!(plans[0].payload.trigger_type, "after:capture:onCapture");
        assert_eq!(plans[0].payload.data["new"]["id"], "o1");

        // An executor/mediated write (no marker) → NO dispatch (loop safety).
        let executor_write = captured_insert("Order", data, None);
        let (event, cdc) = observer_event_to_capture(&executor_write).expect("insert converts");
        assert!(
            plan_after_capture_dispatch(&hooks, &event, cdc.as_deref()).is_empty(),
            "a non-captured (executor) write never dispatches after:capture — loop safety"
        );
    }

    #[test]
    fn delete_reports_removed_row_as_old() {
        let hooks = capture_hooks("onDelete", "after:capture:Order:delete");
        let mut ev = ObserverEntityEvent::new(
            ObserverEventKind::Deleted,
            "Order".to_string(),
            uuid::Uuid::new_v4(),
            json!({ "id": "o9" }),
        );
        ev.cdc_source = Some(CAPTURED_WRITE_MARKER.to_string());

        let (event, cdc) = observer_event_to_capture(&ev).expect("delete converts");
        let plans = plan_after_capture_dispatch(&hooks, &event, cdc.as_deref());
        assert_eq!(plans.len(), 1);
        assert_eq!(
            plans[0].payload.data["old"]["id"], "o9",
            "delete reports the removed row as old"
        );
        assert!(plans[0].payload.data["new"].is_null());
    }

    #[test]
    fn predicates_compose_on_capture_payloads() {
        use fraiseql_functions::triggers::TriggerPredicate;
        // A capture function that only fires when status == "shipped".
        let mut def =
            FunctionDefinition::new("onShipped", "after:capture:Order:insert", RuntimeType::Wasm);
        def.when = vec![TriggerPredicate {
            field:      "status".to_string(),
            eq:         Some(json!("shipped")),
            changed_to: None,
        }];
        let registry = TriggerRegistry::load_from_definitions(&[def]).expect("valid");
        let modules: HashMap<String, FunctionModule> =
            HashMap::from([("onShipped".to_string(), module("onShipped"))]);
        let hooks = BeforeMutationHooks::new(registry, modules, Arc::new(FunctionObserver::new()));

        // status=new → predicate false → no dispatch.
        let e = captured_insert("Order", json!({ "status": "new" }), Some(CAPTURED_WRITE_MARKER));
        let (ev, cdc) = observer_event_to_capture(&e).unwrap();
        assert!(plan_after_capture_dispatch(&hooks, &ev, cdc.as_deref()).is_empty());

        // status=shipped → predicate true → dispatch.
        let e =
            captured_insert("Order", json!({ "status": "shipped" }), Some(CAPTURED_WRITE_MARKER));
        let (ev, cdc) = observer_event_to_capture(&e).unwrap();
        assert_eq!(plan_after_capture_dispatch(&hooks, &ev, cdc.as_deref()).len(), 1);
    }

    // #366: a capture dispatch's idempotency token is stable per change-log row, so a
    // redelivery after a crash (the reader re-processing the same row) re-derives the
    // same token — the at-most-once guarantee for a money/mail after:capture side
    // effect. The token keys on the row image (entity + event kind + old/new), never
    // the processing timestamp or the event id.
    #[test]
    fn capture_dispatch_token_is_stable_per_change_log_row() {
        let hooks = capture_hooks("reconcile", "after:capture:Order:insert");

        // Derive the dispatch token for a captured row via the real capture path.
        // Each call mints a fresh event id + timestamp (as a redelivery would), so an
        // equal token proves the derivation ignores them.
        let token = |data: serde_json::Value| -> String {
            let event = captured_insert("Order", data, Some(CAPTURED_WRITE_MARKER));
            let (fn_event, cdc) = observer_event_to_capture(&event).expect("capture event");
            let plans = plan_after_capture_dispatch(&hooks, &fn_event, cdc.as_deref());
            let plan = plans.first().expect("one capture plan");
            dispatch_idempotency_token(
                None,
                DispatchSource::AfterCapture,
                "reconcile",
                &plan.payload,
            )
        };

        let row = json!({ "id": "o-1", "status": "approved" });
        let first = token(row.clone());
        let redelivery = token(row); // same row image, fresh event id + timestamp
        assert_eq!(
            first, redelivery,
            "redelivery of the same captured row re-derives the same token (at-most-once)"
        );

        // A genuinely different row image derives a different token.
        let other = token(json!({ "id": "o-2", "status": "approved" }));
        assert_ne!(first, other, "a different captured row derives a different token");
    }
}