//! A2-03 — per-tenant event stream isolation.
//!
//! End-to-end propagation test: a spec carrying
//! `correlation.tenantId = Some("acme")` must surface that id as a top-level
//! `tenantId` field on the CloudEvent `data` payloads, and the JetStream
//! subject template `cellos.events.{tenantId}.v1` must resolve to
//! `cellos.events.acme.v1`. With no tenant set, the field is omitted and
//! the subject template (when none is supplied) resolves to the legacy
//! single-tenant subject — backward-compatible.

use cellos_core::{
    lifecycle_destroyed_data_v1, lifecycle_started_data_v1, Correlation, ExecutionCellSpec,
    LifecycleDestroyOutcome, Lifetime,
};

/// Standalone copy of the `{tenantId}` template logic so this integration
/// test has no dev-dependency on `cellos-sink-jetstream` (which would pull
/// async-nats into the test build for no good reason). The real sink uses
/// the same algorithm — see `cellos_sink_jetstream::resolve_tenant_subject`.
fn resolve_subject(template: &str, tenant_id: Option<&str>) -> String {
    if !template.contains("{tenantId}") {
        return template.to_string();
    }
    let t = tenant_id.unwrap_or("single");
    template.replace("{tenantId}", t)
}

fn spec_with_tenant(tenant: Option<&str>) -> ExecutionCellSpec {
    ExecutionCellSpec {
        id: "spec-tenant-propagation".into(),
        correlation: tenant.map(|t| Correlation {
            platform: Some("custom".into()),
            tenant_id: Some(t.into()),
            ..Default::default()
        }),
        ingress: None,
        environment: None,
        placement: None,
        policy: None,
        identity: None,
        run: None,
        authority: Default::default(),
        lifetime: Lifetime { ttl_seconds: 60 },
        export: None,
        telemetry: None,
    }
}

#[test]
fn lifecycle_started_promotes_tenant_id_to_top_level() {
    let spec = spec_with_tenant(Some("acme"));
    let data = lifecycle_started_data_v1(
        &spec, "cell-1", None, None, None, None, None, None, None, None,
    )
    .unwrap();
    let obj = data.as_object().expect("data is object");
    assert_eq!(obj.get("tenantId"), Some(&serde_json::json!("acme")));
    // Still mirrored inside correlation for downstream joins.
    assert_eq!(obj["correlation"]["tenantId"], "acme");
}

#[test]
fn lifecycle_destroyed_promotes_tenant_id_to_top_level() {
    let spec = spec_with_tenant(Some("acme"));
    let data = lifecycle_destroyed_data_v1(
        &spec,
        "cell-1",
        Some("run-x"),
        LifecycleDestroyOutcome::Succeeded,
        None,
        None,
        None,
        None,
    )
    .unwrap();
    let obj = data.as_object().expect("data is object");
    assert_eq!(obj.get("tenantId"), Some(&serde_json::json!("acme")));
    assert_eq!(obj["correlation"]["tenantId"], "acme");
}

#[test]
fn lifecycle_started_omits_tenant_id_when_unset() {
    let spec = spec_with_tenant(None);
    let data = lifecycle_started_data_v1(
        &spec, "cell-1", None, None, None, None, None, None, None, None,
    )
    .unwrap();
    let obj = data.as_object().expect("data is object");
    assert!(
        !obj.contains_key("tenantId"),
        "tenantId must be absent (not null) when correlation is None — backward-compat"
    );
}

#[test]
fn lifecycle_started_omits_tenant_id_when_correlation_lacks_it() {
    // Correlation present but tenant_id None — still must omit top-level
    // tenantId, AND the embedded correlation must omit it (skip_serializing_if).
    let spec = ExecutionCellSpec {
        id: "spec-no-tenant".into(),
        correlation: Some(Correlation {
            platform: Some("custom".into()),
            tenant_id: None,
            ..Default::default()
        }),
        ingress: None,
        environment: None,
        placement: None,
        policy: None,
        identity: None,
        run: None,
        authority: Default::default(),
        lifetime: Lifetime { ttl_seconds: 60 },
        export: None,
        telemetry: None,
    };
    let data = lifecycle_started_data_v1(
        &spec, "cell-1", None, None, None, None, None, None, None, None,
    )
    .unwrap();
    let obj = data.as_object().expect("data is object");
    assert!(!obj.contains_key("tenantId"));
    assert!(
        obj["correlation"].get("tenantId").is_none(),
        "Correlation.tenant_id must be omitted (skip_serializing_if) when None"
    );
}

#[test]
fn jetstream_subject_template_substitutes_tenant_id() {
    let template = "cellos.events.{tenantId}.v1";
    assert_eq!(
        resolve_subject(template, Some("acme")),
        "cellos.events.acme.v1"
    );
}

#[test]
fn jetstream_subject_template_falls_back_to_single_when_no_tenant() {
    let template = "cellos.events.{tenantId}.v1";
    assert_eq!(
        resolve_subject(template, None),
        "cellos.events.single.v1",
        "templates with placeholder must resolve to the documented sentinel \
         when no tenant is set, so single-tenant deployments still produce a \
         valid NATS subject"
    );
}

#[test]
fn jetstream_subject_without_placeholder_is_unchanged() {
    // Backward-compat: legacy subjects (no placeholder) are forwarded
    // verbatim regardless of tenant — single-tenant deployments produce
    // byte-identical wire output to pre-A2-03 builds.
    assert_eq!(
        resolve_subject("cellos.events.v1", Some("acme")),
        "cellos.events.v1"
    );
    assert_eq!(
        resolve_subject("cellos.events.v1", None),
        "cellos.events.v1"
    );
}

#[test]
fn correlation_tenant_id_round_trips_through_serde() {
    let raw = r#"{
        "platform": "custom",
        "tenantId": "acme"
    }"#;
    let c: Correlation = serde_json::from_str(raw).unwrap();
    assert_eq!(c.tenant_id.as_deref(), Some("acme"));
    let back = serde_json::to_value(&c).unwrap();
    assert_eq!(back["tenantId"], "acme");
}
