#[path = "support/utils.rs"]
mod _utils;

use _utils::{boxed_error, cleanup_run, create_client, new_run_id, print_summary, require};
use mubit_sdk::TransportMode;
use serde_json::{json, Value};
use std::error::Error;
use std::time::Instant;
use tokio::time::{sleep, Duration};

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn Error>> {
    let name = "control_ingest_job_lifecycle";
    let started = Instant::now();
    let client = create_client().await?;
    let run_id = new_run_id("control_ingest_job_lifecycle");
    client.set_run_id(Some(run_id.clone()));
    client.set_transport(TransportMode::Http);

    let mut passed = true;
    let mut detail = "validated control ingest job lifecycle".to_string();
    let mut metrics = json!({});

    let scenario = async {
        let accepted = client
            .control
            .ingest(json!({
                "run_id": run_id,
                "agent_id": "rust-sdk-example",
                "idempotency_key": format!("ingest-job-{}", run_id),
                "parallel": false,
                "items": [
                    {
                        "item_id": "incident-latency-1",
                        "content_type": "text",
                        "text": "Incident SEV2-7314 at 2026-02-14T03:22:00Z: checkout API latency in us-east rose from 180ms to 2.4s after cache warmer v1.8.1 rollout; rollback restored baseline.",
                        "payload_json": "",
                        "hints_json": "{\"priority\":\"high\",\"signal\":\"performance\"}",
                        "metadata_json": "{\"source\":\"ops-incident\",\"region\":\"us-east\",\"service\":\"checkout-api\"}",
                    },
                    {
                        "item_id": "customer-advisory-2",
                        "content_type": "text",
                        "text": "Customer advisory draft: EU enterprise tenant requested SLA clarification for batch exports; legal referenced DPA section 4.2 and retention requirement of 30 days.",
                        "payload_json": "",
                        "hints_json": "{\"priority\":\"medium\",\"signal\":\"policy\"}",
                        "metadata_json": "{\"source\":\"customer-success\",\"region\":\"eu-central\",\"team\":\"legal-ops\"}",
                    },
                    {
                        "item_id": "cold-chain-telemetry-3",
                        "content_type": "text",
                        "text": "Warehouse telemetry snapshot: cold-chain lane C7 drifted from 2.1C to 6.4C for 14 minutes; sensor recalibrated and shipment placed on QA hold.",
                        "payload_json": "",
                        "hints_json": "{\"priority\":\"high\",\"signal\":\"logistics\"}",
                        "metadata_json": "{\"source\":\"iot-telemetry\",\"facility\":\"atl-03\",\"lane\":\"C7\"}",
                    }
                ],
            }))
            .await?;

        require(
            accepted
                .get("accepted")
                .and_then(Value::as_bool)
                .unwrap_or(false),
            format!("ingest not accepted: {accepted}"),
        )?;

        let job_id = accepted
            .get("job_id")
            .or_else(|| accepted.get("jobId"))
            .and_then(Value::as_str)
            .ok_or_else(|| boxed_error(format!("ingest did not return job_id: {accepted}")))?
            .to_string();

        let mut job: Option<Value> = None;
        for _ in 0..60 {
            let current = client
                .control
                .get_ingest_job(json!({ "run_id": run_id, "job_id": job_id }))
                .await?;
            if current
                .get("done")
                .and_then(Value::as_bool)
                .unwrap_or(false)
            {
                job = Some(current);
                break;
            }
            sleep(Duration::from_millis(250)).await;
        }

        let job = job.ok_or_else(|| boxed_error("timeout waiting for ingest job completion"))?;

        require(
            job.get("done").and_then(Value::as_bool).unwrap_or(false),
            format!("job should be done: {job}"),
        )?;
        require(
            job.get("status").and_then(Value::as_str) == Some("completed"),
            format!("job should be completed: {job}"),
        )?;

        let traces = job
            .get("traces")
            .and_then(Value::as_array)
            .ok_or_else(|| boxed_error(format!("job traces missing: {job}")))?;
        require(!traces.is_empty(), format!("job traces missing: {job}"))?;

        let mut write_count = 0usize;
        let mut successful_writes = 0usize;
        for trace in traces {
            let writes = trace
                .get("writes")
                .and_then(Value::as_array)
                .cloned()
                .unwrap_or_default();
            write_count += writes.len();
            successful_writes += writes
                .iter()
                .filter(|write| {
                    write
                        .get("success")
                        .and_then(Value::as_bool)
                        .unwrap_or(false)
                })
                .count();
        }

        require(
            write_count > 0,
            format!("expected non-empty ingestion writes: {job}"),
        )?;
        require(
            successful_writes > 0,
            format!("expected at least one successful write: {job}"),
        )?;

        metrics = json!({
            "run_id": run_id,
            "job_id": job_id,
            "trace_count": traces.len(),
            "write_count": write_count,
            "successful_writes": successful_writes,
        });

        Ok::<(), Box<dyn Error>>(())
    }
    .await;

    if let Err(err) = scenario {
        passed = false;
        detail = err.to_string();
    }

    let cleanup_ok = cleanup_run(&client, &run_id).await;
    if !cleanup_ok {
        passed = false;
        detail = format!("{detail} | cleanup failures");
    }

    print_summary(
        name,
        passed,
        &detail,
        &metrics,
        started.elapsed().as_secs_f64(),
        cleanup_ok,
    );

    if passed {
        Ok(())
    } else {
        Err(boxed_error(detail))
    }
}
