dbmcp 0.14.0

//! Functional tests for `SQLite`.
//!
//! Tests exercise the MCP tool layer (not handler methods directly),
//! ensuring the same code path as real MCP clients. Assumes a pre-seeded
//! `database.db` generated by Docker Compose (`docker compose up sqlite`).
//! The `DB_PATH` env var points to the database.
//!
//! ```bash
//! ./tests/run.sh --filter sqlite
//! ```

use dbmcp_config::{Config, DatabaseBackend, DatabaseConfig, PiiConfig, PiiOperator};
use dbmcp_sqlite::SqliteHandler;
use dbmcp_sqlite::types::{
    DropTableRequest, ExplainQueryRequest, ListTablesRequest, ListTriggersRequest, ListViewsRequest, QueryRequest,
    ReadQueryRequest,
};
use serde_json::Value;

fn base_db_config(read_only: bool) -> DatabaseConfig {
    DatabaseConfig {
        backend: DatabaseBackend::Sqlite,
        name: Some(std::env::var("DB_PATH").expect("DB_PATH must be set")),
        read_only,
        ..DatabaseConfig::default()
    }
}

fn handler(read_only: bool) -> SqliteHandler {
    let config = base_db_config(read_only);
    SqliteHandler::new(&Config {
        database: config,
        http: None,
        pii: PiiConfig::default(),
    })
    .expect("handler builds")
}

fn handler_with_page_size(page_size: u16) -> SqliteHandler {
    let config = DatabaseConfig {
        page_size,
        ..base_db_config(false)
    };
    SqliteHandler::new(&Config {
        database: config,
        http: None,
        pii: PiiConfig::default(),
    })
    .expect("handler builds")
}

#[tokio::test]
async fn test_write_query_insert_and_verify() {
    let handler = handler(false);

    let insert = QueryRequest {
        query: "INSERT INTO users (name, email) VALUES ('WriteTest', 'write@test.com')".into(),
    };
    handler.write_query(insert).await.unwrap();

    // Verify
    let select = ReadQueryRequest {
        query: "SELECT name FROM users WHERE email = 'write@test.com'".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    let arr = &rows.rows;
    assert_eq!(arr.len(), 1);
    assert_eq!(arr[0]["name"], "WriteTest");

    // Clean up
    let delete = QueryRequest {
        query: "DELETE FROM users WHERE email = 'write@test.com'".into(),
    };
    handler.write_query(delete).await.unwrap();
}

#[tokio::test]
async fn test_write_query_update() {
    let handler = handler(false);

    let insert = QueryRequest {
        query: "INSERT INTO users (name, email) VALUES ('Before', 'update@test.com')".into(),
    };
    handler.write_query(insert).await.unwrap();

    let update = QueryRequest {
        query: "UPDATE users SET name = 'After' WHERE email = 'update@test.com'".into(),
    };
    handler.write_query(update).await.unwrap();

    let select = ReadQueryRequest {
        query: "SELECT name FROM users WHERE email = 'update@test.com'".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    let arr = &rows.rows;
    assert_eq!(arr[0]["name"], "After");

    // Clean up
    let delete = QueryRequest {
        query: "DELETE FROM users WHERE email = 'update@test.com'".into(),
    };
    handler.write_query(delete).await.unwrap();
}

#[tokio::test]
async fn test_write_query_delete() {
    let handler = handler(false);

    let insert = QueryRequest {
        query: "INSERT INTO users (name, email) VALUES ('Deletable', 'delete@test.com')".into(),
    };
    handler.write_query(insert).await.unwrap();

    let delete = QueryRequest {
        query: "DELETE FROM users WHERE email = 'delete@test.com'".into(),
    };
    handler.write_query(delete).await.unwrap();

    let select = ReadQueryRequest {
        query: "SELECT * FROM users WHERE email = 'delete@test.com'".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    let arr = &rows.rows;
    assert!(arr.is_empty(), "Row should be deleted");
}

#[tokio::test]
async fn test_lists_tables() {
    let handler = handler(false);

    let response = handler.list_tables(ListTablesRequest::default()).await.unwrap();
    let tables = response.entries.as_brief().expect("brief mode");

    for expected in ["users", "posts", "tags", "post_tags"] {
        assert!(
            tables.iter().any(|t| t == expected),
            "Missing '{expected}' in: {tables:?}"
        );
    }
}

#[tokio::test]
async fn test_executes_sql() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT * FROM users ORDER BY id".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    assert_eq!(response.rows.len(), 3, "Expected 3 users, got {}", response.rows.len());
}

#[tokio::test]
async fn test_blocks_writes_in_read_only_mode() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "INSERT INTO users (name, email) VALUES ('Hacker', 'hack@evil.com')".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await;

    assert!(response.is_err(), "Expected error for write in read-only mode");
}

#[tokio::test]
async fn test_query_timeout_fast_query_succeeds() {
    let config = DatabaseConfig {
        query_timeout: Some(5),
        ..base_db_config(false)
    };
    let handler = SqliteHandler::new(&Config {
        database: config,
        http: None,
        pii: PiiConfig::default(),
    })
    .expect("handler builds");
    let request = ReadQueryRequest {
        query: "SELECT 1 AS value".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await;
    assert!(response.is_ok(), "Fast query should succeed within timeout");
}

#[tokio::test]
async fn test_query_timeout_disabled_with_none() {
    let config = DatabaseConfig {
        query_timeout: None,
        ..base_db_config(false)
    };
    let handler = SqliteHandler::new(&Config {
        database: config,
        http: None,
        pii: PiiConfig::default(),
    })
    .expect("handler builds");
    let request = ReadQueryRequest {
        query: "SELECT * FROM users ORDER BY id".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await;
    assert!(response.is_ok(), "Query should succeed without timeout");
}

#[tokio::test]
async fn test_drop_table_success() {
    let handler = handler(false);

    // Create a temporary table
    let create = QueryRequest {
        query: "CREATE TABLE drop_test_simple (id INTEGER PRIMARY KEY)".into(),
    };
    handler.write_query(create).await.unwrap();

    // Drop it
    let drop_request = DropTableRequest {
        table: "drop_test_simple".into(),
    };
    let response = handler.drop_table(drop_request).await.unwrap();
    assert!(response.message.contains("dropped successfully"));

    // Verify it's gone
    let response = handler.list_tables(ListTablesRequest::default()).await.unwrap();
    let tables = response.entries.as_brief().expect("brief mode");
    assert!(
        !tables.iter().any(|t| t == "drop_test_simple"),
        "Table should not exist after drop"
    );
}

#[tokio::test]
async fn test_drop_table_nonexistent() {
    let handler = handler(false);
    let drop_request = DropTableRequest {
        table: "nonexistent_table_xyz".into(),
    };

    let response = handler.drop_table(drop_request).await;
    assert!(response.is_err(), "Expected error for nonexistent table");
}

#[tokio::test]
async fn test_explain_query_select() {
    let handler = handler(false);
    let request = ExplainQueryRequest {
        query: "SELECT * FROM users".into(),
    };

    let response = handler.explain_query(request).await.unwrap();
    let plan = &response.rows;
    assert!(!plan.is_empty(), "Expected non-empty execution plan");
}

#[tokio::test]
async fn test_explain_query_with_join() {
    let handler = handler(false);
    let request = ExplainQueryRequest {
        query: "SELECT p.title, u.name FROM posts p JOIN users u ON p.user_id = u.id".into(),
    };

    let response = handler.explain_query(request).await.unwrap();
    let plan = &response.rows;
    assert!(!plan.is_empty(), "EXPLAIN QUERY PLAN should return plan for JOIN query");
}

#[tokio::test]
async fn test_explain_query_invalid_sql() {
    let handler = handler(false);
    let request = ExplainQueryRequest {
        query: "NOT VALID SQL AT ALL".into(),
    };

    let response = handler.explain_query(request).await;
    assert!(response.is_err(), "Expected error for invalid SQL");
}

#[tokio::test]
async fn test_explain_query_multi_statement_blocked() {
    let handler = handler(true);
    let request = ExplainQueryRequest {
        query: "SELECT 1; CREATE TABLE readonly_probe_explain(i int)".into(),
    };

    let response = handler.explain_query(request).await;
    let err_msg = format!(
        "{:?}",
        response.expect_err("Expected multi-statement EXPLAIN to be blocked")
    );
    assert!(
        err_msg.contains("single statements"),
        "Expected multi-statement error, got: {err_msg}"
    );

    let select = ReadQueryRequest {
        query: "SELECT COUNT(*) AS n FROM sqlite_master WHERE type = 'table' AND name = 'readonly_probe_explain'"
            .into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    assert_eq!(rows.rows[0]["n"], 0, "Second statement must not have executed");
}

#[tokio::test]
async fn test_drop_table_invalid_identifier() {
    let handler = handler(false);
    let drop_request = DropTableRequest { table: String::new() };

    let response = handler.drop_table(drop_request).await;
    assert!(response.is_err(), "Expected error for empty table name");
}

#[tokio::test]
async fn test_read_query_empty_query() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: String::new(),
        cursor: None,
    };

    let response = handler.read_query(request).await;
    assert!(response.is_err(), "Expected error for empty query");
}

#[tokio::test]
async fn test_read_query_whitespace_only_query() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "   \t\n  ".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await;
    assert!(response.is_err(), "Expected error for whitespace-only query");
}

#[tokio::test]
async fn test_read_query_multi_statement_blocked() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT 1; DROP TABLE users".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await;
    assert!(response.is_err(), "Expected error for multi-statement query");
}

#[tokio::test]
async fn test_read_query_into_outfile_blocked() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT * FROM users INTO OUTFILE '/tmp/out'".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await;
    assert!(response.is_err(), "Expected error for INTO OUTFILE");
}

#[tokio::test]
async fn test_read_query_explain_write_blocked() {
    let handler = handler(false);
    for sql in [
        "EXPLAIN DELETE FROM users WHERE id < 0",
        "EXPLAIN QUERY PLAN DELETE FROM users WHERE id < 0",
    ] {
        let request = ReadQueryRequest {
            query: sql.into(),
            cursor: None,
        };
        let err_msg = format!(
            "{:?}",
            handler
                .read_query(request)
                .await
                .expect_err("Expected EXPLAIN write to be blocked")
        );
        assert!(
            err_msg.contains("read-only mode"),
            "Unexpected error for {sql:?}: {err_msg}"
        );
    }

    let request = ReadQueryRequest {
        query: "EXPLAIN QUERY PLAN SELECT * FROM users".into(),
        cursor: None,
    };
    assert!(
        handler.read_query(request).await.is_ok(),
        "EXPLAIN of a read must still work"
    );
}

#[tokio::test]
async fn test_read_query_dangerous_functions_blocked() {
    let handler = handler(false);
    for sql in [
        "SELECT load_extension('/tmp/evil.so')",
        "SELECT writefile('/tmp/x', 'data')",
    ] {
        let request = ReadQueryRequest {
            query: sql.into(),
            cursor: None,
        };
        let err_msg = format!(
            "{:?}",
            handler
                .read_query(request)
                .await
                .expect_err("Expected dangerous function to be blocked")
        );
        assert!(
            err_msg.contains("is not allowed"),
            "Unexpected error for {sql:?}: {err_msg}"
        );
    }
}

#[tokio::test]
async fn test_read_query_with_join() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT p.title, u.name FROM posts p JOIN users u ON p.user_id = u.id ORDER BY p.id".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert!(!rows.is_empty(), "JOIN query should return results");
    assert!(rows[0].get("title").is_some(), "Should have title column");
    assert!(rows[0].get("name").is_some(), "Should have name column");
}

#[tokio::test]
async fn test_write_query_create_table() {
    let handler = handler(false);

    let create = QueryRequest {
        query: "CREATE TABLE write_test_create (id INTEGER PRIMARY KEY, value TEXT)".into(),
    };
    handler.write_query(create).await.unwrap();

    // Verify it appears in list_tables
    let response = handler.list_tables(ListTablesRequest::default()).await.unwrap();
    let tables = response.entries.as_brief().expect("brief mode");
    assert!(
        tables.iter().any(|t| t == "write_test_create"),
        "Created table should appear in list"
    );

    // Clean up
    let drop = DropTableRequest {
        table: "write_test_create".into(),
    };
    handler.drop_table(drop).await.unwrap();
}

#[tokio::test]
async fn test_read_query_empty_result_set() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT * FROM users WHERE email = 'nobody@nowhere.com'".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert!(rows.is_empty(), "Expected empty result set");
}

#[tokio::test]
async fn test_read_query_aggregate() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT COUNT(*) AS total FROM users".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0]["total"], 3);
}

#[tokio::test]
async fn test_read_query_group_by() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT user_id, COUNT(*) AS post_count FROM posts GROUP BY user_id ORDER BY user_id".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert!(rows.len() >= 2, "Expected at least 2 groups");
}

#[tokio::test]
async fn test_read_query_with_comments() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "/* fetch users */ SELECT * FROM users ORDER BY id".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert_eq!(rows.len(), 3, "Comment-prefixed SELECT should work");
}

#[tokio::test]
async fn test_read_query_subquery() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT * FROM users WHERE id IN (SELECT user_id FROM posts WHERE published = 1)".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert!(!rows.is_empty(), "Subquery should return results");
}

#[tokio::test]
async fn test_write_query_invalid_sql() {
    let handler = handler(false);
    let request = QueryRequest {
        query: "NOT VALID SQL AT ALL".into(),
    };

    let response = handler.write_query(request).await;
    assert!(response.is_err(), "Expected error for invalid SQL in write_query");
}

#[tokio::test]
async fn test_read_query_with_limit() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT * FROM users ORDER BY id LIMIT 2".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert_eq!(rows.len(), 2, "LIMIT 2 should return exactly 2 rows");
}

#[tokio::test]
async fn test_read_query_with_line_comment() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "-- get users\nSELECT * FROM users ORDER BY id".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert_eq!(rows.len(), 3, "Line-comment prefixed SELECT should work");
}

#[tokio::test]
async fn test_read_query_null_values() {
    let handler = handler(false);
    let request = ReadQueryRequest {
        query: "SELECT title, body FROM posts WHERE title = 'My First Post'".into(),
        cursor: None,
    };

    let response = handler.read_query(request).await.unwrap();
    let rows = &response.rows;
    assert_eq!(rows.len(), 1);
    assert!(rows[0].get("body").is_some(), "body column should be present");
}

#[tokio::test]
async fn test_drop_table_blocked_in_read_only() {
    let handler = handler(true);
    let drop_request = DropTableRequest { table: "users".into() };

    let response = handler.drop_table(drop_request).await;
    assert!(response.is_err(), "drop_table should be blocked in read-only mode");
}

#[tokio::test]
async fn test_list_tables_returns_empty_for_no_match() {
    let handler = handler(true);
    let request = ReadQueryRequest {
        query: "SELECT name FROM sqlite_master WHERE type='table' AND name='nonexistent_xyz'".into(),
        cursor: None,
    };
    let result = handler.read_query(request).await.unwrap();
    let rows = &result.rows;
    assert!(rows.is_empty(), "query for nonexistent table should return empty array");
}

#[tokio::test]
async fn test_create_drop_table_with_spaces() {
    let handler = handler(false);

    let create = QueryRequest {
        query: "CREATE TABLE \"table with spaces\" (id INTEGER PRIMARY KEY, name TEXT)".into(),
    };
    handler.write_query(create).await.unwrap();

    // Use the enriched listTables in detailed mode to confirm the table is visible
    // with its metadata, exercising identifier handling for spaced names end-to-end.
    let response = handler
        .list_tables(ListTablesRequest {
            search: Some("table with spaces".into()),
            detailed: true,
            ..Default::default()
        })
        .await
        .expect("detailed listTables with spaced name");
    let entries = response.entries.as_detailed().expect("detailed mode");
    assert!(
        entries.contains_key("table with spaces"),
        "spaced-name table must surface as a keyed entry: {entries:?}"
    );

    let drop = DropTableRequest {
        table: "table with spaces".into(),
    };
    handler.drop_table(drop).await.unwrap();
}

async fn collect_all_paged(handler: &SqliteHandler) -> Vec<String> {
    let mut all = Vec::new();
    let mut cursor: Option<dbmcp_server::pagination::Cursor> = None;
    loop {
        let request = ListTablesRequest {
            cursor,
            ..Default::default()
        };
        let response = handler.list_tables(request).await.expect("list page");
        let page = response.entries.as_brief().expect("brief mode").to_vec();
        all.extend(page);
        match response.next_cursor {
            Some(c) => cursor = Some(c),
            None => break,
        }
    }
    all
}

#[tokio::test]
async fn test_list_tables_pagination_traverses_pages() {
    let handler_paged = handler_with_page_size(2);
    let handler_full = handler(true);

    let collected = collect_all_paged(&handler_paged).await;

    let single_page = handler_full
        .list_tables(ListTablesRequest::default())
        .await
        .expect("single page");
    let single_page_vec = single_page.entries.as_brief().expect("brief mode").to_vec();

    assert_eq!(
        collected, single_page_vec,
        "paged traversal must yield identical results (and ordering) to a single full page"
    );
    let unique: std::collections::HashSet<&String> = collected.iter().collect();
    assert_eq!(unique.len(), collected.len(), "no duplicates across pages");
}

#[tokio::test]
async fn test_list_tables_pagination_small_table_set_no_next_cursor() {
    let handler = handler(true);
    let response = handler.list_tables(ListTablesRequest::default()).await.unwrap();
    assert!(
        response.next_cursor.is_none(),
        "seeded fixture below default page_size must not emit nextCursor"
    );
}

#[tokio::test]
async fn test_list_tables_pagination_boundary_page_size_equals_total() {
    let handler_full = handler(true);
    let total = handler_full
        .list_tables(ListTablesRequest::default())
        .await
        .expect("discover total")
        .entries
        .len();
    let page_size = u16::try_from(total).expect("seed total fits in u16");

    let handler_boundary = handler_with_page_size(page_size);
    let response = handler_boundary
        .list_tables(ListTablesRequest::default())
        .await
        .unwrap();
    assert_eq!(
        response.entries.len(),
        total,
        "page_size equal to total must return everything on one page"
    );
    assert!(
        response.next_cursor.is_none(),
        "page_size equal to total must NOT emit nextCursor"
    );
}

#[tokio::test]
async fn test_list_tables_pagination_off_the_end_cursor_returns_empty_page() {
    use dbmcp_server::pagination::Cursor;

    let handler = handler(true);
    let request = ListTablesRequest {
        cursor: Some(Cursor { offset: 10_000 }),
        ..Default::default()
    };
    let response = handler.list_tables(request).await.unwrap();

    assert!(
        response.entries.is_empty(),
        "off-the-end cursor must return empty tables, got {:?}",
        response.entries
    );
    assert!(response.next_cursor.is_none(), "off-the-end must not emit nextCursor");
}

#[tokio::test]
async fn test_list_tables_respects_configured_page_size() {
    let handler = handler_with_page_size(2);
    let first = handler
        .list_tables(ListTablesRequest::default())
        .await
        .expect("first page");
    assert_eq!(first.entries.len(), 2, "configured page_size=2 must cap page 1");
    assert!(
        first.next_cursor.is_some(),
        "page 1 must emit nextCursor when total > page_size"
    );
}

#[tokio::test]
async fn test_list_tables_respects_configured_page_size_minimum() {
    let handler = handler_with_page_size(1);
    let first = handler
        .list_tables(ListTablesRequest::default())
        .await
        .expect("first page");
    assert_eq!(first.entries.len(), 1, "page_size=1 must return one table per page");
    assert!(first.next_cursor.is_some(), "page 1 must emit nextCursor");
}

#[tokio::test]
async fn test_list_tables_pagination_invalid_cursor_rejected_at_deserialize() {
    // Malformed cursors are rejected at serde deserialization time (before any
    // handler runs), which rmcp automatically maps to JSON-RPC `-32602`. Verify
    // the rejection here at the serde layer.
    use serde_json::json;

    let bad_cursors = [
        "!!!not-base64",
        "bm90LWpzb24",
        "eyJ4IjoxfQ",
        "eyJvIjowLCJ2Ijo5fQ",
        "eyJvIjotMSwidiI6MX0",
    ];

    for bad in bad_cursors {
        let err = serde_json::from_value::<ListTablesRequest>(json!({ "cursor": bad }))
            .expect_err(&format!("cursor {bad:?} should be rejected at deserialize time"));
        assert!(
            err.to_string().to_lowercase().contains("cursor")
                || err.to_string().to_lowercase().contains("base64")
                || err.to_string().to_lowercase().contains("malformed")
                || err.to_string().to_lowercase().contains("version"),
            "cursor {bad:?} error is not descriptive: {err}"
        );
    }
}

async fn collect_all_paged_read_query(handler: &SqliteHandler, query: &str) -> Vec<Value> {
    let mut all = Vec::new();
    let mut cursor: Option<dbmcp_server::pagination::Cursor> = None;
    loop {
        let request = ReadQueryRequest {
            query: query.into(),
            cursor,
        };
        let response = handler.read_query(request).await.expect("read_query page");
        all.extend(response.rows);
        match response.next_cursor {
            Some(c) => cursor = Some(c),
            None => break,
        }
    }
    all
}

#[tokio::test]
async fn test_read_query_pagination_traverses_pages() {
    let handler_paged = handler_with_page_size(2);
    let handler_full = handler(true);
    let query = "SELECT id FROM users ORDER BY id";

    let collected = collect_all_paged_read_query(&handler_paged, query).await;

    let single = handler_full
        .read_query(ReadQueryRequest {
            query: query.into(),
            cursor: None,
        })
        .await
        .expect("single page");
    assert_eq!(
        collected, single.rows,
        "paged traversal must yield identical rows (and ordering) to a single full page"
    );
    let ids: Vec<i64> = collected
        .iter()
        .map(|row| row["id"].as_i64().expect("id is integer"))
        .collect();
    assert_eq!(ids, vec![1, 2, 3], "seeded users should be ids 1..=3");
}

#[tokio::test]
async fn test_read_query_pagination_small_result_no_next_cursor() {
    let handler = handler_with_page_size(2);
    let response = handler
        .read_query(ReadQueryRequest {
            query: "SELECT id FROM users WHERE id = 1".into(),
            cursor: None,
        })
        .await
        .unwrap();
    assert!(
        response.next_cursor.is_none(),
        "single-row result must not emit nextCursor"
    );
    assert_eq!(response.rows.len(), 1);
}

#[tokio::test]
async fn test_read_query_pagination_empty_result_no_next_cursor() {
    let handler = handler_with_page_size(2);
    let response = handler
        .read_query(ReadQueryRequest {
            query: "SELECT id FROM users WHERE id = -1".into(),
            cursor: None,
        })
        .await
        .unwrap();
    assert!(&response.rows.is_empty());
    assert!(response.next_cursor.is_none());
}

#[tokio::test]
async fn test_read_query_pagination_preserves_inner_limit() {
    // Caller's inner LIMIT/OFFSET stays inside the subquery wrap.
    let handler = handler_with_page_size(2);
    let response = handler
        .read_query(ReadQueryRequest {
            query: "SELECT id FROM users ORDER BY id LIMIT 1 OFFSET 1".into(),
            cursor: None,
        })
        .await
        .unwrap();
    let rows = &response.rows;
    assert_eq!(rows.len(), 1);
    assert_eq!(
        rows[0]["id"].as_i64(),
        Some(2),
        "inner OFFSET 1 LIMIT 1 must return id=2"
    );
    assert!(response.next_cursor.is_none(), "inner 1-row result fits on one page");
}

#[tokio::test]
async fn test_read_query_pagination_off_the_end_cursor_returns_empty() {
    use dbmcp_server::pagination::Cursor;
    let handler = handler_with_page_size(2);
    let response = handler
        .read_query(ReadQueryRequest {
            query: "SELECT id FROM users ORDER BY id".into(),
            cursor: Some(Cursor { offset: 10_000 }),
        })
        .await
        .unwrap();
    assert!(&response.rows.is_empty());
    assert!(response.next_cursor.is_none());
}

#[tokio::test]
async fn test_read_query_pagination_survives_trailing_line_comment() {
    // A trailing `-- comment` in the caller's SQL must not swallow the
    // subquery-wrap's closing paren + LIMIT/OFFSET. Regression guard for
    // the newline-before-`)` fix in dbmcp_sql::pagination::with_limit_offset.
    let handler = handler_with_page_size(2);
    let response = handler
        .read_query(ReadQueryRequest {
            query: "SELECT id FROM users ORDER BY id -- fetch users".into(),
            cursor: None,
        })
        .await
        .expect("trailing comment must not break the subquery wrap");
    let rows = &response.rows;
    assert_eq!(
        rows.len(),
        2,
        "page 1 should cap at page_size=2 even with trailing comment"
    );
    assert!(response.next_cursor.is_some());
}

#[tokio::test]
async fn test_read_query_pagination_strips_trailing_semicolon() {
    let handler = handler_with_page_size(2);
    let response = handler
        .read_query(ReadQueryRequest {
            query: "SELECT id FROM users WHERE id = 1;".into(),
            cursor: None,
        })
        .await
        .unwrap();
    let rows = &response.rows;
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0]["id"].as_i64(), Some(1));
}

#[tokio::test]
async fn test_read_query_pagination_invalid_cursor_rejected_at_deserialize() {
    use serde_json::json;

    let bad_cursors = ["!!!not-base64", "bm90LWpzb24", "eyJ4IjoxfQ", "eyJvZmZzZXQiOi0xfQ"];

    for bad in bad_cursors {
        let err = serde_json::from_value::<ReadQueryRequest>(json!({
            "query": "SELECT 1",
            "cursor": bad,
        }))
        .expect_err(&format!("cursor {bad:?} should be rejected at deserialize time"));
        let msg = err.to_string().to_lowercase();
        assert!(
            msg.contains("cursor") || msg.contains("base64") || msg.contains("malformed"),
            "cursor {bad:?} error is not descriptive: {err}"
        );
    }
}

#[tokio::test]
async fn test_read_query_cursor_does_not_bypass_read_only() {
    use dbmcp_server::pagination::Cursor;
    let handler = handler_with_page_size(2);
    let result = handler
        .read_query(ReadQueryRequest {
            query: "DELETE FROM users WHERE id = 1".into(),
            cursor: Some(Cursor { offset: 0 }),
        })
        .await;
    assert!(
        result.is_err(),
        "DELETE must fail read-only check even with a valid cursor"
    );
}

#[tokio::test]
async fn test_read_query_non_select_single_page_with_cursor_ignored() {
    // EXPLAIN is classified as NonSelect; cursor must be ignored (no error,
    // no nextCursor, response identical to the no-cursor call).
    use dbmcp_server::pagination::Cursor;
    let handler = handler_with_page_size(2);

    let without_cursor = handler
        .read_query(ReadQueryRequest {
            query: "EXPLAIN SELECT 1".into(),
            cursor: None,
        })
        .await
        .expect("EXPLAIN without cursor should succeed");

    let with_cursor = handler
        .read_query(ReadQueryRequest {
            query: "EXPLAIN SELECT 1".into(),
            cursor: Some(Cursor { offset: 100 }),
        })
        .await
        .expect("EXPLAIN with cursor should succeed — cursor must be ignored");

    assert!(without_cursor.next_cursor.is_none());
    assert!(with_cursor.next_cursor.is_none());
    assert_eq!(
        without_cursor.rows, with_cursor.rows,
        "cursor must be silently ignored for non-SELECT statements"
    );
}

#[tokio::test]
async fn test_read_query_temporal_columns_preserved() {
    // Feature 038: SQLite stores temporal values as TEXT verbatim. This test
    // is a regression guard ensuring the existing TEXT decode path keeps
    // returning the exact stored text (no formatting applied by the server).
    let handler = handler(false);

    let response = handler
        .read_query(ReadQueryRequest {
            query: "SELECT date, time, timestamp FROM temporal WHERE id = 1".into(),
            cursor: None,
        })
        .await
        .expect("temporal SELECT should succeed");

    let arr = &response.rows;
    assert_eq!(arr.len(), 1, "temporal seeds exactly one row");
    assert_eq!(arr[0]["date"], "2026-04-20");
    assert_eq!(arr[0]["time"], "14:30:00");
    // Note the space (not T) between date and time — SQLite stores the
    // literal text we inserted; this is the existing behavior we are
    // explicitly asserting does not regress.
    assert_eq!(arr[0]["timestamp"], "2026-04-20 14:30:00");
}

#[tokio::test]
async fn test_list_views_returns_seeded_views() {
    let handler = handler(true);
    let response = handler
        .list_views(ListViewsRequest { cursor: None })
        .await
        .expect("list_views");

    let names = response.entries.as_brief().expect("brief mode");
    assert!(
        names.contains(&"active_users".to_string()),
        "expected seeded active_users view, got {names:?}"
    );
    assert!(
        names.contains(&"published_posts".to_string()),
        "expected seeded published_posts view, got {names:?}"
    );
}

#[tokio::test]
async fn test_list_views_excludes_base_tables() {
    let handler = handler(true);
    let response = handler
        .list_views(ListViewsRequest { cursor: None })
        .await
        .expect("list_views");

    let names = response.entries.as_brief().expect("brief mode");
    for table in ["users", "posts", "tags", "post_tags", "temporal"] {
        assert!(
            !names.contains(&table.to_string()),
            "base table `{table}` must not appear in listViews, got {names:?}"
        );
    }
}

#[tokio::test]
async fn test_list_views_pagination_traverses_pages() {
    let handler_paged = handler_with_page_size(1);
    let handler_full = handler(true);

    let mut all = Vec::new();
    let mut cursor: Option<dbmcp_server::pagination::Cursor> = None;
    loop {
        let response = handler_paged
            .list_views(ListViewsRequest { cursor })
            .await
            .expect("paged list_views");
        all.extend(response.entries.as_brief().expect("brief").iter().cloned());
        match response.next_cursor {
            Some(c) => cursor = Some(c),
            None => break,
        }
    }

    let single = handler_full
        .list_views(ListViewsRequest { cursor: None })
        .await
        .expect("single-page list_views");

    let single_names = single.entries.as_brief().expect("brief").to_vec();
    assert_eq!(all, single_names, "paginated traversal should equal single page");
}

#[tokio::test]
async fn test_list_views_works_in_read_only_mode() {
    let handler = handler(true);
    let response = handler
        .list_views(ListViewsRequest { cursor: None })
        .await
        .expect("list_views in read-only mode");

    assert!(
        !response.entries.as_brief().expect("brief").is_empty(),
        "read-only mode must still allow listViews"
    );
}

#[tokio::test]
async fn test_list_triggers_returns_seeded_triggers() {
    let handler = handler(true);
    let response = handler
        .list_triggers(ListTriggersRequest::default())
        .await
        .expect("list_triggers");

    let names = response.entries.as_brief().expect("brief mode");
    assert!(
        names.contains(&"users_before_insert".to_string()),
        "expected seeded users_before_insert trigger, got {names:?}"
    );
    assert!(
        names.contains(&"posts_before_update".to_string()),
        "expected seeded posts_before_update trigger, got {names:?}"
    );
}

#[tokio::test]
async fn test_list_triggers_works_in_read_only_mode() {
    let handler = handler(true);
    let response = handler
        .list_triggers(ListTriggersRequest::default())
        .await
        .expect("list_triggers in read-only mode");

    assert!(
        !response.entries.is_empty(),
        "read-only mode must still allow listTriggers"
    );
}

#[tokio::test]
async fn test_list_tables_search_filter_returns_only_matches() {
    let handler = handler(true);
    let response = handler
        .list_tables(ListTablesRequest {
            search: Some("post".into()),
            ..Default::default()
        })
        .await
        .expect("brief + search");
    let names = response.entries.as_brief().expect("brief").to_vec();
    assert_eq!(
        names,
        vec!["post_tags", "posts", "posts_fts"],
        "search 'post' must match exactly these three tables"
    );
}

#[tokio::test]
async fn test_list_tables_search_is_case_insensitive() {
    let handler = handler(true);
    let response = handler
        .list_tables(ListTablesRequest {
            search: Some("PoSt".into()),
            ..Default::default()
        })
        .await
        .expect("brief + search");
    let names = response.entries.as_brief().expect("brief").to_vec();
    assert_eq!(
        names,
        vec!["post_tags", "posts", "posts_fts"],
        "mixed-case 'PoSt' must yield the same set as 'post'"
    );
}

#[tokio::test]
async fn test_list_tables_search_no_match_returns_empty() {
    let handler = handler(true);
    let response = handler
        .list_tables(ListTablesRequest {
            search: Some("zzznosuch".into()),
            ..Default::default()
        })
        .await
        .expect("brief + search");
    assert!(response.entries.as_brief().expect("brief").is_empty());
    assert!(response.next_cursor.is_none());
}

#[tokio::test]
async fn test_list_tables_search_supports_like_wildcards() {
    let handler = handler(true);
    // `p_sts` matches `posts` (single-char `_`) but NOT `post_tags` (would need `p____tags`).
    let response = handler
        .list_tables(ListTablesRequest {
            search: Some("p_sts".into()),
            ..Default::default()
        })
        .await
        .expect("brief + search");
    let names = response.entries.as_brief().expect("brief").to_vec();
    assert!(
        names.iter().any(|n| n == "posts"),
        "'_' wildcard should match 'posts': {names:?}"
    );
    assert!(
        !names.iter().any(|n| n == "post_tags"),
        "'_' wildcard should NOT match 'post_tags': {names:?}"
    );
}

#[tokio::test]
async fn test_list_tables_search_empty_is_same_as_no_filter() {
    let handler = handler(true);
    let whitespace = handler
        .list_tables(ListTablesRequest {
            search: Some("   ".into()),
            ..Default::default()
        })
        .await
        .expect("whitespace search");
    let unfiltered = handler
        .list_tables(ListTablesRequest::default())
        .await
        .expect("no search");
    assert_eq!(
        whitespace.entries.as_brief().expect("brief"),
        unfiltered.entries.as_brief().expect("brief"),
        "whitespace-only search must be equivalent to no filter"
    );
}

#[tokio::test]
async fn test_list_tables_search_sql_meta_payloads_are_safe() {
    let handler = handler(true);
    // SQL-injection-flavoured payload via bind parameter — must not error, must not drop users.
    let response = handler
        .list_tables(ListTablesRequest {
            search: Some("'; DROP TABLE users; --".into()),
            ..Default::default()
        })
        .await
        .expect("bound payload must not error");
    assert!(response.entries.as_brief().expect("brief").is_empty());

    let follow_up = handler
        .list_tables(ListTablesRequest::default())
        .await
        .expect("users still present");
    assert!(
        follow_up
            .entries
            .as_brief()
            .expect("brief")
            .iter()
            .any(|n| n == "users"),
        "users table must survive any bind-based SQL meta payload"
    );
}

#[tokio::test]
async fn test_list_tables_search_paginates_filtered_results() {
    let handler = handler_with_page_size(1);
    let mut collected = Vec::new();
    let mut cursor: Option<dbmcp_server::pagination::Cursor> = None;
    loop {
        let response = handler
            .list_tables(ListTablesRequest {
                cursor,
                search: Some("post".into()),
                ..Default::default()
            })
            .await
            .expect("paged brief + search");
        collected.extend(response.entries.as_brief().expect("brief").to_vec());
        match response.next_cursor {
            Some(c) => cursor = Some(c),
            None => break,
        }
    }
    assert_eq!(
        collected,
        vec!["post_tags", "posts", "posts_fts"],
        "paginated filter must yield the full filtered set in order"
    );
}

async fn detailed_entries(handler: &SqliteHandler, search: &str) -> indexmap::IndexMap<String, Value> {
    let response = handler
        .list_tables(ListTablesRequest {
            search: Some(search.into()),
            detailed: true,
            ..Default::default()
        })
        .await
        .expect("detailed + search");
    response.entries.as_detailed().expect("detailed mode").clone()
}

#[tokio::test]
async fn test_list_tables_detailed_returns_full_metadata_for_posts() {
    let handler = handler(true);
    let entries = detailed_entries(&handler, "posts").await;
    let posts = entries.get("posts").expect("posts entry present");

    assert_eq!(posts["kind"], "TABLE");
    assert_eq!(posts["schema"], "main");
    assert!(posts["owner"].is_null(), "owner always null on SQLite");
    assert!(posts["comment"].is_null(), "comment always null on SQLite");

    // Columns: 5 with 1-based ordinalPosition.
    let columns = posts["columns"].as_array().expect("columns array");
    assert_eq!(columns.len(), 5, "posts has 5 columns: {columns:?}");
    let id_col = columns.iter().find(|c| c["name"] == "id").expect("id column present");
    assert_eq!(id_col["ordinalPosition"], 1, "ordinalPosition is 1-based (cid + 1)");
    assert!(id_col["comment"].is_null(), "column comment always null on SQLite");

    // Constraints: PK, FK, UNIQUE — no CHECK.
    let constraints = posts["constraints"].as_array().expect("constraints array");
    let types: Vec<&str> = constraints.iter().map(|c| c["type"].as_str().unwrap_or("")).collect();
    assert!(
        types.contains(&"PRIMARY KEY"),
        "PRIMARY KEY constraint expected: {types:?}"
    );
    assert!(
        types.contains(&"FOREIGN KEY"),
        "FOREIGN KEY constraint expected: {types:?}"
    );
    assert!(
        types.contains(&"UNIQUE"),
        "UNIQUE constraint expected from posts_user_title_uidx: {types:?}"
    );
    assert!(
        !types.contains(&"CHECK"),
        "CHECK constraints must never appear on SQLite: {types:?}"
    );

    // FK referencedTable/columns
    let fk = constraints
        .iter()
        .find(|c| c["type"] == "FOREIGN KEY")
        .expect("FK present");
    assert_eq!(fk["referencedTable"], "users");
    assert_eq!(fk["referencedColumns"], serde_json::json!(["id"]));

    // Indexes: two user-declared — posts_user_title_uidx (unique) and posts_published_idx.
    // posts.id is `INTEGER PRIMARY KEY` which is a rowid alias on SQLite, so there is no
    // separate autoindex row for it; primary-index synthesis is covered by the post_tags
    // autoindex test below.
    let indexes = posts["indexes"].as_array().expect("indexes array");
    let names: Vec<&str> = indexes.iter().map(|i| i["name"].as_str().unwrap_or("")).collect();
    assert!(
        names.contains(&"posts_user_title_uidx"),
        "unique index present: {names:?}"
    );
    assert!(
        names.contains(&"posts_published_idx"),
        "secondary index present: {names:?}"
    );
    let unique_idx = indexes.iter().find(|i| i["name"] == "posts_user_title_uidx").unwrap();
    assert_eq!(unique_idx["unique"], true);
    let secondary_idx = indexes.iter().find(|i| i["name"] == "posts_published_idx").unwrap();
    assert_eq!(secondary_idx["unique"], false);
    assert_eq!(secondary_idx["primary"], false);
}

#[tokio::test]
async fn test_list_tables_detailed_distinguishes_virtual_tables() {
    let handler = handler(true);
    let entries = detailed_entries(&handler, "fts").await;
    let fts = entries.get("posts_fts").expect("posts_fts entry present");
    assert_eq!(
        fts["kind"], "VIRTUAL_TABLE",
        "FTS5 virtual table must be tagged VIRTUAL_TABLE"
    );
    let columns = fts["columns"].as_array().expect("columns array");
    assert!(
        !columns.is_empty(),
        "virtual table columns must come from pragma_table_info"
    );
}

#[tokio::test]
async fn test_list_tables_detailed_handles_without_rowid_table() {
    let handler = handler(true);
    let entries = detailed_entries(&handler, "lookup_codes").await;
    let lookup = entries.get("lookup_codes").expect("lookup_codes entry present");
    let columns = lookup["columns"].as_array().expect("columns array");
    let first = columns.first().expect("at least one column");
    assert_eq!(first["name"], "code");
    assert_eq!(first["ordinalPosition"], 1);
    let constraints = lookup["constraints"].as_array().expect("constraints array");
    let pk = constraints
        .iter()
        .find(|c| c["type"] == "PRIMARY KEY")
        .expect("PK present on WITHOUT ROWID table");
    assert_eq!(pk["columns"], serde_json::json!(["code"]));
}

#[tokio::test]
async fn test_list_tables_detailed_synthesises_autoindex_definition() {
    // `post_tags` has a composite PRIMARY KEY, which is the one case where SQLite
    // creates a real `sqlite_autoindex_*` row in `pragma_index_list` (INTEGER PRIMARY
    // KEY tables reuse the rowid and get no autoindex). This is the right fixture
    // for exercising R4 synthesis.
    let handler = handler(true);
    let entries = detailed_entries(&handler, "post_tags").await;
    let table = entries.get("post_tags").expect("post_tags entry present");
    let indexes = table["indexes"].as_array().expect("indexes array");
    let autoindex = indexes
        .iter()
        .find(|i| i["name"].as_str().is_some_and(|n| n.starts_with("sqlite_autoindex_")))
        .expect("composite-PK table must have a sqlite_autoindex_ entry");
    assert_eq!(autoindex["primary"], true, "PK autoindex must be primary");
    assert_eq!(autoindex["unique"], true, "PK autoindex must be unique");
    let definition = autoindex["definition"].as_str().expect("definition string");
    assert!(
        !definition.is_empty(),
        "autoindex definition must be synthesised, never null/empty"
    );
    let upper = definition.to_uppercase();
    assert!(
        upper.contains("CREATE") && upper.contains("INDEX"),
        "synthesised definition should look like a CREATE [UNIQUE] INDEX statement: {definition}"
    );
    // Synthesised PK index lists (post_id, tag_id) — confirm both columns appear.
    assert!(
        definition.contains("post_id") && definition.contains("tag_id"),
        "synthesised autoindex definition must reference the composite PK columns: {definition}"
    );
}

#[tokio::test]
async fn test_list_tables_detailed_reports_trigger() {
    let handler = handler(true);
    let entries = detailed_entries(&handler, "posts").await;
    let posts = entries.get("posts").expect("posts entry present");
    let triggers = posts["triggers"].as_array().expect("triggers array");
    let trigger = triggers
        .iter()
        .find(|t| t["name"] == "posts_before_update")
        .expect("seeded posts_before_update trigger must surface");
    assert!(
        trigger["definition"]
            .as_str()
            .unwrap_or("")
            .to_uppercase()
            .contains("CREATE TRIGGER"),
        "trigger definition should be the raw CREATE TRIGGER text"
    );
    assert_eq!(trigger["enabled"], true, "SQLite triggers always report enabled: true");
}

#[tokio::test]
async fn test_list_tables_detailed_paginates() {
    let handler = handler_with_page_size(1);
    let mut collected = Vec::new();
    let mut cursor: Option<dbmcp_server::pagination::Cursor> = None;
    loop {
        let response = handler
            .list_tables(ListTablesRequest {
                cursor,
                search: Some("post".into()),
                detailed: true,
            })
            .await
            .expect("paged detailed");
        let page = response.entries.as_detailed().expect("detailed");
        assert!(page.len() <= 1, "page_size=1 caps to 1 per page");
        collected.extend(page.keys().cloned());
        match response.next_cursor {
            Some(c) => cursor = Some(c),
            None => break,
        }
    }
    assert_eq!(
        collected,
        vec!["post_tags", "posts", "posts_fts"],
        "detailed pagination must yield filtered sequence"
    );
}

#[tokio::test]
async fn test_list_tables_detailed_with_search_only_fetches_filtered_subset() {
    let handler = handler(true);
    let entries = detailed_entries(&handler, "post").await;
    let names: Vec<_> = entries.keys().cloned().collect();
    assert_eq!(
        names,
        vec!["post_tags", "posts", "posts_fts"],
        "detailed + search must yield only matching tables"
    );
}

#[tokio::test]
async fn test_list_tables_detailed_values_have_no_redundant_name_field() {
    let handler = handler(true);
    let entries = detailed_entries(&handler, "post").await;
    assert!(!entries.is_empty(), "fixture must yield at least one match for 'post'");
    for (key, value) in &entries {
        assert!(value.is_object(), "value for {key} must be a JSON object: {value}");
        assert!(
            value.get("name").is_none(),
            "FR-002 violated: value for {key} still carries 'name': {value}"
        );
    }
}

#[tokio::test]
async fn test_list_tables_detailed_key_order_matches_brief_order() {
    let handler = handler(true);
    let brief = handler
        .list_tables(ListTablesRequest {
            search: Some("post".into()),
            detailed: false,
            ..Default::default()
        })
        .await
        .expect("brief + search");
    let detailed = detailed_entries(&handler, "post").await;
    let brief_names: Vec<String> = brief.entries.as_brief().expect("brief mode").to_vec();
    let detailed_keys: Vec<String> = detailed.keys().cloned().collect();
    assert_eq!(
        brief_names, detailed_keys,
        "detailed key order must match brief string order — FR-010"
    );
}

// =====================================================================
// listTriggers — feature 054 (search + detailed)
// =====================================================================

const AUDIT_TRIGGERS: &[&str] = &[
    "posts_audit_after_insert",
    "users_audit_after_delete",
    "users_audit_after_insert",
    "users_audit_after_update",
];

#[tokio::test]
async fn test_list_triggers_search_filter_returns_only_matches() {
    let handler = handler(true);
    let response = handler
        .list_triggers(ListTriggersRequest {
            search: Some("audit".into()),
            ..Default::default()
        })
        .await
        .expect("brief + search");
    let names = response.entries.as_brief().expect("brief mode").to_vec();
    let expected: Vec<String> = AUDIT_TRIGGERS.iter().map(|s| (*s).to_string()).collect();
    assert_eq!(
        names, expected,
        "search 'audit' must match exactly the four *_audit_* triggers, alphabetically"
    );
}

#[tokio::test]
async fn test_list_triggers_search_is_case_insensitive() {
    let handler = handler(true);
    let lower = handler
        .list_triggers(ListTriggersRequest {
            search: Some("audit".into()),
            ..Default::default()
        })
        .await
        .expect("lower");
    let upper = handler
        .list_triggers(ListTriggersRequest {
            search: Some("AUDIT".into()),
            ..Default::default()
        })
        .await
        .expect("upper");
    assert_eq!(
        lower.entries.as_brief().expect("brief"),
        upper.entries.as_brief().expect("brief"),
        "case-insensitive (SQLite `LIKE` is ASCII-case-insensitive by default)"
    );
}

#[tokio::test]
async fn test_list_triggers_search_no_match_returns_empty() {
    let handler = handler(true);
    let response = handler
        .list_triggers(ListTriggersRequest {
            search: Some("nonexistent_xyz".into()),
            ..Default::default()
        })
        .await
        .expect("brief + search");
    assert!(response.entries.as_brief().expect("brief").is_empty());
    assert!(response.next_cursor.is_none());
}

#[tokio::test]
async fn test_list_triggers_search_supports_wildcard_semantics() {
    let handler = handler(true);
    // `%audit%` is `LIKE`-equivalent to plain `audit` substring — both must yield identical sets.
    let bare = handler
        .list_triggers(ListTriggersRequest {
            search: Some("audit".into()),
            ..Default::default()
        })
        .await
        .expect("bare");
    let wildcards = handler
        .list_triggers(ListTriggersRequest {
            search: Some("%audit%".into()),
            ..Default::default()
        })
        .await
        .expect("wildcards");
    assert_eq!(
        bare.entries.as_brief().expect("brief"),
        wildcards.entries.as_brief().expect("brief"),
        "%/_ retain LIKE wildcard semantics consistent with listTables"
    );
}

#[tokio::test]
async fn test_list_triggers_search_sql_meta_payloads_are_safe() {
    let handler = handler(true);
    for payload in ["'", ";", "--", "\"", "/* */", "'; DROP TABLE users; --"] {
        let response = handler
            .list_triggers(ListTriggersRequest {
                search: Some(payload.into()),
                ..Default::default()
            })
            .await
            .unwrap_or_else(|e| panic!("payload {payload:?} must not error: {e:?}"));
        // Bound payload — no SQL parse error, possibly empty result.
        let _ = response.entries.as_brief().expect("brief").to_vec();
    }
    // Users table must survive every adversarial payload.
    let follow_up = handler
        .list_tables(ListTablesRequest::default())
        .await
        .expect("users table check");
    assert!(
        follow_up
            .entries
            .as_brief()
            .expect("brief")
            .iter()
            .any(|n| n == "users"),
        "users table must survive every adversarial search payload"
    );
}

#[tokio::test]
async fn test_list_triggers_search_paginates_filtered_results() {
    let handler = handler_with_page_size(2);
    let mut collected: Vec<String> = Vec::new();
    let mut cursor: Option<dbmcp_server::pagination::Cursor> = None;
    loop {
        let response = handler
            .list_triggers(ListTriggersRequest {
                cursor,
                search: Some("audit".into()),
                ..Default::default()
            })
            .await
            .expect("paged brief + search");
        let page = response.entries.as_brief().expect("brief").to_vec();
        assert!(page.len() <= 2, "page exceeds page_size=2: {page:?}");
        collected.extend(page);
        match response.next_cursor {
            Some(c) => cursor = Some(c),
            None => break,
        }
    }
    let expected: Vec<String> = AUDIT_TRIGGERS.iter().map(|s| (*s).to_string()).collect();
    assert_eq!(
        collected, expected,
        "paginated audit-filter must yield the full set in alphabetical order"
    );
}

async fn detailed_trigger_entries(handler: &SqliteHandler, search: &str) -> indexmap::IndexMap<String, Value> {
    let response = handler
        .list_triggers(ListTriggersRequest {
            search: Some(search.into()),
            detailed: true,
            ..Default::default()
        })
        .await
        .expect("detailed + search");
    response.entries.as_detailed().expect("detailed mode").clone()
}

#[tokio::test]
async fn test_list_triggers_detailed_returns_keyed_object_with_three_fields() {
    let handler = handler(true);
    let entries = detailed_trigger_entries(&handler, "users_audit_after_insert").await;
    assert_eq!(entries.len(), 1, "exactly one match for the exact name");
    let entry = entries
        .get("users_audit_after_insert")
        .expect("entry keyed by trigger name");
    let obj = entry.as_object().expect("entry is an object");
    let keys: Vec<&str> = obj.keys().map(String::as_str).collect();
    assert_eq!(keys.len(), 3, "detailed entry has exactly three fields, got {keys:?}");
    assert_eq!(entry["schema"], "main", "schema literal");
    assert_eq!(entry["table"], "users", "table = tbl_name");
    let definition = entry["definition"].as_str().expect("definition is a string");
    assert!(
        definition.starts_with("CREATE TRIGGER"),
        "definition must start with the original CREATE TRIGGER text: {definition:?}"
    );
    assert!(
        definition.contains("AFTER INSERT ON users"),
        "definition must reflect the seeded body: {definition:?}"
    );
}

#[tokio::test]
async fn test_list_triggers_detailed_omits_unrelated_keys() {
    let handler = handler(true);
    let entries = detailed_trigger_entries(&handler, "audit").await;
    assert!(!entries.is_empty(), "audit search must match");
    for (name, entry) in &entries {
        let obj = entry.as_object().expect("entry object");
        for forbidden in [
            "timing",
            "events",
            "activationLevel",
            "status",
            "functionName",
            "sqlMode",
            "characterSetClient",
            "collationConnection",
            "databaseCollation",
            "created",
            "name",
        ] {
            assert!(
                !obj.contains_key(forbidden),
                "trigger {name:?} must not carry forbidden field {forbidden:?} (per Clarification Q1 / FR-005)"
            );
        }
    }
}

#[tokio::test]
async fn test_list_triggers_detailed_instead_of_view_carries_view_name_in_table() {
    let handler = handler(true);
    let entries = detailed_trigger_entries(&handler, "published_posts_instead_of_delete").await;
    let entry = entries
        .get("published_posts_instead_of_delete")
        .expect("INSTEAD OF trigger entry");
    assert_eq!(
        entry["table"], "published_posts",
        "INSTEAD OF triggers attach to a view; `table` reports the view name"
    );
    let definition = entry["definition"].as_str().expect("definition string");
    assert!(
        definition.contains("INSTEAD OF DELETE ON published_posts"),
        "definition must reflect the INSTEAD OF clause: {definition:?}"
    );
}

#[tokio::test]
async fn test_list_triggers_detailed_definition_round_trips_multiline_body() {
    let handler = handler(true);
    let entries = detailed_trigger_entries(&handler, "users_audit_after_insert").await;
    let entry = entries
        .get("users_audit_after_insert")
        .expect("multi-statement trigger entry");
    let definition = entry["definition"].as_str().expect("definition string");
    assert!(
        definition.contains("a note\nspans two lines"),
        "definition must preserve the seeded literal newline + quote characters byte-for-byte: {definition:?}"
    );
}

#[tokio::test]
async fn test_list_triggers_brief_mode_wire_shape_unchanged() {
    let handler = handler(true);
    let response = handler
        .list_triggers(ListTriggersRequest::default())
        .await
        .expect("brief default");
    let names = response.entries.as_brief().expect("brief mode");
    assert!(
        names.iter().all(|n| !n.is_empty()),
        "brief response is a flat array of bare strings: {names:?}"
    );
    // Existing seeded triggers from before this feature must still be listed.
    assert!(names.contains(&"users_before_insert".to_string()));
    assert!(names.contains(&"posts_before_update".to_string()));
}

#[tokio::test]
async fn test_list_triggers_detailed_search_narrows_payload() {
    let handler = handler(true);
    let entries = detailed_trigger_entries(&handler, "audit").await;
    for name in entries.keys() {
        assert!(
            name.contains("audit"),
            "search 'audit' + detailed must not surface unrelated triggers: got {name:?}"
        );
    }
    assert!(
        !entries.contains_key("users_before_insert"),
        "non-audit triggers must not appear in detailed audit-filtered response"
    );
    assert!(
        !entries.contains_key("posts_before_update"),
        "non-audit triggers must not appear in detailed audit-filtered response"
    );
}

#[tokio::test]
async fn test_list_triggers_detailed_paginates() {
    let handler = handler_with_page_size(2);
    let mut collected: Vec<String> = Vec::new();
    let mut cursor: Option<dbmcp_server::pagination::Cursor> = None;
    loop {
        let response = handler
            .list_triggers(ListTriggersRequest {
                cursor,
                search: Some("audit".into()),
                detailed: true,
            })
            .await
            .expect("paged detailed + search");
        let page = response.entries.as_detailed().expect("detailed mode");
        assert!(page.len() <= 2, "page exceeds page_size=2: {page:?}");
        collected.extend(page.keys().cloned());
        match response.next_cursor {
            Some(c) => cursor = Some(c),
            None => break,
        }
    }
    let expected: Vec<String> = AUDIT_TRIGGERS.iter().map(|s| (*s).to_string()).collect();
    assert_eq!(
        collected, expected,
        "paginated detailed audit-filter must yield the full set in alphabetical order"
    );
}

#[tokio::test]
async fn test_list_triggers_detailed_omits_null_sql_rows_brief_lists_them() {
    let handler = handler(false);

    // Seed a synthetic NULL-sql trigger row using PRAGMA writable_schema.
    // Scoped to this test only; reset PRAGMA back to OFF immediately.
    handler
        .write_query(QueryRequest {
            query: "PRAGMA writable_schema = ON".into(),
        })
        .await
        .expect("writable_schema on");
    handler
        .write_query(QueryRequest {
            query: "INSERT INTO sqlite_schema (type, name, tbl_name, rootpage, sql) \
                    VALUES ('trigger', 'orphan_trigger', 'users', 0, NULL)"
                .into(),
        })
        .await
        .expect("synthetic NULL-sql row");
    handler
        .write_query(QueryRequest {
            query: "PRAGMA writable_schema = OFF".into(),
        })
        .await
        .expect("writable_schema off");

    let brief = handler
        .list_triggers(ListTriggersRequest {
            search: Some("orphan".into()),
            ..Default::default()
        })
        .await
        .expect("brief search");
    let names = brief.entries.as_brief().expect("brief").to_vec();
    assert_eq!(
        names,
        vec!["orphan_trigger"],
        "brief mode lists NULL-sql triggers by name"
    );

    let detailed = handler
        .list_triggers(ListTriggersRequest {
            search: Some("orphan".into()),
            detailed: true,
            ..Default::default()
        })
        .await
        .expect("detailed search");
    let keyed = detailed.entries.as_detailed().expect("detailed");
    assert!(
        keyed.is_empty(),
        "detailed mode silently omits NULL-sql triggers (FR-018 / Clarification Q2): {keyed:?}"
    );

    // Cleanup.
    handler
        .write_query(QueryRequest {
            query: "PRAGMA writable_schema = ON".into(),
        })
        .await
        .expect("writable_schema on");
    handler
        .write_query(QueryRequest {
            query: "DELETE FROM sqlite_schema WHERE name = 'orphan_trigger'".into(),
        })
        .await
        .expect("cleanup synthetic row");
    handler
        .write_query(QueryRequest {
            query: "PRAGMA writable_schema = OFF".into(),
        })
        .await
        .expect("writable_schema off");
}

fn handler_with_redaction(redact_pii: bool) -> SqliteHandler {
    SqliteHandler::new(&Config {
        database: base_db_config(false),
        http: None,
        pii: PiiConfig {
            enabled: redact_pii,
            operator: PiiOperator::Replace,
            ..PiiConfig::default()
        },
    })
    .expect("handler builds")
}

fn handler_with_operator(operator: PiiOperator) -> SqliteHandler {
    SqliteHandler::new(&Config {
        database: base_db_config(false),
        http: None,
        pii: PiiConfig {
            enabled: true,
            operator,
            ..PiiConfig::default()
        },
    })
    .expect("handler builds")
}

#[tokio::test]
async fn read_query_redacts_email_when_enabled() {
    let handler = handler_with_redaction(true);
    let select = ReadQueryRequest {
        query: "SELECT 'ping me at jane.doe@example.com' AS msg".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    assert_eq!(rows.rows.len(), 1);
    assert_eq!(rows.rows[0]["msg"], "ping me at <EMAIL_ADDRESS>");
}

#[tokio::test]
async fn read_query_unchanged_when_disabled() {
    let handler = handler_with_redaction(false);
    let select = ReadQueryRequest {
        query: "SELECT 'ping me at jane.doe@example.com' AS msg".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    assert_eq!(rows.rows[0]["msg"], "ping me at jane.doe@example.com");
}

#[tokio::test]
async fn read_query_does_not_redact_non_string_values() {
    let handler = handler_with_redaction(true);
    let select = ReadQueryRequest {
        query: "SELECT 42 AS n, 1 AS flag, NULL AS missing".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    assert_eq!(rows.rows[0]["n"], 42);
    assert_eq!(rows.rows[0]["flag"], 1);
    assert!(rows.rows[0]["missing"].is_null());
}

#[tokio::test]
async fn read_query_does_not_redact_column_names() {
    let handler = handler_with_redaction(true);
    let select = ReadQueryRequest {
        query: "SELECT 1 AS \"user@example.com\"".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    let obj = rows.rows[0].as_object().expect("row object");
    assert!(
        obj.contains_key("user@example.com"),
        "column name preserved verbatim: {obj:?}"
    );
}

#[tokio::test]
async fn write_query_redacts_when_enabled() {
    let handler = handler_with_redaction(true);
    let insert = QueryRequest {
        query: "INSERT INTO users (name, email) VALUES ('PIIWrite', 'piiwrite@example.com')".into(),
    };
    handler.write_query(insert).await.unwrap();

    let select = ReadQueryRequest {
        query: "SELECT email FROM users WHERE name = 'PIIWrite'".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    assert_eq!(rows.rows[0]["email"], "<EMAIL_ADDRESS>");

    let cleanup = QueryRequest {
        query: "DELETE FROM users WHERE name = 'PIIWrite'".into(),
    };
    handler_with_redaction(false).write_query(cleanup).await.unwrap();
}

#[tokio::test]
async fn explain_query_redacts_when_enabled() {
    let handler = handler_with_redaction(true);
    let explain = ExplainQueryRequest {
        query: "SELECT 'ping me at jane.doe@example.com'".into(),
    };
    let rows = handler.explain_query(explain).await.unwrap();
    // SQLite EXPLAIN QUERY PLAN may not embed the literal — assert it is at
    // least not leaked when the analyzer would have caught it.
    let serialized = serde_json::to_string(&rows.rows).unwrap();
    assert!(
        !serialized.contains("jane.doe@example.com"),
        "raw email leaked into EXPLAIN output: {serialized}"
    );
}

#[tokio::test]
async fn read_query_mask_operator_replaces_with_asterisks() {
    let handler = handler_with_operator(PiiOperator::Mask);
    let select = ReadQueryRequest {
        query: "SELECT 'jane.doe@example.com' AS msg".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    let out = rows.rows[0]["msg"].as_str().unwrap();
    assert_eq!(out.len(), "jane.doe@example.com".len(), "mask preserves length");
    assert!(out.chars().all(|c| c == '*'), "mask must use '*': {out}");
}

#[tokio::test]
async fn read_query_redact_operator_returns_empty_span() {
    let handler = handler_with_operator(PiiOperator::Redact);
    let select = ReadQueryRequest {
        query: "SELECT 'jane.doe@example.com' AS msg".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    assert_eq!(rows.rows[0]["msg"], "");
}

#[tokio::test]
async fn read_query_hash_operator_emits_stable_digest() {
    let handler = handler_with_operator(PiiOperator::Hash);
    let select = ReadQueryRequest {
        query: "SELECT 'jane.doe@example.com' AS a, 'jane.doe@example.com' AS b".into(),
        cursor: None,
    };
    let rows = handler.read_query(select).await.unwrap();
    let a = rows.rows[0]["a"].as_str().unwrap();
    let b = rows.rows[0]["b"].as_str().unwrap();
    assert_eq!(a, b, "same input must hash to same digest");
    assert_ne!(a, "jane.doe@example.com");
    assert_eq!(a.len(), 64, "SHA-256 hex digest is 64 chars: {a}");
    assert!(a.chars().all(|c| c.is_ascii_hexdigit()), "digest must be hex: {a}");
}