// Copyright 2025 The Drasi Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Oracle LogMiner helpers.

use crate::scn::Scn;
use anyhow::{anyhow, Result};
use oracle::Connection;
use sqlparser::{
    ast::{AssignmentTarget, Expr, SetExpr, Statement},
    dialect::GenericDialect,
    parser::Parser,
};
use std::collections::HashMap;

pub struct LogMinerGuard<'a> {
    conn: &'a Connection,
    active: bool,
}

impl<'a> LogMinerGuard<'a> {
    pub fn new(conn: &'a Connection, start_scn: Scn, end_scn: Scn) -> Result<Self> {
        let sql = format!(
            "BEGIN DBMS_LOGMNR.START_LOGMNR(\n                STARTSCN => {},\n                ENDSCN   => {},\n                OPTIONS  => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG\n                          + DBMS_LOGMNR.COMMITTED_DATA_ONLY\n                          + DBMS_LOGMNR.NO_SQL_DELIMITER\n            ); END;",
            start_scn.0, end_scn.0
        );
        conn.execute(&sql, &[])?;
        Ok(Self { conn, active: true })
    }

    pub fn stop(&mut self) -> Result<()> {
        if self.active {
            self.conn
                .execute("BEGIN DBMS_LOGMNR.END_LOGMNR; END;", &[])?;
            self.active = false;
        }
        Ok(())
    }
}

impl Drop for LogMinerGuard<'_> {
    fn drop(&mut self) {
        if self.active {
            let _ = self.conn.execute("BEGIN DBMS_LOGMNR.END_LOGMNR; END;", &[]);
            self.active = false;
        }
    }
}

pub fn parse_sql_undo_insert(sql_undo: &str) -> Result<HashMap<String, String>> {
    let statement = parse_sql_undo_statement(sql_undo, "INSERT")?;

    let (column_names, parsed_values) = match statement {
        Statement::Insert(insert) => {
            let column_names = insert
                .columns
                .iter()
                .map(|column| normalize_sql_name(&column.to_string()))
                .collect::<Vec<_>>();

            let rows = match insert.source.as_ref().map(|query| query.body.as_ref()) {
                Some(SetExpr::Values(values)) => &values.rows,
                _ => {
                    return Err(anyhow!(
                        "SQL_UNDO INSERT has no VALUES clause (len={})",
                        sql_undo.len()
                    ))
                }
            };

            let row = rows.first().ok_or_else(|| {
                anyhow!("SQL_UNDO INSERT VALUES is empty (len={})", sql_undo.len())
            })?;

            (
                column_names,
                row.iter().map(expr_to_sql_string).collect::<Vec<_>>(),
            )
        }
        _ => {
            return Err(anyhow!(
                "SQL_UNDO is not an INSERT statement (len={})",
                sql_undo.len()
            ))
        }
    };

    if column_names.len() != parsed_values.len() {
        return Err(anyhow!(
            "SQL_UNDO INSERT column count {} does not match value count {}",
            column_names.len(),
            parsed_values.len()
        ));
    }

    Ok(column_names.into_iter().zip(parsed_values).collect())
}

pub fn parse_sql_undo_update(sql_undo: &str) -> Result<HashMap<String, String>> {
    let statement = parse_sql_undo_statement(sql_undo, "UPDATE")?;

    let assignments = match statement {
        Statement::Update { assignments, .. } => assignments,
        _ => {
            return Err(anyhow!(
                "SQL_UNDO is not an UPDATE statement (len={})",
                sql_undo.len()
            ))
        }
    };

    let mut parsed = HashMap::new();
    for assignment in assignments {
        let column = match assignment.target {
            AssignmentTarget::ColumnName(column) => normalize_sql_name(&column.to_string()),
            _ => continue,
        };
        parsed.insert(column, expr_to_sql_string(&assignment.value));
    }

    Ok(parsed)
}

fn parse_sql_undo_statement(sql_undo: &str, statement_kind: &str) -> Result<Statement> {
    let dialect = GenericDialect {};
    let statements = Parser::parse_sql(&dialect, sql_undo).map_err(|_| {
        anyhow!(
            "Failed to parse SQL_UNDO {statement_kind} (len={})",
            sql_undo.len()
        )
    })?;

    statements.into_iter().next().ok_or_else(|| {
        anyhow!(
            "Empty SQL_UNDO {statement_kind} statement (len={})",
            sql_undo.len()
        )
    })
}

fn normalize_sql_name(name: &str) -> String {
    name.trim().trim_matches('"').to_lowercase()
}

fn expr_to_sql_string(expr: &Expr) -> String {
    expr.to_string()
}

pub fn split_sql_values(values: &str) -> Vec<String> {
    let mut result = Vec::new();
    let mut current = String::new();
    let mut in_string = false;
    let mut paren_depth = 0i32;
    let mut chars = values.chars().peekable();

    while let Some(ch) = chars.next() {
        match ch {
            '\'' if !in_string => {
                in_string = true;
                current.push(ch);
            }
            '\'' if in_string => {
                current.push(ch);
                if chars.peek() == Some(&'\'') {
                    current.push(chars.next().expect("peeked escaped quote must exist"));
                } else {
                    in_string = false;
                }
            }
            '(' if !in_string => {
                paren_depth += 1;
                current.push(ch);
            }
            ')' if !in_string => {
                if paren_depth > 0 {
                    paren_depth -= 1;
                }
                current.push(ch);
            }
            ',' if !in_string && paren_depth == 0 => {
                result.push(current.trim().to_string());
                current.clear();
            }
            _ => current.push(ch),
        }
    }

    if !current.trim().is_empty() {
        result.push(current.trim().to_string());
    }

    result
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_split_sql_values() {
        let values = split_sql_values("1, 'O''Brien', TO_DATE('2024-01-01', 'YYYY-MM-DD'), NULL");
        assert_eq!(values.len(), 4);
        assert_eq!(values[1], "'O''Brien'");
        assert_eq!(values[2], "TO_DATE('2024-01-01', 'YYYY-MM-DD')");
    }

    #[test]
    fn test_parse_sql_undo_insert() {
        let parsed = parse_sql_undo_insert(
            r#"INSERT INTO "SYSTEM"."DRASI_TEST"("ID","NAME") VALUES (2,'Bob')"#,
        )
        .unwrap();
        assert_eq!(parsed.get("id").map(String::as_str), Some("2"));
        assert_eq!(parsed.get("name").map(String::as_str), Some("'Bob'"));
    }

    #[test]
    fn test_parse_sql_undo_update() {
        let parsed = parse_sql_undo_update(
            r#"update "SYSTEM"."DRASI_TEST" set "NAME" = 'Bob', "UPDATED_AT" = TO_DATE('2024-01-01', 'YYYY-MM-DD') where "ID" = 2"#,
        )
        .unwrap();
        assert_eq!(parsed.get("name").map(String::as_str), Some("'Bob'"));
        assert_eq!(
            parsed.get("updated_at").map(String::as_str),
            Some("TO_DATE('2024-01-01', 'YYYY-MM-DD')")
        );
    }
}
