use serde_json::json;
use sqlx::{Pool, Postgres};

use windmill_test_utils::*;

fn client() -> reqwest::Client {
    reqwest::Client::new()
}

fn authed(builder: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
    builder.header("Authorization", "Bearer SECRET_TOKEN")
}

fn ws_url(port: u16, endpoint: &str) -> String {
    format!("http://localhost:{port}/api/w/test-workspace/users/{endpoint}")
}

fn global_url(port: u16, endpoint: &str) -> String {
    format!("http://localhost:{port}/api/users/{endpoint}")
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_preview(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    let resp = authed(client().get(ws_url(port, "offboard_preview/test-user-2")))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    let body: serde_json::Value = resp.json().await?;
    assert_eq!(body["owned"]["scripts"].as_array().unwrap().len(), 3); // script_a, script_b, conflict_script
    assert_eq!(body["owned"]["flows"].as_array().unwrap().len(), 1); // flow_a
    assert_eq!(body["owned"]["resources"].as_array().unwrap().len(), 1); // res_a
    assert_eq!(body["owned"]["variables"].as_array().unwrap().len(), 1); // var_a
    assert_eq!(body["owned"]["schedules"].as_array().unwrap().len(), 1); // sched_a
    assert_eq!(
        body["owned"]["triggers"]["http_trigger"]
            .as_array()
            .unwrap()
            .len(),
        1
    ); // webhook_a (dynamic trigger query)
    assert_eq!(body["tokens"].as_array().unwrap().len(), 1); // OFFBOARD_TOKEN_1
    assert_eq!(body["tokens"][0]["label"], "offboard test");
    assert_eq!(body["http_triggers"], 1); // webhook_a
    assert_eq!(body["email_triggers"], 0);
    // Operator references: schedule + trigger at f/test-folder/ run as u/test-user-2
    assert_eq!(
        body["executing_on_behalf"]["schedules"]
            .as_array()
            .unwrap()
            .len(),
        1
    );
    assert_eq!(
        body["executing_on_behalf"]["triggers"]["http_trigger"]
            .as_array()
            .unwrap()
            .len(),
        1,
        "shared trigger permissioned_as should be detected"
    );

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_to_user(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Remove conflict scripts so offboard can proceed
    sqlx::query!(
        "DELETE FROM script WHERE hash IN (1003, 1004) AND workspace_id = 'test-workspace'"
    )
    .execute(&db)
    .await?;

    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "u/test-user",
            "new_on_behalf_of_user": "test-user",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    let body: serde_json::Value = resp.json().await?;
    assert!(body["conflicts"].as_array().map_or(true, |a| a.is_empty()));
    assert!(body["summary"].is_object());

    let summary = &body["summary"];
    assert!(summary["scripts_reassigned"].as_i64().unwrap() > 0);
    assert!(summary["flows_reassigned"].as_i64().unwrap() > 0);
    assert!(summary["resources_reassigned"].as_i64().unwrap() > 0);
    assert!(summary["variables_reassigned"].as_i64().unwrap() > 0);
    assert!(summary["schedules_reassigned"].as_i64().unwrap() > 0);

    // Verify scripts moved
    let moved = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM script WHERE path LIKE 'u/test-user/%' AND workspace_id = 'test-workspace' AND NOT archived AND NOT deleted"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(0);
    assert!(moved > 0, "scripts should be under u/test-user now");

    let old = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM script WHERE path LIKE 'u/test-user-2/%' AND workspace_id = 'test-workspace' AND NOT archived AND NOT deleted"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(0);
    assert_eq!(old, 0, "no scripts should remain under u/test-user-2");

    // Verify schedule permissioned_as updated
    let perm = sqlx::query_scalar!(
        "SELECT permissioned_as FROM schedule WHERE path = 'u/test-user/sched_a' AND workspace_id = 'test-workspace'"
    )
    .fetch_optional(&db)
    .await?;
    assert_eq!(perm.as_deref(), Some("u/test-user"));

    // Verify user deleted from workspace
    let user_exists = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM usr WHERE username = 'test-user-2' AND workspace_id = 'test-workspace')"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(true);
    assert!(!user_exists, "user should be removed from workspace");

    // Verify tokens revoked (default behavior when no reassign_tokens_to)
    let token_count = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM token WHERE owner = 'u/test-user-2' AND workspace_id = 'test-workspace'"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(1);
    assert_eq!(token_count, 0, "tokens should be revoked");

    // Verify HTTP trigger reassigned (dynamic trigger query)
    let trigger_moved = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM http_trigger WHERE path = 'u/test-user/webhook_a' AND workspace_id = 'test-workspace'"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(0);
    assert_eq!(trigger_moved, 1, "http trigger should be reassigned");

    let trigger_perm = sqlx::query_scalar!(
        "SELECT permissioned_as FROM http_trigger WHERE path = 'u/test-user/webhook_a' AND workspace_id = 'test-workspace'"
    )
    .fetch_optional(&db)
    .await?;
    assert_eq!(
        trigger_perm.as_deref(),
        Some("u/test-user"),
        "trigger permissioned_as should be updated"
    );

    // Verify extra_perms cleaned up (dynamic extra_perms query)
    let extra_perms = sqlx::query_scalar!(
        "SELECT extra_perms::text FROM script WHERE path = 'f/test-folder/shared_script' AND workspace_id = 'test-workspace'"
    )
    .fetch_optional(&db)
    .await?;
    assert!(
        !extra_perms
            .unwrap_or_default()
            .unwrap_or_default()
            .contains("test-user-2"),
        "extra_perms should no longer reference test-user-2"
    );

    // Verify operator schedule updated (schedule not under user's path)
    let shared_sched_perm = sqlx::query_scalar!(
        "SELECT permissioned_as FROM schedule WHERE path = 'f/test-folder/sched_shared' AND workspace_id = 'test-workspace'"
    )
    .fetch_optional(&db)
    .await?;
    assert_eq!(
        shared_sched_perm.as_deref(),
        Some("u/test-user"),
        "shared schedule permissioned_as should be updated"
    );

    // Verify shared trigger permissioned_as updated (line 951 - dynamic query)
    let shared_trigger_perm = sqlx::query_scalar!(
        "SELECT permissioned_as FROM http_trigger WHERE path = 'f/test-folder/webhook_shared' AND workspace_id = 'test-workspace'"
    )
    .fetch_optional(&db)
    .await?;
    assert_eq!(
        shared_trigger_perm.as_deref(),
        Some("u/test-user"),
        "shared trigger permissioned_as should be updated"
    );

    // Verify extra_perms cleaned on trigger (line 983 - dynamic query on trigger table)
    let trigger_extra_perms = sqlx::query_scalar!(
        "SELECT extra_perms::text FROM http_trigger WHERE path = 'f/test-folder/webhook_shared' AND workspace_id = 'test-workspace'"
    )
    .fetch_optional(&db)
    .await?;
    assert!(
        !trigger_extra_perms
            .unwrap_or_default()
            .unwrap_or_default()
            .contains("test-user-2"),
        "trigger extra_perms should no longer reference test-user-2"
    );

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_to_folder(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "f/test-folder",
            "new_on_behalf_of_user": "test-user",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    let body: serde_json::Value = resp.json().await?;
    assert!(body["conflicts"].as_array().map_or(true, |a| a.is_empty()));

    // Verify specific script moved to folder (not just pre-existing shared_script)
    let script_a_moved = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM script WHERE path = 'f/test-folder/script_a' AND workspace_id = 'test-workspace')"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(false);
    assert!(script_a_moved, "script_a should be moved to f/test-folder/");

    // Verify no scripts remain under old path
    let old_scripts = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM script WHERE path LIKE 'u/test-user-2/%' AND workspace_id = 'test-workspace' AND NOT archived AND NOT deleted"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(1);
    assert_eq!(
        old_scripts, 0,
        "no scripts should remain under u/test-user-2"
    );

    // Verify schedule permissioned_as is u/test-user (operator), not g/f/test-folder
    let perm = sqlx::query_scalar!(
        "SELECT permissioned_as FROM schedule WHERE path = 'f/test-folder/sched_a' AND workspace_id = 'test-workspace'"
    )
    .fetch_optional(&db)
    .await?;
    assert_eq!(perm.as_deref(), Some("u/test-user"));

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_reassign_only(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Remove conflict scripts so reassignment can proceed
    sqlx::query!(
        "DELETE FROM script WHERE hash IN (1003, 1004) AND workspace_id = 'test-workspace'"
    )
    .execute(&db)
    .await?;

    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "u/test-user",
            "new_on_behalf_of_user": "test-user",
            "delete_user": false
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    let body: serde_json::Value = resp.json().await?;
    assert!(
        body["conflicts"].as_array().map_or(true, |c| c.is_empty()),
        "should have no conflicts"
    );
    assert!(body["summary"].is_object(), "should have a summary");

    // Verify user still exists in workspace
    let user_exists = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM usr WHERE username = 'test-user-2' AND workspace_id = 'test-workspace')"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(false);
    assert!(
        user_exists,
        "user should still exist when delete_user=false"
    );

    // Verify items were actually reassigned
    let moved = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM script WHERE path LIKE 'u/test-user/%' AND workspace_id = 'test-workspace'"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(0);
    assert!(moved > 0, "scripts should be under u/test-user now");

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_conflicts(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // conflict_script exists under both u/test-user-2/ and u/test-user/
    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "u/test-user",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    let body: serde_json::Value = resp.json().await?;
    let conflicts = body["conflicts"]
        .as_array()
        .expect("conflicts should be array");
    assert!(!conflicts.is_empty(), "should have path conflicts");
    assert!(
        conflicts
            .iter()
            .any(|c| c.as_str().unwrap().contains("conflict_script")),
        "conflict should mention conflict_script"
    );
    // No summary when conflicts exist
    assert!(body["summary"].is_null());

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_tokens_deleted(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Remove conflict_script so offboard can proceed
    sqlx::query!("DELETE FROM script WHERE hash = 1004 AND workspace_id = 'test-workspace'")
        .execute(&db)
        .await?;

    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "u/test-user",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    // Verify tokens are deleted
    let token_count = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM token WHERE owner = 'u/test-user-2' AND workspace_id = 'test-workspace'"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(1);
    assert_eq!(token_count, 0, "tokens should be deleted after offboarding");

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_folder_requires_operator(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Missing new_on_behalf_of_user when reassigning to folder should fail
    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "f/test-folder",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 400);

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_global_offboard_preview(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    let resp = authed(client().get(global_url(port, "offboard_preview/test2@windmill.dev")))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    let body: serde_json::Value = resp.json().await?;
    let workspaces = body["workspaces"]
        .as_array()
        .expect("should have workspaces");
    assert_eq!(workspaces.len(), 2, "user is in two workspaces");

    // Both workspaces should have the user's items
    for ws in workspaces {
        assert_eq!(ws["username"], "test-user-2");
        assert!(
            !ws["preview"]["owned"]["scripts"]
                .as_array()
                .unwrap()
                .is_empty(),
            "workspace {} should have owned scripts",
            ws["workspace_id"]
        );
    }

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_global_offboard_execution(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Remove conflict scripts so offboard can proceed in test-workspace
    sqlx::query!(
        "DELETE FROM script WHERE hash IN (1003, 1004) AND workspace_id = 'test-workspace'"
    )
    .execute(&db)
    .await?;

    let resp = authed(client().post(global_url(port, "offboard/test2@windmill.dev")))
        .json(&json!({
            "reassignments": {
                "test-workspace": {
                    "reassign_to": "u/test-user",
                    "new_on_behalf_of_user": "test-user"
                },
                "test-workspace-2": {
                    "reassign_to": "u/test-user",
                    "new_on_behalf_of_user": "test-user"
                }
            },
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);

    let body: serde_json::Value = resp.json().await?;
    assert!(
        body["conflicts"].as_array().map_or(true, |c| c.is_empty()),
        "should have no conflicts"
    );
    assert!(body["summary"].is_object(), "should have a summary");
    let summary = &body["summary"];
    assert!(
        summary["scripts_reassigned"].as_i64().unwrap() > 0,
        "should have reassigned scripts across workspaces"
    );

    // Verify items moved in workspace 1
    let ws1_moved = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM script WHERE path LIKE 'u/test-user/%' AND workspace_id = 'test-workspace' AND NOT archived AND NOT deleted"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(0);
    assert!(ws1_moved > 0, "scripts should be moved in test-workspace");

    // Verify items moved in workspace 2
    let ws2_moved = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM script WHERE path LIKE 'u/test-user/%' AND workspace_id = 'test-workspace-2' AND NOT archived AND NOT deleted"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(0);
    assert!(ws2_moved > 0, "scripts should be moved in test-workspace-2");

    // Verify user deleted from both workspaces
    let ws1_user = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM usr WHERE email = 'test2@windmill.dev' AND workspace_id = 'test-workspace')"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(true);
    assert!(!ws1_user, "user should be removed from test-workspace");

    let ws2_user = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM usr WHERE email = 'test2@windmill.dev' AND workspace_id = 'test-workspace-2')"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(true);
    assert!(!ws2_user, "user should be removed from test-workspace-2");

    // Verify global auth deleted (password row)
    let password_exists = sqlx::query_scalar!(
        "SELECT EXISTS(SELECT 1 FROM password WHERE email = 'test2@windmill.dev')"
    )
    .fetch_one(&db)
    .await?
    .unwrap_or(true);
    assert!(!password_exists, "password should be deleted from instance");

    Ok(())
}

#[sqlx::test(migrations = "../migrations", fixtures("base", "offboarding_test"))]
async fn test_offboard_invalid_target(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // Reassign to nonexistent user
    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "u/nonexistent-user",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(
        resp.status(),
        400,
        "should reject reassignment to nonexistent user"
    );

    // Reassign to nonexistent folder
    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "f/nonexistent-folder",
            "new_on_behalf_of_user": "test-user",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(
        resp.status(),
        400,
        "should reject reassignment to nonexistent folder"
    );

    // Invalid target format
    let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
        .json(&json!({
            "reassign_to": "invalid-format",
            "delete_user": true
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 400, "should reject invalid target format");

    Ok(())
}