use athena_rs::client::{AthenaClient, backend::BackendType};
use serde_json::Value;
use testcontainers::{
    ContainerAsync, GenericImage, ImageExt, core::IntoContainerPort, runners::AsyncRunner,
};
use tokio_postgres::NoTls;

async fn start_postgres() -> (ContainerAsync<GenericImage>, String) {
    let image = GenericImage::new("postgres", "16-alpine").with_exposed_port(5432.tcp());
    let request = image
        .with_env_var("POSTGRES_PASSWORD", "postgres")
        .with_env_var("POSTGRES_DB", "postgres");
    let container = request.start().await.expect("start postgres container");
    let port = container
        .get_host_port_ipv4(5432)
        .await
        .expect("resolve mapped port");
    let conn_str = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
    wait_for_postgres(&conn_str).await;
    (container, conn_str)
}

async fn wait_for_postgres(conn_str: &str) {
    for _ in 0..40 {
        if tokio_postgres::connect(conn_str, NoTls).await.is_ok() {
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
    }
    panic!("postgres did not become ready in time");
}

async fn seed_orchestra(conn_str: &str) {
    let (client, connection) = tokio_postgres::connect(conn_str, NoTls)
        .await
        .expect("connect postgres");
    tokio::spawn(async move {
        let _ = connection.await;
    });

    client
        .batch_execute(
            r#"
            DROP TABLE IF EXISTS instruments;
            DROP TABLE IF EXISTS orchestral_sections;
            CREATE TABLE orchestral_sections (
                id SERIAL PRIMARY KEY,
                name TEXT NOT NULL
            );
            CREATE TABLE instruments (
                id SERIAL PRIMARY KEY,
                name TEXT NOT NULL,
                section_id INTEGER REFERENCES orchestral_sections(id)
            );
            INSERT INTO orchestral_sections (name) VALUES
                ('strings'), ('woodwinds'), ('percussion');
            INSERT INTO instruments (name, section_id) VALUES
                ('violin', 1), ('viola', 1), ('flute', 2), ('oboe', 2);
            "#,
        )
        .await
        .expect("seed orchestra tables");
}

async fn seed_shifts(conn_str: &str) {
    let (client, connection) = tokio_postgres::connect(conn_str, NoTls)
        .await
        .expect("connect postgres");
    tokio::spawn(async move {
        let _ = connection.await;
    });

    client
        .batch_execute(
            r#"
            DROP TABLE IF EXISTS shifts;
            DROP TABLE IF EXISTS scans;
            CREATE TABLE scans (
                id SERIAL PRIMARY KEY,
                user_id INT NOT NULL,
                badge_scan_time TIMESTAMP NOT NULL
            );
            CREATE TABLE shifts (
                id SERIAL PRIMARY KEY,
                user_id INT NOT NULL,
                scan_id_start INT REFERENCES scans(id),
                scan_id_end INT REFERENCES scans(id),
                attendance_status TEXT
            );
            INSERT INTO scans (user_id, badge_scan_time) VALUES
                (1, NOW()), (1, NOW() + interval '8 hours');
            INSERT INTO shifts (user_id, scan_id_start, scan_id_end, attendance_status)
            VALUES (1, 1, 2, 'present');
            "#,
        )
        .await
        .expect("seed shifts");
}

async fn seed_many_to_many(conn_str: &str) {
    let (client, connection) = tokio_postgres::connect(conn_str, NoTls)
        .await
        .expect("connect postgres");
    tokio::spawn(async move {
        let _ = connection.await;
    });

    client
        .batch_execute(
            r#"
            DROP TABLE IF EXISTS members;
            DROP TABLE IF EXISTS users;
            DROP TABLE IF EXISTS teams;
            CREATE TABLE users (id SERIAL PRIMARY KEY, name TEXT NOT NULL);
            CREATE TABLE teams (id SERIAL PRIMARY KEY, team_name TEXT NOT NULL);
            CREATE TABLE members (
                user_id INT REFERENCES users(id),
                team_id INT REFERENCES teams(id),
                PRIMARY KEY (user_id, team_id)
            );
            INSERT INTO users (name) VALUES ('alice'), ('bob'), ('carol');
            INSERT INTO teams (team_name) VALUES ('eng'), ('ops');
            INSERT INTO members (user_id, team_id) VALUES
                (1, 1), (2, 1), (3, 2);
            "#,
        )
        .await
        .expect("seed many-to-many");
}

async fn make_client(conn_str: &str) -> AthenaClient {
    let builder = AthenaClient::builder()
        .backend(BackendType::PostgreSQL)
        .url(conn_str.to_string());
    AthenaClient::build(builder).await.expect("build client")
}

#[tokio::test]
#[ignore = "requires local Docker daemon"]
async fn left_join_nested_relation_matches_supabase_semantics() {
    let (_node, conn_str) = start_postgres().await;
    seed_orchestra(&conn_str).await;
    let client = make_client(&conn_str).await;

    let result = client
        .select("orchestral_sections")
        .raw_select("id,name,instruments(id,name)")
        .where_eq("instruments.name", "flute")
        .execute()
        .await
        .expect("query");

    // left join semantics: all parents returned, empty arrays when no match
    assert_eq!(result.rows.len(), 3);
    let woodwinds = result
        .rows
        .iter()
        .find(|row| row.get("name") == Some(&Value::String("woodwinds".into())))
        .expect("woodwinds present");
    assert_eq!(
        woodwinds
            .get("instruments")
            .and_then(|v| v.as_array())
            .map(|arr| arr.len()),
        Some(1)
    );
    let strings = result
        .rows
        .iter()
        .find(|row| row.get("name") == Some(&Value::String("strings".into())))
        .expect("strings present");
    assert_eq!(
        strings
            .get("instruments")
            .and_then(|v| v.as_array())
            .map(|arr| arr.len()),
        Some(0)
    );
}

#[tokio::test]
#[ignore = "requires local Docker daemon"]
async fn inner_join_nested_relation_filters_parent_rows() {
    let (_node, conn_str) = start_postgres().await;
    seed_orchestra(&conn_str).await;
    let client = make_client(&conn_str).await;

    let result = client
        .select("orchestral_sections")
        .raw_select("id,name,instruments!inner(id,name)")
        .where_eq("instruments.name", "flute")
        .execute()
        .await
        .expect("query");

    assert_eq!(result.rows.len(), 1);
    let row = &result.rows[0];
    assert_eq!(row.get("name"), Some(&Value::String("woodwinds".into())));
    let instruments = row
        .get("instruments")
        .and_then(|v| v.as_array())
        .expect("array");
    assert_eq!(instruments.len(), 1);
}

#[tokio::test]
#[ignore = "requires local Docker daemon"]
async fn explicit_foreign_key_direction_is_honored() {
    let (_node, conn_str) = start_postgres().await;
    seed_shifts(&conn_str).await;
    let client = make_client(&conn_str).await;

    let result = client
        .select("shifts")
        .raw_select(
            "id,user_id,attendance_status,\
            start_scan:scans!parent.scan_id_start(id,user_id,badge_scan_time),\
            end_scan:scans!parent.scan_id_end(id,user_id,badge_scan_time)",
        )
        .where_eq("start_scan.user_id", 1)
        .execute()
        .await
        .expect("query");

    assert_eq!(result.rows.len(), 1);
    let row = &result.rows[0];
    let start_scan = row
        .get("start_scan")
        .and_then(|v| v.get("id"))
        .and_then(Value::as_i64);
    assert_eq!(start_scan, Some(1));
}

#[tokio::test]
#[ignore = "requires local Docker daemon"]
async fn many_to_many_relations_are_aggregated() {
    let (_node, conn_str) = start_postgres().await;
    seed_many_to_many(&conn_str).await;
    let client = make_client(&conn_str).await;

    let result = client
        .select("teams")
        .raw_select("id,team_name,users(id,name)")
        .order_by("id", athena_rs::client::query_builder::OrderDirection::Asc)
        .execute()
        .await
        .expect("query");

    assert_eq!(result.rows.len(), 2);
    let eng = &result.rows[0];
    let users = eng
        .get("users")
        .and_then(|v| v.as_array())
        .expect("users array");
    assert_eq!(users.len(), 2);
}
