ppoppo-token 0.49.0

//! M37 propagation observability — engine layer (Phase 5 commit 5.3).
//!
//! Verifies that every port call site emits the canonical
//! `revocation.checked` tracing event so ops dashboards can measure
//! the ≤5min SLA. The substrate-side propagation mechanism (KVRocks
//! TTL + push paths) lives outside ppoppo-token; the engine's
//! responsibility is to emit the data needed to MEASURE it.
//!
//! Field contract (locked here, dashboards key off):
//!
//! - `target  = "ppoppo_token::revocation"` — RUST_LOG filter selector
//! - `message = "revocation.checked"`       — captured into the `message`
//!   field by `tracing`'s `Display`-formatted final string arg
//! - `port    = "replay" | "epoch"` — which axis fired
//! - `outcome = "admit" | "reject" | "transient"`
//! - `reason  = "replayed" | "stale"` (only on reject)
//! - `sub`    — subject ULID for per-account correlation
//!
//! Drift between this contract and the engine emit sites would silently
//! break operator dashboards — these tests are the regression guard.

#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

mod common;

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

use common::{CapturedEvent, EventCapture, MemoryReplayDefense};
use ppoppo_token::SigningKey;
use ppoppo_token::access_token::EpochEnforcement;
use ppoppo_token::access_token::{IssueConfig, IssueRequest, VerifyConfig, issue, verify};
use tracing_subscriber::Registry;
use tracing_subscriber::layer::SubscriberExt;

const EPOCH_OFF: EpochEnforcement = EpochEnforcement::Unenforced {
    reason: "test: epoch axis not under test",
};

const TEST_SUB: &str = "01HSAB00000000000000000000";
const TEST_CLIENT_ID: &str = "ppoppo-internal";
const TTL_15M: Duration = Duration::from_secs(900);
const ISSUER: &str = "https://accounts.ppoppo.com";
const AUDIENCE: &str = "ppoppo";

fn mint_token(signer: &SigningKey) -> String {
    let issue_cfg = IssueConfig::access_token(ISSUER, AUDIENCE, signer.kid());
    let req = IssueRequest::new(TEST_SUB, TEST_CLIENT_ID, TTL_15M);
    issue(&req, &issue_cfg, signer, jiff::Timestamp::now().as_second())
        .expect("issue should succeed")
}

/// Find the single revocation event matching `port` + `outcome`.
/// Asserts uniqueness — multiple matches mean the engine fired the
/// event twice for one verify call (regression).
fn one_event(events: &[CapturedEvent], port: &str, outcome: &str) -> CapturedEvent {
    let matches: Vec<_> = events
        .iter()
        .filter(|e| {
            e.fields.get("port").map(String::as_str) == Some(port)
                && e.fields.get("outcome").map(String::as_str) == Some(outcome)
        })
        .collect();
    assert_eq!(
        matches.len(),
        1,
        "expected exactly one event with port={port:?} outcome={outcome:?}, got {}: {:?}",
        matches.len(),
        events,
    );
    matches[0].clone()
}

// ── M35 replay port observability ────────────────────────────────────────

#[tokio::test]
async fn replay_admit_emits_revocation_checked_admit() {
    let capture = EventCapture::default();
    let subscriber = Registry::default().with(capture.clone());
    let _guard = tracing::subscriber::set_default(subscriber);

    let (signer, key_set) = SigningKey::test_pair();
    let token = mint_token(&signer);
    let port = Arc::new(MemoryReplayDefense::new());
    let cfg = VerifyConfig::access_token(ISSUER, AUDIENCE, EPOCH_OFF).with_replay_defense(port);
    verify(&token, &cfg, &key_set, jiff::Timestamp::now().as_second())
        .await
        .expect("verify");

    drop(_guard);

    let events = capture.revocation_events();
    let event = one_event(&events, "replay", "admit");
    assert_eq!(
        event.fields.get("message").map(String::as_str),
        Some("revocation.checked"),
        "message field carries the canonical event name",
    );
    assert_eq!(event.fields.get("sub").map(String::as_str), Some(TEST_SUB));
    assert_eq!(
        event.level, "TRACE",
        "admit fires at TRACE (high-frequency path)"
    );
}

#[tokio::test]
async fn replay_replayed_emits_reject_with_reason_replayed() {
    let capture = EventCapture::default();
    let subscriber = Registry::default().with(capture.clone());
    let _guard = tracing::subscriber::set_default(subscriber);

    let (signer, key_set) = SigningKey::test_pair();
    let token = mint_token(&signer);
    let port = Arc::new(MemoryReplayDefense::new());
    let cfg = VerifyConfig::access_token(ISSUER, AUDIENCE, EPOCH_OFF).with_replay_defense(port);
    verify(&token, &cfg, &key_set, jiff::Timestamp::now().as_second())
        .await
        .expect("first verify admits");
    let _ = verify(&token, &cfg, &key_set, jiff::Timestamp::now().as_second()).await; // second rejects

    drop(_guard);

    let events = capture.revocation_events();
    let event = one_event(&events, "replay", "reject");
    assert_eq!(
        event.fields.get("reason").map(String::as_str),
        Some("replayed"),
    );
    assert_eq!(
        event.level, "WARN",
        "reject fires at WARN (security-relevant)"
    );
}

#[tokio::test]
async fn replay_transient_emits_transient_outcome() {
    let capture = EventCapture::default();
    let subscriber = Registry::default().with(capture.clone());
    let _guard = tracing::subscriber::set_default(subscriber);

    let (signer, key_set) = SigningKey::test_pair();
    let token = mint_token(&signer);
    let port = Arc::new(MemoryReplayDefense::failing());
    let cfg = VerifyConfig::access_token(ISSUER, AUDIENCE, EPOCH_OFF).with_replay_defense(port);
    let _ = verify(&token, &cfg, &key_set, jiff::Timestamp::now().as_second()).await;

    drop(_guard);

    let event = one_event(&capture.revocation_events(), "replay", "transient");
    assert!(
        event.fields.contains_key("detail"),
        "transient event must carry adapter detail for ops triage",
    );
    assert_eq!(
        event.level, "WARN",
        "transient fires at WARN (ops-relevant)"
    );
}

// ── Short-circuit silence ────────────────────────────────────────────────

#[tokio::test]
async fn no_event_when_replay_port_is_none() {
    // SLA dashboards count substrate calls — short-circuited verifies
    // (no port wired) must NOT contribute. Otherwise propagation
    // latency averages get diluted by no-op verifies.
    let capture = EventCapture::default();
    let subscriber = Registry::default().with(capture.clone());
    let _guard = tracing::subscriber::set_default(subscriber);

    let (signer, key_set) = SigningKey::test_pair();
    let token = mint_token(&signer);
    let cfg = VerifyConfig::access_token(ISSUER, AUDIENCE, EPOCH_OFF);
    verify(&token, &cfg, &key_set, jiff::Timestamp::now().as_second())
        .await
        .expect("verify");

    drop(_guard);

    assert!(
        capture.revocation_events().is_empty(),
        "no port wired → no event (SLA dashboards must not count short-circuits)",
    );
}