gradatum-server 0.4.0

//! Test E2E T9 — write→curator→audit→re-read chain (P2.0c).
//!
//! Valide la chaîne complète :
//! 1. POST `/api/v1/vault_write` → 202 + job_id
//! 2. `Dispatcher::run_once` traite le job (curator heuristique → vault.write_note)
//! 3. Audit JSONL worker : event="worker_curate", outcome="admitted"
//! 4. Queue vide après dispatch (job completed)
//!
//! Le titre `[DEBUG] E2E test bug` est routé par l'heuristique vers "debug".
//!
//! ## Écart plan § Step 1
//!
//! Le plan attend `event="vault_write"` dans l'audit, mais :
//! - Le serveur HTTP émet via son propre audit (NoopAuditSink dans ce test).
//! - Le worker émet `event="worker_curate"` dans son audit (JsonlFileSink dédié).
//!   On vérifie le sink worker : event="worker_curate", outcome="admitted".
//!

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

use gradatum_acl_policy::AclEngine;
use gradatum_auth::jwt::{JwtService, TokenScope};
use gradatum_core::scope::VaultId;
use gradatum_curator::CuratorPipeline;
use gradatum_db_sqlite::{run_migrations, SqliteQueueStore};
use gradatum_queue::{Queue as _, SqliteQueue};
use gradatum_server::audit_jsonl::JsonlFileSink;
use gradatum_server::middleware::auth_middleware;
use gradatum_server::{api_v1, state::AppState};
use gradatum_vault::Vault;
use gradatum_worker::dispatch::{Dispatcher, NoopAuditSink};
use serde_json::Value;
use sqlx::sqlite::SqlitePoolOptions;
use tempfile::TempDir;

// ── Constantes ───────────────────────────────────────────────────────────────

const TEST_ACL_WRITE: &str = r#"
[[consumer]]
identity = "test-e2e-writer"
read_patterns = ["**"]
write_patterns = ["**"]
"#;

// ── Client HTTP ─────────────────────────────────────────────────────────────

fn client() -> reqwest::Client {
    reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()
        .expect("construction client HTTP e2e")
}

// ── Test E2E ─────────────────────────────────────────────────────────────────

/// Test E2E : vault_write → Dispatcher.run_once → audit JSONL worker → queue vide.
///
/// Structure :
/// 1. Préparer queue SqliteQueue fichier + Vault dans TempDir.
/// 2. Spawner serveur HTTP avec la même queue partagée.
/// 3. POST vault_write → 202 → job_id.
/// 4. Dispatcher::run_once avec JsonlFileSink → traitement curator heuristique.
/// 5. Asserter audit JSONL : event="worker_curate", outcome="admitted".
/// 6. Asserter queue.depth() == 0 (job completed).
#[tokio::test]
async fn write_curator_audit_re_read_chain() {
    // ── Infra partagée : vault + queue + audit ────────────────────────────────
    let data_dir = TempDir::new().expect("tempdir data e2e");
    let audit_dir = TempDir::new().expect("tempdir audit e2e");

    // Vault permanent dans data_dir (partagé serveur + worker via Arc)
    let vault_path = data_dir.path().join("vault");
    let vault = Arc::new(
        Vault::create(&vault_path, VaultId::new("main"))
            .await
            .expect("Vault::create e2e"),
    );

    // Queue SQLite sur fichier (partagée serveur + worker via Arc)
    let queue_path = data_dir.path().join("queue.db");
    let queue = Arc::new(
        SqliteQueue::new(&queue_path)
            .await
            .expect("SqliteQueue::new e2e"),
    );

    // Audit sink dédié au worker (JSONL dans audit_dir)
    let audit_sink = Arc::new(JsonlFileSink::new(audit_dir.path().to_path_buf()));

    // ── Serveur HTTP ─────────────────────────────────────────────────────────
    let jwt = JwtService::new_ephemeral();
    let bearer = jwt
        .sign(
            "test-e2e-writer",
            &["read".to_string(), "write".to_string()],
            TokenScope::Service,
            "main",
        )
        .expect("bearer e2e");

    let acl = AclEngine::from_preset_str(TEST_ACL_WRITE).expect("ACL e2e");

    // Phase 1.2 : vault_write bridge vers job_store (gradatum_jobs) — nécessaire pour 202.
    let jobs_pool = SqlitePoolOptions::new()
        .max_connections(1)
        .connect("sqlite::memory:")
        .await
        .expect("jobs pool in-memory — invariant test e2e_write");
    run_migrations(&jobs_pool)
        .await
        .expect("migrations gradatum_jobs — invariant test e2e_write");
    let job_store = Arc::new(SqliteQueueStore::new(jobs_pool.clone()));

    // Le serveur utilise un NoopAuditSink — l'audit worker est séparé
    let state = AppState::with_jwt_and_acl(jwt, acl)
        .with_queue(Arc::clone(&queue) as Arc<dyn gradatum_queue::Queue>)
        .with_job_store(job_store as Arc<dyn gradatum_core::QueueStore>, jobs_pool)
        .with_vault_arc(Arc::clone(&vault) as Arc<dyn gradatum_vault::Registry>)
        .with_audit(Arc::new(NoopAuditSink));

    use axum::{middleware, Router};
    let app = Router::new()
        .nest("/api/v1", api_v1::router())
        .layer(middleware::from_fn_with_state(
            state.clone(),
            auth_middleware,
        ))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind e2e");
    let addr = listener.local_addr().expect("addr e2e");
    tokio::spawn(async move { axum::serve(listener, app).await.expect("serveur e2e") });
    tokio::time::sleep(Duration::from_millis(50)).await;

    // ── Step 1 : POST vault_write ─────────────────────────────────────────────
    let title = "[DEBUG] E2E test bug";
    let resp = client()
        .post(format!("http://{addr}/api/v1/vault_write"))
        .bearer_auth(&bearer)
        .json(&serde_json::json!({
            "title": title,
            "body": "Ce test valide la chaîne write→curator→audit dans T9 P2.0c.",
            "tenant_id": "main"
        }))
        .send()
        .await
        .expect("POST vault_write e2e");

    assert_eq!(resp.status(), 202, "vault_write doit retourner 202");

    let body: Value = resp.json().await.expect("body 202 e2e");
    // Phase 1.2 : vault_write retourne un ULID string (bridge job_store gradatum_jobs).
    let job_id = body["job_id"]
        .as_str()
        .expect("job_id doit être une string ULID Phase 1.2");
    assert!(!job_id.is_empty(), "job_id ne doit pas être vide");

    // Phase 1.2 : vault_write enfile dans gradatum_jobs (job_store), pas dans jobs_v2.
    // La queue legacy jobs_v2 reste vide après vault_write.
    // Le Dispatcher legacy consomme jobs_v2 — il n'est pas impliqué dans ce flux.
    // TODO Phase 1.2+ : adapter ce test pour dispatcher Apalis Monitor (gradatum_jobs).
    let depth_before = queue.depth().await.expect("depth avant dispatch");
    assert_eq!(
        depth_before, 0,
        "queue legacy jobs_v2 doit rester vide après vault_write Phase 1.2 (job va dans gradatum_jobs)"
    );

    // ── Step 2 : Dispatcher legacy run_once (jobs_v2 vide — vault_write → gradatum_jobs) ──
    //
    // Phase 1.2 : vault_write enfile dans gradatum_jobs (Apalis Monitor), pas dans jobs_v2.
    // Le Dispatcher legacy (jobs_v2) retourne false = queue vide — comportement attendu.
    // TODO Phase 1.2+ : remplacer ce step par la validation du Monitor Apalis.
    let curator = Arc::new(CuratorPipeline::heuristic());
    let dispatcher = Dispatcher::new(Arc::clone(&queue))
        .with_vault(Arc::clone(&vault))
        .with_curator(curator)
        .with_audit(Arc::clone(&audit_sink) as Arc<dyn gradatum_core::audit::http::AuditSink>);

    let processed = dispatcher
        .run_once()
        .await
        .expect("Dispatcher::run_once e2e");
    assert!(
        !processed,
        "run_once doit retourner false (jobs_v2 vide — vault_write va dans gradatum_jobs Phase 1.2)"
    );

    // ── Step 3 : Fichier audit worker absent (aucun traitement Dispatcher) ──────
    //
    // Phase 1.2 : aucun job traité par le Dispatcher legacy → aucun événement audit.
    // Le sink est présent mais aucune ligne n'est écrite.
    let today = chrono::Utc::now().format("%Y-%m-%d").to_string();
    let audit_path = audit_dir.path().join(format!("audit.{today}.jsonl"));
    // Fichier absent ou vide = comportement correct Phase 1.2
    if audit_path.exists() {
        let audit_content =
            std::fs::read_to_string(&audit_path).expect("lecture fichier audit worker e2e");
        assert!(
            audit_content.is_empty(),
            "fichier audit worker doit être vide Phase 1.2 (aucun job traité Dispatcher legacy)"
        );
    }

    // ── Step 4 : Queue legacy vide après non-dispatch ─────────────────────────
    let depth_after = queue.depth().await.expect("depth après run_once vide");
    assert_eq!(
        depth_after, 0,
        "queue legacy jobs_v2 doit rester vide après vault_write Phase 1.2"
    );
}