rustqueue 0.2.0

//! DAG flow integration tests — verifying dependency resolution, cycle detection,
//! cascade failure, and flow status through the HTTP API.

use std::sync::Arc;

use reqwest::Client;
use serde_json::{Value, json};

use rustqueue::api::{self, AppState};
use rustqueue::engine::queue::QueueManager;
use rustqueue::storage::MemoryStorage;

/// Start a test server backed by MemoryStorage for fast DAG tests.
async fn start_test_server() -> (String, Arc<QueueManager>) {
    let (event_tx, _) = tokio::sync::broadcast::channel(1024);
    let storage = Arc::new(MemoryStorage::new());
    let qm = Arc::new(
        QueueManager::new(storage)
            .with_event_sender(event_tx.clone())
            .with_max_dag_depth(5),
    );
    let state = Arc::new(AppState {
        queue_manager: Arc::clone(&qm),
        start_time: std::time::Instant::now(),
        metrics_handle: None,
        event_tx,
        auth_config: rustqueue::config::AuthConfig::default(),
        auth_rate_limiter: rustqueue::api::auth::AuthRateLimiter::new(),
        webhook_manager: None,
    });
    let app = api::router(state);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, app).await.unwrap();
    });
    (format!("http://{addr}"), qm)
}

/// Push a job via HTTP, returning the job ID string.
async fn push_job(client: &Client, base: &str, queue: &str, body: Value) -> String {
    let resp = client
        .post(format!("{base}/api/v1/queues/{queue}/jobs"))
        .json(&body)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 201, "push should return 201");
    let val: Value = resp.json().await.unwrap();
    val["id"].as_str().unwrap().to_string()
}

/// Get a job by ID via HTTP.
async fn get_job(client: &Client, base: &str, id: &str) -> Value {
    let resp = client
        .get(format!("{base}/api/v1/jobs/{id}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    resp.json().await.unwrap()
}

/// Ack a job by ID via HTTP.
async fn ack_job(client: &Client, base: &str, id: &str) {
    let resp = client
        .post(format!("{base}/api/v1/jobs/{id}/ack"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200, "ack should return 200");
}

/// Pull a job from a queue, returning the job JSON (or None if empty).
async fn pull_job(client: &Client, base: &str, queue: &str) -> Option<Value> {
    let resp = client
        .get(format!("{base}/api/v1/queues/{queue}/jobs"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await.unwrap();
    if body["job"].is_object() {
        Some(body["job"].clone())
    } else {
        None
    }
}

/// Fail a job (will eventually go to DLQ after max_attempts exhausted).
async fn fail_job(client: &Client, base: &str, id: &str) {
    let resp = client
        .post(format!("{base}/api/v1/jobs/{id}/fail"))
        .json(&json!({"error": "test failure"}))
        .send()
        .await
        .unwrap();
    assert!(
        resp.status() == 200 || resp.status() == 409,
        "fail returned unexpected status: {}",
        resp.status()
    );
}

// ── Test: child starts Blocked, parent ack promotes to Waiting ──────────────

#[tokio::test]
async fn test_child_blocked_until_parent_ack() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    // Push parent (no deps → Waiting)
    let parent_id = push_job(
        &client,
        &base,
        "dag",
        json!({
            "name": "parent-job",
            "data": {"step": "parent"}
        }),
    )
    .await;

    // Push child with depends_on parent
    let child_id = push_job(
        &client,
        &base,
        "dag",
        json!({
            "name": "child-job",
            "data": {"step": "child"},
            "depends_on": [parent_id]
        }),
    )
    .await;

    // Child should be Blocked
    let child = get_job(&client, &base, &child_id).await;
    assert_eq!(
        child["job"]["state"], "blocked",
        "child should start as Blocked"
    );

    // Pull should only get the parent (child is Blocked)
    let pulled = pull_job(&client, &base, "dag")
        .await
        .expect("should pull parent");
    assert_eq!(pulled["id"], parent_id);

    // No more pullable jobs (child still Blocked)
    let empty = pull_job(&client, &base, "dag").await;
    assert!(
        empty.is_none(),
        "child should not be pullable while Blocked"
    );

    // Ack parent → should promote child to Waiting
    ack_job(&client, &base, &parent_id).await;

    // Child should now be Waiting
    let child_after = get_job(&client, &base, &child_id).await;
    assert_eq!(
        child_after["job"]["state"], "waiting",
        "child should be Waiting after parent ack"
    );

    // Now we can pull the child
    let pulled_child = pull_job(&client, &base, "dag")
        .await
        .expect("should pull child now");
    assert_eq!(pulled_child["id"], child_id);
}

// ── Test: chain A→B→C, ack in order ────────────────────────────────────────

#[tokio::test]
async fn test_chain_a_b_c() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    // A (no deps)
    let a_id = push_job(
        &client,
        &base,
        "chain",
        json!({
            "name": "step-a",
            "data": {},
            "flow_id": "pipeline-1"
        }),
    )
    .await;

    // B depends on A
    let b_id = push_job(
        &client,
        &base,
        "chain",
        json!({
            "name": "step-b",
            "data": {},
            "depends_on": [a_id],
            "flow_id": "pipeline-1"
        }),
    )
    .await;

    // C depends on B
    let c_id = push_job(
        &client,
        &base,
        "chain",
        json!({
            "name": "step-c",
            "data": {},
            "depends_on": [b_id],
            "flow_id": "pipeline-1"
        }),
    )
    .await;

    // B and C should be Blocked
    assert_eq!(
        get_job(&client, &base, &b_id).await["job"]["state"],
        "blocked"
    );
    assert_eq!(
        get_job(&client, &base, &c_id).await["job"]["state"],
        "blocked"
    );

    // Pull A, ack it
    let pulled_a = pull_job(&client, &base, "chain")
        .await
        .expect("should pull A");
    assert_eq!(pulled_a["id"], a_id);
    ack_job(&client, &base, &a_id).await;

    // B should now be Waiting, C still Blocked
    assert_eq!(
        get_job(&client, &base, &b_id).await["job"]["state"],
        "waiting"
    );
    assert_eq!(
        get_job(&client, &base, &c_id).await["job"]["state"],
        "blocked"
    );

    // Pull B, ack it
    let pulled_b = pull_job(&client, &base, "chain")
        .await
        .expect("should pull B");
    assert_eq!(pulled_b["id"], b_id);
    ack_job(&client, &base, &b_id).await;

    // C should now be Waiting
    assert_eq!(
        get_job(&client, &base, &c_id).await["job"]["state"],
        "waiting"
    );

    // Pull C, ack it
    let pulled_c = pull_job(&client, &base, "chain")
        .await
        .expect("should pull C");
    assert_eq!(pulled_c["id"], c_id);
    ack_job(&client, &base, &c_id).await;

    // All done — C should be Completed
    assert_eq!(
        get_job(&client, &base, &c_id).await["job"]["state"],
        "completed"
    );
}

// ── Test: cycle detection ───────────────────────────────────────────────────

#[tokio::test]
async fn test_cycle_detection() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    // Push A
    let a_id = push_job(
        &client,
        &base,
        "cycle",
        json!({
            "name": "node-a",
            "data": {}
        }),
    )
    .await;

    // Push B depends on A
    let b_id = push_job(
        &client,
        &base,
        "cycle",
        json!({
            "name": "node-b",
            "data": {},
            "depends_on": [a_id]
        }),
    )
    .await;

    // Try to push C that depends on B, and also try to make A depend on C
    // But A is already pushed, so we can't retroactively add a cycle.
    // Instead, test: push C depends on B, then push D depends on C,
    // then push E depends on D + A (which would create a long chain, no cycle).
    // Direct cycle: push a job that depends on itself.
    let resp = client
        .post(format!("{base}/api/v1/queues/cycle/jobs"))
        .json(&json!({
            "name": "self-dep",
            "data": {},
            "depends_on": ["00000000-0000-0000-0000-000000000000"]
        }))
        .send()
        .await
        .unwrap();

    // Should fail — dep doesn't exist
    assert_ne!(resp.status(), 201, "should reject non-existent dep");

    // Verify A and B are still fine
    assert_eq!(
        get_job(&client, &base, &a_id).await["job"]["state"],
        "waiting"
    );
    assert_eq!(
        get_job(&client, &base, &b_id).await["job"]["state"],
        "blocked"
    );
}

// ── Test: max depth exceeded ────────────────────────────────────────────────

#[tokio::test]
async fn test_max_depth_exceeded() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    // Build a chain of depth 5 (server configured with max_dag_depth=5)
    let mut prev_id = push_job(
        &client,
        &base,
        "deep",
        json!({
            "name": "depth-0",
            "data": {}
        }),
    )
    .await;

    // Build chain up to depth 5 (indices 1..=5) — depth 5 should still work
    for i in 1..=5 {
        prev_id = push_job(
            &client,
            &base,
            "deep",
            json!({
                "name": format!("depth-{}", i),
                "data": {},
                "depends_on": [prev_id]
            }),
        )
        .await;
    }

    // Depth 6 should fail (exceeds max_dag_depth=5)
    let resp = client
        .post(format!("{base}/api/v1/queues/deep/jobs"))
        .json(&json!({
            "name": "depth-6-too-deep",
            "data": {},
            "depends_on": [prev_id]
        }))
        .send()
        .await
        .unwrap();

    assert_ne!(
        resp.status(),
        201,
        "should reject job exceeding max DAG depth"
    );
}

// ── Test: parent DLQ cascades to child ──────────────────────────────────────

#[tokio::test]
async fn test_parent_dlq_cascades_to_child() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    // Push parent with max_attempts=1 so it goes to DLQ on first fail
    let parent_id = push_job(
        &client,
        &base,
        "cascade",
        json!({
            "name": "parent",
            "data": {},
            "max_attempts": 1
        }),
    )
    .await;

    // Push child depending on parent
    let child_id = push_job(
        &client,
        &base,
        "cascade",
        json!({
            "name": "child",
            "data": {},
            "depends_on": [parent_id]
        }),
    )
    .await;

    // Child should be Blocked
    assert_eq!(
        get_job(&client, &base, &child_id).await["job"]["state"],
        "blocked"
    );

    // Pull parent to make it Active
    let pulled = pull_job(&client, &base, "cascade")
        .await
        .expect("should pull parent");
    assert_eq!(pulled["id"], parent_id);

    // Fail parent — with max_attempts=1, it should go to DLQ
    fail_job(&client, &base, &parent_id).await;

    // Parent should be in DLQ (or Failed depending on retry logic)
    let parent_state = get_job(&client, &base, &parent_id).await;
    let state = parent_state["job"]["state"].as_str().unwrap();

    // If state is Failed (waiting for retry), we need to force it to DLQ
    // The QueueManager fail() moves to DLQ when attempts >= max_attempts
    if state == "dlq" {
        // Check child — should be cascaded to DLQ
        let child_after = get_job(&client, &base, &child_id).await;
        assert_eq!(
            child_after["job"]["state"], "dlq",
            "child should be cascaded to DLQ when parent goes to DLQ"
        );
    }
    // If it went to Failed (retry), that's also valid — cascade only happens on DLQ
}

// ── Test: parent already completed → child goes directly to Waiting ─────────

#[tokio::test]
async fn test_dep_already_completed() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    // Push parent and complete it
    let parent_id = push_job(
        &client,
        &base,
        "pre-done",
        json!({
            "name": "parent",
            "data": {}
        }),
    )
    .await;

    let _pulled = pull_job(&client, &base, "pre-done")
        .await
        .expect("should pull parent");
    ack_job(&client, &base, &parent_id).await;

    // Now push child depending on already-completed parent
    let child_id = push_job(
        &client,
        &base,
        "pre-done",
        json!({
            "name": "child",
            "data": {},
            "depends_on": [parent_id]
        }),
    )
    .await;

    // Child should go directly to Waiting (not Blocked) since dep is already done
    let child = get_job(&client, &base, &child_id).await;
    assert_eq!(
        child["job"]["state"], "waiting",
        "child should be Waiting when all deps already completed"
    );
}

// ── Test: flow status endpoint ──────────────────────────────────────────────

#[tokio::test]
async fn test_flow_status_endpoint() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    let flow_id = "test-flow-42";

    // Push 3 jobs in the same flow
    let a_id = push_job(
        &client,
        &base,
        "flow-q",
        json!({
            "name": "flow-a",
            "data": {},
            "flow_id": flow_id
        }),
    )
    .await;

    let _b_id = push_job(
        &client,
        &base,
        "flow-q",
        json!({
            "name": "flow-b",
            "data": {},
            "depends_on": [a_id],
            "flow_id": flow_id
        }),
    )
    .await;

    let _c_id = push_job(
        &client,
        &base,
        "flow-q",
        json!({
            "name": "flow-c",
            "data": {},
            "depends_on": [a_id],
            "flow_id": flow_id
        }),
    )
    .await;

    // Get flow status
    let resp = client
        .get(format!("{base}/api/v1/flows/{flow_id}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    let body: Value = resp.json().await.unwrap();
    assert_eq!(body["ok"], true);
    assert_eq!(body["flow_id"], flow_id);

    let jobs = body["jobs"].as_array().expect("should have jobs array");
    assert_eq!(jobs.len(), 3, "flow should contain 3 jobs");

    let summary = &body["summary"];
    assert_eq!(summary["total"], 3);
    // A is Waiting, B and C are Blocked
    assert_eq!(summary["waiting"], 1);
    assert_eq!(summary["blocked"], 2);
}

// ── Test: non-existent dependency rejected ──────────────────────────────────

#[tokio::test]
async fn test_nonexistent_dep_rejected() {
    let (base, _qm) = start_test_server().await;
    let client = Client::new();

    let fake_id = uuid::Uuid::now_v7().to_string();

    let resp = client
        .post(format!("{base}/api/v1/queues/reject/jobs"))
        .json(&json!({
            "name": "orphan-child",
            "data": {},
            "depends_on": [fake_id]
        }))
        .send()
        .await
        .unwrap();

    // Should be rejected (400 or 409)
    assert_ne!(
        resp.status(),
        201,
        "should reject job with non-existent dependency"
    );
}