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

const WS: &str = "test-workspace";

/// A reqwest client that sends `Authorization: Bearer <token>`. The base
/// fixture seeds: SECRET_TOKEN (test-user, admin), SECRET_TOKEN_2
/// (test-user-2, non-admin), SECRET_TOKEN_3 (test-user-3, non-admin).
fn client_for(token: &str) -> reqwest::Client {
    let mut headers = reqwest::header::HeaderMap::new();
    headers.insert(
        reqwest::header::AUTHORIZATION,
        reqwest::header::HeaderValue::from_str(&format!("Bearer {token}")).unwrap(),
    );
    reqwest::ClientBuilder::new()
        .default_headers(headers)
        .build()
        .unwrap()
}

fn save_url(port: u16, kind: &str, path: &str) -> String {
    format!("http://localhost:{port}/api/w/{WS}/drafts/update/{kind}/{path}")
}

async fn draft_count(db: &Pool<Postgres>, path: &str, kind: &str, email: &str) -> i64 {
    sqlx::query_scalar::<_, i64>(
        "SELECT count(*) FROM draft WHERE workspace_id = $1 AND path = $2 \
         AND typ = $3::text::DRAFT_KIND AND email = $4",
    )
    .bind(WS)
    .bind(path)
    .bind(kind)
    .bind(email)
    .fetch_one(db)
    .await
    .unwrap()
}

/// Upsert → conflict (stale last_sync) → force-overwrite → delete, the
/// optimistic-concurrency contract `update_draft` exists to enforce.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_update_draft_conflict_lifecycle(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let c = client_for("SECRET_TOKEN");
    let path = "u/test-user/draft_x";
    let url = save_url(port, "script", path);

    // First save: no last_sync ("treat as fresh") → saved.
    let r = c
        .post(&url)
        .json(&json!({ "value": { "n": 1 } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "first save");
    let body: serde_json::Value = r.json().await?;
    assert_eq!(body["status"], "saved");
    let ts1 = body["current_timestamp"].as_str().unwrap().to_string();
    assert_eq!(
        draft_count(&db, path, "script", "test@windmill.dev").await,
        1
    );

    // A tiny gap so the next now() is strictly greater than ts1.
    tokio::time::sleep(std::time::Duration::from_millis(15)).await;

    // Save with the matching last_sync → not stale → saved, newer ts.
    let r = c
        .post(&url)
        .json(&json!({ "value": { "n": 2 }, "last_sync": ts1 }))
        .send()
        .await?;
    let body: serde_json::Value = r.json().await?;
    assert_eq!(body["status"], "saved", "in-order save");
    let ts2 = body["current_timestamp"].as_str().unwrap().to_string();
    assert_ne!(ts1, ts2, "timestamp should advance");

    // Save with the now-stale ts1 → conflict, server reports its current ts.
    let r = c
        .post(&url)
        .json(&json!({ "value": { "n": 3 }, "last_sync": ts1 }))
        .send()
        .await?;
    let body: serde_json::Value = r.json().await?;
    assert_eq!(body["status"], "conflict", "stale save must conflict");
    assert_eq!(body["current_timestamp"].as_str().unwrap(), ts2);

    // The conflicting write must NOT have landed — value is still {n:2}.
    let stored: serde_json::Value = sqlx::query_scalar::<_, sqlx::types::Json<serde_json::Value>>(
        "SELECT value FROM draft WHERE workspace_id = $1 AND path = $2 \
         AND typ = 'script' AND email = 'test@windmill.dev'",
    )
    .bind(WS)
    .bind(path)
    .fetch_one(&db)
    .await?
    .0;
    assert_eq!(stored["n"], 2, "conflicting write must be rejected");

    // force = true overrides the conflict check.
    let r = c
        .post(&url)
        .json(&json!({ "value": { "n": 3 }, "last_sync": ts1, "force": true }))
        .send()
        .await?;
    assert_eq!(
        r.json::<serde_json::Value>().await?["status"],
        "saved",
        "force"
    );

    // Delete (value: null) → saved, row gone.
    let r = c.post(&url).json(&json!({ "value": null })).send().await?;
    assert_eq!(
        r.json::<serde_json::Value>().await?["status"],
        "saved",
        "delete"
    );
    assert_eq!(
        draft_count(&db, path, "script", "test@windmill.dev").await,
        0,
        "row removed after delete"
    );

    Ok(())
}

/// require_can_write_path: own namespace allowed, another user's namespace
/// rejected, operators rejected outright.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_update_draft_write_authorization(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let user2 = client_for("SECRET_TOKEN_2"); // test-user-2, non-admin

    // Own namespace → allowed.
    let r = user2
        .post(save_url(port, "script", "u/test-user-2/own"))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "own namespace allowed");
    let r = user2
        .post(save_url(port, "data_pipeline", "u/test-user-2/pipeline"))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "a developer saves data pipeline drafts");

    // Another user's namespace, no grant → rejected.
    let r = user2
        .post(save_url(port, "script", "u/test-user/theirs"))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(r.status(), 401, "other user's namespace rejected");

    // Operators can't save drafts at all.
    sqlx::query(
        "UPDATE usr SET operator = true WHERE workspace_id = $1 AND username = 'test-user-3'",
    )
    .bind(WS)
    .execute(&db)
    .await?;
    let op = client_for("SECRET_TOKEN_3");
    let r = op
        .post(save_url(port, "script", "u/test-user-3/own"))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(r.status(), 403, "operator rejected");
    let r = op
        .post(save_url(port, "data_pipeline", "u/test-user-3/pipeline"))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(
        r.status(),
        403,
        "a role that cannot code has no pipeline drafts"
    );

    Ok(())
}

/// The item-level extra_perms fallback: a user granted write on a deployed
/// item (via the Share dialog) can save a draft on it even though it's
/// outside their namespace. Regression test for the authz drop.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_update_draft_extra_perms_writer(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let path = "u/test-user/shared";

    // A deployed script owned by test-user, shared with test-user-2 (write).
    sqlx::query(
        "INSERT INTO script (workspace_id, hash, path, summary, description, content, \
            language, schema, extra_perms, created_by) \
         VALUES ($1, 1, $2, '', '', 'x', 'deno', '{}'::jsonb, \
            '{\"u/test-user-2\": true}'::jsonb, 'test-user')",
    )
    .bind(WS)
    .bind(path)
    .execute(&db)
    .await?;

    let user2 = client_for("SECRET_TOKEN_2");
    let r = user2
        .post(save_url(port, "script", path))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(
        r.status(),
        200,
        "extra_perms writer can save a draft: {}",
        r.text().await?
    );

    // Without a grant on a different shared item → still rejected.
    sqlx::query(
        "INSERT INTO script (workspace_id, hash, path, summary, description, content, \
            language, schema, extra_perms, created_by) \
         VALUES ($1, 2, 'u/test-user/private', '', '', 'x', 'deno', '{}'::jsonb, \
            '{}'::jsonb, 'test-user')",
    )
    .bind(WS)
    .execute(&db)
    .await?;
    let r = user2
        .post(save_url(port, "script", "u/test-user/private"))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(r.status(), 401, "no grant → rejected");

    // A READ-ONLY grant (`extra_perms` value false) must not allow draft
    // saves: the write check defers to RLS via `SELECT ... FOR UPDATE`,
    // and locking applies the UPDATE policies — visibility under the
    // SELECT policy alone isn't enough. Pins the FOR UPDATE semantics the
    // probe relies on.
    sqlx::query(
        "INSERT INTO script (workspace_id, hash, path, summary, description, content, \
            language, schema, extra_perms, created_by) \
         VALUES ($1, 3, 'u/test-user/readonly', '', '', 'x', 'deno', '{}'::jsonb, \
            '{\"u/test-user-2\": false}'::jsonb, 'test-user')",
    )
    .bind(WS)
    .execute(&db)
    .await?;
    let r = user2
        .post(save_url(port, "script", "u/test-user/readonly"))
        .json(&json!({ "value": { "a": 1 } }))
        .send()
        .await?;
    assert_eq!(r.status(), 401, "read-only grant → rejected");

    Ok(())
}

/// Cross-user draft viewing (`GET /drafts/get/{kind}/{path}`) is disabled
/// for the drawer kinds (resource/variable/triggers) so a viewer can't
/// read another user's draft; it stays available for script/flow/app.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_cross_user_draft_privacy(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();

    // test-user saves a variable draft and a script draft in a shared folder.
    let admin = client_for("SECRET_TOKEN");
    admin
        .post(save_url(port, "variable", "f/shared/v"))
        .json(&json!({ "value": { "variable": { "value": "x", "is_secret": false } } }))
        .send()
        .await?;

    let user2 = client_for("SECRET_TOKEN_2");
    // Variable is a drawer kind → cross-user view is forbidden regardless of
    // path access (the kind gate fires first).
    let r = user2
        .get(format!(
            "http://localhost:{port}/api/w/{WS}/drafts/get/variable/f/shared/v?username=test-user"
        ))
        .send()
        .await?;
    assert_eq!(
        r.status(),
        404,
        "variable drafts are private to their owner"
    );

    // Sharing kinds (script) are NOT gated by the kind check — a missing
    // draft / no access yields 404 too, but the "private to their owner"
    // wording is specific to the drawer kinds, so assert it's absent here.
    let r = user2
        .get(format!(
            "http://localhost:{port}/api/w/{WS}/drafts/get/script/f/shared/s?username=test-user"
        ))
        .send()
        .await?;
    let body = r.text().await?;
    assert!(
        !body.contains("private to their owner"),
        "script kind must not be blocked by the cross-user privacy gate: {body}"
    );

    Ok(())
}

/// Moving a session's drafts into a fork is all-or-nothing: a draft the caller already has in
/// the fork blocks the whole transfer rather than one side silently winning, unless the caller
/// asks to replace it, and the source copy is removed only together with a successful copy.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_transfer_drafts_to_fork(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
    let c = client_for("SECRET_TOKEN");
    let me = "test@windmill.dev";
    let fork = "wm-fork-transfer";
    let transfer_url = format!("http://localhost:{port}/api/w/{WS}/drafts/transfer");
    let fork_count = |path: &'static str| {
        sqlx::query_scalar::<_, i64>(
            "SELECT count(*) FROM draft WHERE workspace_id = $1 AND path = $2 AND email = $3",
        )
        .bind(fork)
        .bind(path)
        .bind(me)
        .fetch_one(&db)
    };

    let r = c
        .post(save_url(port, "script", "u/test-user/a"))
        .json(&json!({ "value": { "content": "a" } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200);
    let r = c
        .post(format!(
            "http://localhost:{port}/api/w/{WS}/workspaces/create_fork"
        ))
        .json(&json!({ "id": fork, "name": "transfer" }))
        .send()
        .await?;
    assert!(r.status().is_success(), "create fork: {}", r.text().await?);
    // Created after the fork, so only the source has it.
    let r = c
        .post(save_url(port, "script", "u/test-user/b"))
        .json(&json!({ "value": { "content": "b" } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200);

    let items = json!([
        { "kind": "script", "path": "u/test-user/a" },
        { "kind": "script", "path": "u/test-user/b" }
    ]);

    // A token bound to the source authenticates on its routes but must not write the fork.
    sqlx::query(
        "INSERT INTO token(token_hash, token_prefix, token, email, label, super_admin, workspace_id)
         VALUES (encode(sha256('BOUND_TOKEN'::bytea), 'hex'), 'BOUND_TOK', 'BOUND_TOKEN', $1, 'bound', true, $2)",
    )
    .bind(me)
    .bind(WS)
    .execute(&db)
    .await?;
    let r = client_for("BOUND_TOKEN")
        .post(&transfer_url)
        .json(&json!({ "target_workspace": fork, "items": items }))
        .send()
        .await?;
    assert_eq!(r.status(), 403);
    assert_eq!(fork_count("u/test-user/b").await?, 0);

    // `a` was cloned into the fork: the whole transfer is refused and nothing moves.
    let body: serde_json::Value = c
        .post(&transfer_url)
        .json(&json!({ "target_workspace": fork, "items": items, "remove_from_source": true }))
        .send()
        .await?
        .json()
        .await?;
    assert_eq!(
        body["conflicts"],
        json!([{ "kind": "script", "path": "u/test-user/a" }])
    );
    assert_eq!(fork_count("u/test-user/b").await?, 0);
    assert_eq!(draft_count(&db, "u/test-user/a", "script", me).await, 1);
    assert_eq!(draft_count(&db, "u/test-user/b", "script", me).await, 1);

    // A save after the clone: replacing the clone's copy must carry it into the fork.
    let r = c
        .post(save_url(port, "script", "u/test-user/a"))
        .json(&json!({ "value": { "content": "a2" } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200);
    let body: serde_json::Value = c
        .post(&transfer_url)
        .json(&json!({
            "target_workspace": fork, "items": items,
            "remove_from_source": true, "replace_existing": true
        }))
        .send()
        .await?
        .json()
        .await?;
    assert_eq!(body["conflicts"], json!([]));
    assert_eq!(body["copied"].as_array().unwrap().len(), 2);
    assert_eq!(body["removed"].as_array().unwrap().len(), 2);
    assert_eq!(fork_count("u/test-user/a").await?, 1);
    assert_eq!(fork_count("u/test-user/b").await?, 1);
    let a_in_fork: String = sqlx::query_scalar(
        "SELECT value->>'content' FROM draft WHERE workspace_id = $1 AND path = 'u/test-user/a'",
    )
    .bind(fork)
    .fetch_one(&db)
    .await?;
    assert_eq!(a_in_fork, "a2");
    assert_eq!(draft_count(&db, "u/test-user/a", "script", me).await, 0);
    assert_eq!(draft_count(&db, "u/test-user/b", "script", me).await, 0);

    // With nothing left in the source to replace them, the fork's drafts stay.
    let r = c
        .post(&transfer_url)
        .json(&json!({
            "target_workspace": fork, "items": items,
            "remove_from_source": true, "replace_existing": true
        }))
        .send()
        .await?;
    assert_eq!(r.status(), 200);
    assert_eq!(fork_count("u/test-user/a").await?, 1);
    assert_eq!(fork_count("u/test-user/b").await?, 1);

    // An app policy naming someone who is not a member of the fork is re-pointed at the user, as
    // the fork's own clone does: a deploy in the fork rejects an identity that does not resolve.
    let r = c
        .post(save_url(port, "raw_app", "u/test-user/app"))
        .json(&json!({ "value": { "policy": {
            "on_behalf_of": "u/not-in-fork", "on_behalf_of_email": "ghost@windmill.dev"
        } } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "{}", r.text().await?);
    let r = c
        .post(&transfer_url)
        .json(&json!({
            "target_workspace": fork,
            "items": [{ "kind": "raw_app", "path": "u/test-user/app" }]
        }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "{}", r.text().await?);
    let on_behalf_of: String = sqlx::query_scalar(
        "SELECT value->'policy'->>'on_behalf_of' FROM draft
         WHERE workspace_id = $1 AND path = 'u/test-user/app'",
    )
    .bind(fork)
    .fetch_one(&db)
    .await?;
    assert_eq!(on_behalf_of, "u/test-user");

    // A script version the fork does not have (deployed in the source after the fork was made)
    // is no lineage there: a deploy naming it as parent would be refused.
    let r = c
        .post(save_url(port, "script", "u/test-user/lineage"))
        .json(&json!({ "value": { "content": "l", "parent_hash": "0123456789abcdef" } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "{}", r.text().await?);
    let r = c
        .post(&transfer_url)
        .json(&json!({
            "target_workspace": fork,
            "items": [{ "kind": "script", "path": "u/test-user/lineage" }]
        }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "{}", r.text().await?);
    let (parent_hash, base): (Option<String>, Option<String>) = sqlx::query_as(
        "SELECT value->>'parent_hash', base FROM draft
         WHERE workspace_id = $1 AND path = 'u/test-user/lineage'",
    )
    .bind(fork)
    .fetch_one(&db)
    .await?;
    assert_eq!((parent_hash, base), (None, None));

    // A version the fork has keeps its lineage. Its hash has the high bit set, so the match also
    // holds for a negative i64, which serializes as 16 hex digits.
    sqlx::query(
        "INSERT INTO script (workspace_id, hash, path, summary, description, content, created_by)
         VALUES ($1, -1, 'u/test-user/lineage', '', '', 'l', 'test-user')",
    )
    .bind(fork)
    .execute(&db)
    .await?;
    let r = c
        .post(save_url(port, "script", "u/test-user/lineage"))
        .json(&json!({ "value": { "content": "l2", "parent_hash": "ffffffffffffffff" } }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "{}", r.text().await?);
    let r = c
        .post(&transfer_url)
        .json(&json!({
            "target_workspace": fork,
            "items": [{ "kind": "script", "path": "u/test-user/lineage" }],
            "replace_existing": true
        }))
        .send()
        .await?;
    assert_eq!(r.status(), 200, "{}", r.text().await?);
    let (parent_hash, base): (Option<String>, Option<String>) = sqlx::query_as(
        "SELECT value->>'parent_hash', base FROM draft
         WHERE workspace_id = $1 AND path = 'u/test-user/lineage'",
    )
    .bind(fork)
    .fetch_one(&db)
    .await?;
    assert_eq!(
        (parent_hash.as_deref(), base.as_deref()),
        (Some("ffffffffffffffff"), Some("ffffffffffffffff"))
    );

    // Only a fork of the source is a valid target.
    let r = c
        .post(format!(
            "http://localhost:{port}/api/w/{fork}/drafts/transfer"
        ))
        .json(&json!({ "target_workspace": WS, "items": items }))
        .send()
        .await?;
    assert_eq!(r.status(), 400);

    Ok(())
}