dbcli 0.1.0

//! MySQL implementation of `execute_raw_sql`.
//!
//! Executes one or more SQL statements against a MySQL database and returns
//! a [`Vec<SqlResult>`] — one entry per statement.

use futures::TryStreamExt;
use sqlx::mysql::{MySqlPool, MySqlRow};
use sqlx::Either;
use crate::SqlResult;

/// Execute one or more SQL statements against a MySQL database.
///
/// Supports multiple statements separated by `;`.
/// Each statement that produces rows (SELECT, SHOW, EXPLAIN, etc.) is automatically
/// converted to JSON via [`crate::to_json::mysql::to_json`] and returned as a
/// [`SqlResult::Query`]. Statements that do not produce rows return a
/// [`SqlResult::Execute`] with the count of affected rows.
///
/// # Arguments
///
/// * `pool` - MySQL connection pool
/// * `sql`  - SQL string; may contain multiple statements separated by `;`
///
/// # Returns
///
/// A [`Vec<SqlResult>`] with one entry per statement result, in execution order.
///
/// # Multi-statement Support
///
/// MySQL requires the `CLIENT_MULTI_STATEMENTS` flag to be enabled for multi-statement
/// execution. With `sqlx`, either append `?multi_statements=on` to your connection URL
/// or call [`sqlx::mysql::MySqlConnectOptions::multi_statements`]`(true)` when building
/// the pool options.
///
/// ```text
/// mysql://user:pass@host/db?multi_statements=on
/// ```
///
/// # Example
///
/// ```rust,no_run
/// # use sqlx::mysql::MySqlPool;
/// # async fn run(pool: &MySqlPool) -> anyhow::Result<()> {
/// let results = dbcli::execute::mysql::execute_raw_sql(
///     pool,
///     "INSERT INTO users(name) VALUES('alice'); SELECT * FROM users;",
/// ).await?;
///
/// for result in &results {
///     println!("{}", serde_json::to_string_pretty(result)?);
/// }
/// # Ok(())
/// # }
/// ```
pub async fn execute_raw_sql(
    pool: &MySqlPool,
    sql: &str,
) -> anyhow::Result<Vec<SqlResult>> {
    let mut stream = sqlx::raw_sql(sql).fetch_many(pool);
    let mut results: Vec<SqlResult> = Vec::new();
    let mut current_rows: Vec<MySqlRow> = Vec::new();

    while let Some(item) = stream.try_next().await? {
        match item {
            Either::Left(result) => {
                if !current_rows.is_empty() {
                    // Accumulated rows belong to the preceding SELECT/SHOW statement.
                    let (data, columns) = crate::to_json::mysql::to_json(
                        std::mem::take(&mut current_rows),
                    )?;
                    results.push(SqlResult::Query { data, columns });
                } else {
                    // No rows accumulated — this is a DML/DDL result.
                    results.push(SqlResult::Execute {
                        rows_affected: result.rows_affected(),
                    });
                }
            }
            Either::Right(row) => {
                current_rows.push(row);
            }
        }
    }

    // Flush any remaining rows (e.g. the last statement was a SELECT with no
    // trailing Either::Left summary emitted by the driver).
    if !current_rows.is_empty() {
        let (data, columns) = crate::to_json::mysql::to_json(current_rows)?;
        results.push(SqlResult::Query { data, columns });
    }

    Ok(results)
}