hyperdb-mcp 0.2.1

// Copyright (c) 2026, Salesforce, Inc. All rights reserved.
// SPDX-License-Identifier: Apache-2.0 OR MIT

//! Tests that verify ingest operations are transactional: a failure
//! midway through a batch leaves zero rows behind.

mod common;

use common::TestEngine;
use hyperdb_mcp::engine::Engine;
use hyperdb_mcp::ingest::{ingest_json, IngestOptions};
use serde_json::Value;

/// Run a SELECT and return the rows. A known pre-existing wire-protocol
/// desync in hyperdb-api occasionally returns an empty first response right
/// after a Hyper-level error (e.g. a failed INSERT inside a transaction).
/// Retrying once is enough to flush the stale state.
fn query_resilient(engine: &Engine, sql: &str) -> Vec<Value> {
    let first = engine.execute_query_to_json(sql).unwrap();
    if first.is_empty() {
        engine.execute_query_to_json(sql).unwrap()
    } else {
        first
    }
}

/// Sanity check: the transaction helper itself rolls back correctly on
/// error. Inserts one row successfully, then triggers an error, then
/// verifies the row is gone.
#[test]
fn execute_in_transaction_rolls_back_on_error() {
    use hyperdb_mcp::error::{ErrorCode, McpError};
    let te = TestEngine::new_ephemeral();
    te.engine
        .execute_command("CREATE TABLE direct (v INT)")
        .unwrap();

    let result: Result<(), McpError> = te.engine.execute_in_transaction(|engine| {
        engine.execute_command("INSERT INTO direct VALUES (1)")?;
        engine.execute_command("INSERT INTO direct VALUES (2)")?;
        Err(McpError::new(ErrorCode::InternalError, "simulated failure"))
    });
    assert!(result.is_err());

    let rows = te
        .engine
        .execute_query_to_json("SELECT COUNT(*) as cnt FROM direct")
        .unwrap();
    let count = rows[0]["cnt"].as_i64().unwrap();
    assert_eq!(count, 0, "rollback should leave zero rows");
}

/// Sanity check: successful commits stick.
#[test]
fn execute_in_transaction_commits_on_success() {
    let te = TestEngine::new_ephemeral();
    te.engine
        .execute_command("CREATE TABLE direct (v INT)")
        .unwrap();

    te.engine
        .execute_in_transaction(|engine| {
            engine.execute_command("INSERT INTO direct VALUES (10)")?;
            engine.execute_command("INSERT INTO direct VALUES (20)")?;
            Ok(())
        })
        .unwrap();

    let rows = te
        .engine
        .execute_query_to_json("SELECT COUNT(*) as cnt FROM direct")
        .unwrap();
    let count = rows[0]["cnt"].as_i64().unwrap();
    assert_eq!(count, 2);
}

/// A panic inside the transaction closure (e.g. an unwrap on None, array
/// indexing OOB, arithmetic overflow) must not leave an open transaction
/// on the connection. Without the `catch_unwind` guard in
/// `execute_in_transaction`, the next operation would hit "transaction
/// already in progress" and the engine would be wedged until restart.
#[test]
fn execute_in_transaction_rolls_back_on_panic() {
    let te = TestEngine::new_ephemeral();
    te.engine
        .execute_command("CREATE TABLE direct (v INT)")
        .unwrap();

    // Trigger a panic inside the closure. `catch_unwind` at the
    // std::panic boundary lets the test assert the engine is still
    // usable afterwards without aborting the whole test binary.
    let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        te.engine.execute_in_transaction::<_, ()>(|engine| {
            engine.execute_command("INSERT INTO direct VALUES (1)")?;
            // Simulate a programmer error mid-transaction — any panic
            // will do. Using `panic!` directly keeps clippy from
            // second-guessing a synthetic `.unwrap()` on a literal.
            panic!("simulated closure bug");
        })
    }));
    assert!(outcome.is_err(), "panic should propagate out");

    // Engine must not be wedged — a fresh INSERT should succeed, and
    // the row from *before* the panic must have been rolled back.
    te.engine
        .execute_command("INSERT INTO direct VALUES (42)")
        .expect("engine must still be usable after panic inside tx");

    let rows = query_resilient(&te.engine, "SELECT v FROM direct ORDER BY v");
    assert_eq!(rows.len(), 1, "only the post-panic row should survive");
    assert_eq!(rows[0]["v"].as_i64().unwrap(), 42);
}

/// A failing JSON ingest in append mode must roll back all prior rows from
/// the same call. The target table is pre-created with a NOT NULL constraint;
/// our ingest JSON has three rows where the third violates the constraint
/// (missing the non-null key, which our INSERT emits as NULL).
#[test]
fn failed_json_ingest_rolls_back_partial_inserts() {
    let te = TestEngine::new_ephemeral();

    te.engine
        .execute_command("CREATE TABLE t (id INT NOT NULL, name TEXT)")
        .unwrap();

    let data = r#"[
        {"id": 1, "name": "Alice"},
        {"id": 2, "name": "Bob"},
        {"name": "NoId"}
    ]"#;

    let opts = IngestOptions {
        table: "t".into(),
        mode: "append".into(),
        schema_override: None,
        merge_key: None,
        target_db: None,
    };

    let result = ingest_json(&te.engine, data, &opts);
    assert!(result.is_err(), "ingest should fail on NOT NULL violation");

    // Crucially: the first two rows must have been rolled back. If
    // transactions were missing, we'd see count == 2 here.
    let rows = query_resilient(&te.engine, "SELECT COUNT(*) as cnt FROM t");
    let count = rows[0]["cnt"].as_i64().expect("count should be int");
    assert_eq!(
        count, 0,
        "expected rollback to leave zero rows, but found {count}"
    );
}

/// A successful ingest in replace mode commits both the DROP TABLE and all
/// INSERTs atomically — no observable intermediate state.
#[test]
fn successful_replace_commits_atomically() {
    let te = TestEngine::new_ephemeral();

    // Pre-populate the table with data that a replace-mode ingest should overwrite.
    te.engine
        .execute_command("CREATE TABLE t (id INT, name TEXT)")
        .unwrap();
    te.engine
        .execute_command("INSERT INTO t VALUES (99, 'Old')")
        .unwrap();

    let data = r#"[
        {"id": 1, "name": "Alice"},
        {"id": 2, "name": "Bob"}
    ]"#;
    let opts = IngestOptions {
        table: "t".into(),
        mode: "replace".into(),
        schema_override: None,
        merge_key: None,
        target_db: None,
    };
    let result = ingest_json(&te.engine, data, &opts).unwrap();
    assert_eq!(result.rows, 2);

    let rows = query_resilient(&te.engine, "SELECT COUNT(*) as cnt FROM t");
    assert_eq!(rows[0]["cnt"].as_i64().unwrap(), 2);

    // The old row is gone (replace mode drops and recreates).
    let rows = query_resilient(&te.engine, "SELECT COUNT(*) as cnt FROM t WHERE id = 99");
    assert_eq!(rows[0]["cnt"].as_i64().unwrap(), 0);
}

/// Replace-mode ingest with a mid-flight INSERT failure leaves the new
/// (empty) table behind, not partial rows.
///
/// Note: Hyper treats DDL (DROP TABLE, CREATE TABLE) as auto-committed
/// even inside a transaction, so replace-mode ingest cannot restore the
/// original table if it fails. The DDL is already durable by the time
/// the INSERTs run. What we can and do guarantee is that partial INSERTs
/// are rolled back, so the post-failure table is empty rather than
/// partially populated.
#[test]
fn failed_replace_leaves_empty_table_not_partial() {
    let te = TestEngine::new_ephemeral();

    te.engine
        .execute_command("CREATE TABLE t (id INT, name TEXT)")
        .unwrap();
    te.engine
        .execute_command("INSERT INTO t VALUES (100, 'Original')")
        .unwrap();

    // Override forces id to INT. Second row's string value will fail to
    // INSERT because it's sent as a quoted string and the implicit cast
    // 'bogus' -> INT fails at the server.
    let data = r#"[
        {"id": 1},
        {"id": "bogus"}
    ]"#;
    let mut override_map = serde_json::Map::new();
    override_map.insert("id".into(), serde_json::Value::String("INT".into()));
    let opts = IngestOptions {
        table: "t".into(),
        mode: "replace".into(),
        schema_override: Some(override_map),
        merge_key: None,
        target_db: None,
    };

    let result = ingest_json(&te.engine, data, &opts);
    assert!(result.is_err(), "ingest should fail on type cast error");

    // The pre-failure INSERT was rolled back — the table exists but has
    // zero rows, not one. (The original 'Original' row is gone due to
    // auto-committed DDL, so we can't get it back either way.)
    let rows = query_resilient(&te.engine, "SELECT COUNT(*) as cnt FROM t");
    let count = rows[0]["cnt"].as_i64().expect("count should be int");
    assert_eq!(
        count, 0,
        "expected partial INSERTs to be rolled back, leaving 0 rows"
    );
}