#![allow(dead_code, unused_imports)]

use super::common::*;

use serde_json::Value;
use serial_test::serial;

const WORKSPACE_ID: &str = "ws_default";

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn portable_yaml_v2_is_redacted_and_semantic_replay_is_a_noop() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("operation-yaml-v2");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;
    let created = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload("http://127.0.0.1:9", "portable_v2"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let operation_id = created["operation_id"].as_str().unwrap();

    let yaml = client
        .get(format!("{base_url}/operations/{operation_id}/export"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    assert!(yaml.contains("format_version: '2'") || yaml.contains("format_version: 2"));
    let parsed: serde_yaml::Value = serde_yaml::from_str(&yaml).unwrap();
    let operation = parsed["operation"].as_mapping().unwrap();
    for forbidden in ["id", "status", "created_at", "wizard_state", "samples"] {
        assert!(!operation.contains_key(serde_yaml::Value::String(forbidden.to_owned())));
    }

    let replay = client
        .post(format!("{base_url}/operations/import?mode=upsert"))
        .header("content-type", "application/yaml")
        .body(yaml.clone())
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    assert_eq!(replay["operation_id"], operation_id);
    assert_eq!(replay["version"], 1);

    let changed_yaml = yaml.replace(
        "display_name: Create Lead",
        "display_name: Portable Changed",
    );
    assert_ne!(changed_yaml, yaml);
    let missing = client
        .post(format!("{base_url}/operations/import?mode=upsert"))
        .header("content-type", "application/yaml")
        .body(changed_yaml.clone())
        .send()
        .await
        .unwrap();
    assert_eq!(missing.status(), reqwest::StatusCode::PRECONDITION_REQUIRED);
    let missing = missing.json::<Value>().await.unwrap();
    assert_eq!(missing["error"]["code"], "operation_precondition_required");

    let changed = client
        .post(format!("{base_url}/operations/import?mode=upsert"))
        .header("content-type", "application/yaml")
        .header(
            reqwest::header::IF_MATCH,
            operation_etag(&client, &base_url, operation_id).await,
        )
        .body(changed_yaml)
        .send()
        .await
        .unwrap();
    assert_eq!(changed.status(), reqwest::StatusCode::OK);
    let changed = changed.json::<Value>().await.unwrap();
    assert_eq!(changed["version"], 2);

    let current_yaml = client
        .get(format!("{base_url}/operations/{operation_id}/export"))
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();
    let archived = client
        .post(format!("{base_url}/operations/{operation_id}/archive"))
        .header(
            reqwest::header::IF_MATCH,
            operation_etag(&client, &base_url, operation_id).await,
        )
        .send()
        .await
        .unwrap();
    assert!(archived.status().is_success());
    let replay = client
        .post(format!("{base_url}/operations/import?mode=upsert"))
        .header("content-type", "application/yaml")
        .body(current_yaml)
        .send()
        .await
        .unwrap();
    assert_eq!(replay.status(), reqwest::StatusCode::CONFLICT);
    assert!(replay.text().await.unwrap().contains("operation_archived"));
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn hostile_and_oversized_yaml_fail_before_mutation_without_echoing_canary() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("operation-yaml-hostile");
    let base_url = spawn_admin_api(build_test_app(registry.clone(), storage_root)).await;
    let client = authorized_client(&base_url).await;
    let canary = "yaml-secret-canary-never-reflect";
    let hostile = format!(
        "format_version: '2'\nkind: operation\noperation:\n  name: bad\n  unknown_secret: {canary}\n"
    );
    let response = client
        .post(format!("{base_url}/operations/import"))
        .header("content-type", "application/yaml")
        .body(hostile)
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::UNPROCESSABLE_ENTITY);
    let body = response.text().await.unwrap();
    assert!(!body.contains(canary));
    assert!(body.contains("operation_yaml_invalid"));
    let body: Value = serde_json::from_str(&body).unwrap();
    assert_eq!(body["error"]["code"], "operation_yaml_invalid");

    let oversized = "x".repeat(256 * 1024 + 1);
    let response = client
        .post(format!("{base_url}/operations/import"))
        .header("content-type", "application/yaml")
        .body(oversized)
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::PAYLOAD_TOO_LARGE);
    let count = registry
        .list_operations(&crank_core::WorkspaceId::new(WORKSPACE_ID))
        .await
        .unwrap()
        .len();
    assert_eq!(count, 0);

    let alias = "format_version: '2'\nkind: operation\noperation: &shared\n  name: alias\n";
    let response = client
        .post(format!("{base_url}/operations/import"))
        .header("content-type", "application/yaml")
        .body(alias)
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::UNPROCESSABLE_ENTITY);
    assert!(
        response
            .text()
            .await
            .unwrap()
            .contains("operation_yaml_unsupported")
    );

    let response = client
        .post(format!("{base_url}/operations/import"))
        .header("content-type", "application/yaml")
        .body(vec![0xff, 0xfe, 0xfd])
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::UNPROCESSABLE_ENTITY);
    assert!(
        response
            .text()
            .await
            .unwrap()
            .contains("operation_yaml_invalid")
    );

    let mut credential_payload = serde_json::to_value(test_operation_payload(
        "http://127.0.0.1:9",
        "portable_credential_header",
    ))
    .unwrap();
    credential_payload["target"]["static_headers"]["X-Auth-Token"] =
        Value::String("credential-canary".to_owned());
    let created = client
        .post(format!("{base_url}/operations"))
        .json(&credential_payload)
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let credential_operation_id = created["operation_id"].as_str().unwrap();
    let response = client
        .get(format!(
            "{base_url}/operations/{credential_operation_id}/export"
        ))
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::UNPROCESSABLE_ENTITY);
    assert!(!response.text().await.unwrap().contains("credential-canary"));

    let mut portable = serde_json::to_value(test_operation_payload(
        "http://127.0.0.1:9",
        "unknown_nested_yaml",
    ))
    .unwrap();
    portable.as_object_mut().unwrap().remove("wizard_state");
    portable["target"]["unknown_nested"] = Value::String(canary.to_owned());
    let document = serde_yaml::to_string(&serde_json::json!({
        "format_version": "2",
        "kind": "operation",
        "operation": portable
    }))
    .unwrap();
    let response = client
        .post(format!("{base_url}/operations/import"))
        .header("content-type", "application/yaml")
        .body(document)
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::UNPROCESSABLE_ENTITY);
    assert!(!response.text().await.unwrap().contains(canary));
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn operation_mutations_require_identity_bound_etags_and_versions_are_stable() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("operation-etag");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;

    let first = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload("http://127.0.0.1:9", "etag_first"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let second = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload("http://127.0.0.1:9", "etag_second"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let first_id = first["operation_id"].as_str().unwrap();
    let second_id = second["operation_id"].as_str().unwrap();

    let missing = client
        .post(format!("{base_url}/operations/{first_id}/archive"))
        .send()
        .await
        .unwrap();
    assert_eq!(missing.status(), reqwest::StatusCode::PRECONDITION_REQUIRED);
    let missing = missing.json::<Value>().await.unwrap();
    assert_eq!(missing["error"]["code"], "operation_precondition_required");

    let first_etag = operation_etag(&client, &base_url, first_id).await;
    let replay = client
        .post(format!("{base_url}/operations/{second_id}/archive"))
        .header(reqwest::header::IF_MATCH, first_etag)
        .send()
        .await
        .unwrap();
    assert_eq!(replay.status(), reqwest::StatusCode::CONFLICT);
    let replay = replay.json::<Value>().await.unwrap();
    assert_eq!(replay["error"]["code"], "operation_stale_version");

    let v1 = client
        .get(format!("{base_url}/operations/{first_id}/versions/1"))
        .send()
        .await
        .unwrap();
    assert_eq!(v1.status(), reqwest::StatusCode::OK);
    let v1_etag = v1.headers()[reqwest::header::ETAG]
        .to_str()
        .unwrap()
        .to_owned();
    let mut update =
        serde_json::to_value(test_operation_payload("http://127.0.0.1:9", "etag_first")).unwrap();
    update["display_name"] = Value::String("Changed Draft".to_owned());
    let updated = client
        .patch(format!("{base_url}/operations/{first_id}"))
        .header(
            reqwest::header::IF_MATCH,
            operation_etag(&client, &base_url, first_id).await,
        )
        .json(&update)
        .send()
        .await
        .unwrap();
    assert_eq!(updated.status(), reqwest::StatusCode::OK);

    let v1_after = client
        .get(format!("{base_url}/operations/{first_id}/versions/1"))
        .send()
        .await
        .unwrap();
    assert_eq!(v1_after.status(), reqwest::StatusCode::OK);
    assert_eq!(
        v1_after.headers()[reqwest::header::ETAG].to_str().unwrap(),
        v1_etag
    );
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn same_etag_concurrent_patches_have_exactly_one_winner() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("operation-etag-race");
    let base_url = spawn_admin_api(build_test_app(registry.clone(), storage_root)).await;
    let client = authorized_client(&base_url).await;
    let created = client
        .post(format!("{base_url}/operations"))
        .json(&test_operation_payload("http://127.0.0.1:9", "etag_race"))
        .send()
        .await
        .unwrap()
        .json::<Value>()
        .await
        .unwrap();
    let operation_id = created["operation_id"].as_str().unwrap();
    let etag = operation_etag(&client, &base_url, operation_id).await;
    let mut first =
        serde_json::to_value(test_operation_payload("http://127.0.0.1:9", "etag_race")).unwrap();
    first["display_name"] = Value::String("first winner candidate".to_owned());
    let mut second = first.clone();
    second["display_name"] = Value::String("second winner candidate".to_owned());
    let url = format!("{base_url}/operations/{operation_id}");
    let first_request = client
        .patch(&url)
        .header(reqwest::header::IF_MATCH, &etag)
        .json(&first)
        .send();
    let second_request = client
        .patch(&url)
        .header(reqwest::header::IF_MATCH, &etag)
        .json(&second)
        .send();
    let (first_response, second_response) = tokio::join!(first_request, second_request);
    let statuses = [
        first_response.unwrap().status(),
        second_response.unwrap().status(),
    ];
    assert_eq!(
        statuses.iter().filter(|status| status.is_success()).count(),
        1
    );
    assert_eq!(
        statuses
            .iter()
            .filter(|status| **status == reqwest::StatusCode::CONFLICT)
            .count(),
        1
    );
    let summary = registry
        .get_operation_summary(
            &crank_core::WorkspaceId::new(WORKSPACE_ID),
            &crank_core::OperationId::new(operation_id),
        )
        .await
        .unwrap()
        .unwrap();
    assert_eq!(summary.current_draft_version, 2);
}

#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn operation_rejects_unscoped_auth_profile_reference_before_save() {
    let registry = test_registry().await;
    let storage_root = test_storage_root("operation-auth-reference");
    let base_url = spawn_admin_api(build_test_app(registry, storage_root)).await;
    let client = authorized_client(&base_url).await;
    let mut payload = serde_json::to_value(test_operation_payload(
        "http://127.0.0.1:9",
        "missing_auth_profile",
    ))
    .unwrap();
    payload["execution_config"]["auth_profile_ref"] =
        Value::String("auth_foreign_or_missing".to_owned());
    let response = client
        .post(format!("{base_url}/operations"))
        .json(&payload)
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), reqwest::StatusCode::UNPROCESSABLE_ENTITY);
    let body = response.text().await.unwrap();
    assert!(body.contains("operation_auth_profile_invalid"));
    assert!(!body.contains("auth_foreign_or_missing"));
}