//! γ port-and-adapter boundary tests for [`pas_external::AuditSink`]
//! and the [`pas_external::RateLimitedAuditSink`] composition adapter
//! (Phase 9.C — M48 + M49).
//!
//! Mirror of `bearer_verifier_boundary.rs` shape: tests live at the
//! port boundary, not at internal mechanics. Token-bucket math is
//! unit-tested in `audit::rate_limit::tests`; AuditEvent serde lives
//! in `audit::sink::tests`. This file proves the *port contracts*
//! that downstream consumers (PCS chat-auth, future RCW/CTW
//! middleware) rely on:
//!
//! 1. **Sink contracts** — Noop drops, Memory records in order with
//!    full hint fidelity.
//! 2. **Composition contracts** — `RateLimitedAuditSink` admits under
//!    quota, drops over quota, isolates per source key.
//! 3. **Object-safety + dyn dispatch** — every adapter is usable
//!    behind `Arc<dyn AuditSink>`, the runtime shape that
//!    `PasJwtVerifier::with_audit` consumes (Phase 9.D).
//! 4. **Ops observability** — `dropped_total` accessor reports exactly
//!    the drops, regardless of admit count.

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

use pas_external::{
    AuditEvent, AuditSink, MemoryAuditSink, MemoryRateLimiter, NoopAuditSink,
    RateLimitedAuditSink, RateLimiter, VerifyErrorKind,
};
use time::OffsetDateTime;

fn fixture(kind: VerifyErrorKind, source_id: &str) -> AuditEvent {
    AuditEvent {
        kind,
        occurred_at: OffsetDateTime::UNIX_EPOCH,
        source_id: source_id.to_owned(),
        client_id_hint: None,
        kid_hint: None,
        metadata: BTreeMap::new(),
    }
}

#[tokio::test]
async fn noop_sink_drops_all_events_silently() {
    let sink = NoopAuditSink;
    sink.record_failure(fixture(VerifyErrorKind::Expired, "x"))
        .await;
    sink.record_failure(fixture(VerifyErrorKind::SignatureInvalid, "y"))
        .await;
    // Contract: no panic, no observable effect. The point of Noop is
    // explicit-but-silent — the consumer chose this by not calling
    // `with_audit(...)` on PasJwtVerifier.
}

#[tokio::test]
async fn memory_sink_records_with_full_hint_fidelity() {
    let sink = MemoryAuditSink::new();
    let event = AuditEvent::from_hints(
        VerifyErrorKind::SignatureInvalid,
        OffsetDateTime::UNIX_EPOCH,
        Some("rcw-client".to_owned()),
        Some("k1".to_owned()),
        BTreeMap::new(),
    );

    sink.record_failure(event).await;

    let recorded = sink.events();
    assert_eq!(recorded.len(), 1);
    let e = &recorded[0];
    assert_eq!(e.kind, VerifyErrorKind::SignatureInvalid);
    assert_eq!(e.source_id, "rcw-client::k1");
    assert_eq!(e.client_id_hint.as_deref(), Some("rcw-client"));
    assert_eq!(e.kid_hint.as_deref(), Some("k1"));
}

#[tokio::test]
async fn rate_limited_sink_admits_under_quota() {
    let memory = Arc::new(MemoryAuditSink::new());
    let limiter: Arc<dyn RateLimiter> =
        Arc::new(MemoryRateLimiter::new(3, Duration::from_secs(60)));
    let limited =
        RateLimitedAuditSink::new(memory.clone() as Arc<dyn AuditSink>, limiter);

    for _ in 0..3 {
        limited
            .record_failure(fixture(VerifyErrorKind::Expired, "rcw::k1"))
            .await;
    }
    assert_eq!(memory.len(), 3);
    assert_eq!(limited.dropped_total(), 0);
}

#[tokio::test]
async fn rate_limited_sink_drops_over_quota_silently() {
    let memory = Arc::new(MemoryAuditSink::new());
    let limiter: Arc<dyn RateLimiter> =
        Arc::new(MemoryRateLimiter::new(2, Duration::from_secs(60)));
    let limited =
        RateLimitedAuditSink::new(memory.clone() as Arc<dyn AuditSink>, limiter);

    for _ in 0..5 {
        limited
            .record_failure(fixture(VerifyErrorKind::SignatureInvalid, "rcw::k1"))
            .await;
    }

    assert_eq!(memory.len(), 2, "only first two admitted");
    assert_eq!(limited.dropped_total(), 3, "remaining three dropped");
}

#[tokio::test]
async fn rate_limited_sink_isolates_per_compound_source_key() {
    // Phase 9 design call (e): compound `client_id_hint ‖ kid_hint`
    // means (rcw, k1) and (ctw, k1) get separate quotas even though
    // the kid is identical. Regression here would silently bucket
    // all sources together, defeating per-source attribution.
    let memory = Arc::new(MemoryAuditSink::new());
    let limiter: Arc<dyn RateLimiter> =
        Arc::new(MemoryRateLimiter::new(1, Duration::from_secs(60)));
    let limited =
        RateLimitedAuditSink::new(memory.clone() as Arc<dyn AuditSink>, limiter);

    // Three distinct compound keys; each admits once.
    limited
        .record_failure(fixture(VerifyErrorKind::Expired, "rcw::k1"))
        .await;
    limited
        .record_failure(fixture(VerifyErrorKind::Expired, "ctw::k1"))
        .await;
    limited
        .record_failure(fixture(VerifyErrorKind::Expired, "ctw::k2"))
        .await;
    assert_eq!(memory.len(), 3);
    assert_eq!(limited.dropped_total(), 0);

    // Each key now exhausted independently.
    limited
        .record_failure(fixture(VerifyErrorKind::Expired, "rcw::k1"))
        .await;
    limited
        .record_failure(fixture(VerifyErrorKind::Expired, "ctw::k1"))
        .await;
    assert_eq!(memory.len(), 3, "no admits past per-key quota");
    assert_eq!(limited.dropped_total(), 2);
}

#[tokio::test]
async fn rate_limited_sink_works_behind_dyn_audit_sink() {
    // Runtime shape consumer middleware injects: composed adapter
    // hidden behind `Arc<dyn AuditSink>`. If this stops compiling,
    // the verifier wiring (Phase 9.D) breaks.
    let memory = Arc::new(MemoryAuditSink::new());
    let limited: Arc<dyn AuditSink> = Arc::new(RateLimitedAuditSink::new(
        memory.clone() as Arc<dyn AuditSink>,
        Arc::new(MemoryRateLimiter::default()),
    ));
    limited
        .record_failure(fixture(VerifyErrorKind::Expired, "anon::nokid"))
        .await;
    assert_eq!(memory.len(), 1);
}

/// Compile-time guard: every shipped adapter is `Arc<dyn AuditSink>`-shaped.
#[allow(dead_code)]
fn dyn_object_safety() {
    let _: Arc<dyn AuditSink> = Arc::new(NoopAuditSink);
    let _: Arc<dyn AuditSink> = Arc::new(MemoryAuditSink::new());
    let _: Arc<dyn AuditSink> = Arc::new(RateLimitedAuditSink::new(
        Arc::new(NoopAuditSink),
        Arc::new(MemoryRateLimiter::default()),
    ));
    let _: Arc<dyn RateLimiter> = Arc::new(MemoryRateLimiter::default());
}
