//! Shared row/timestamp handling for catalog backends.
//!
//! These backends store DuckLake catalog rows with identical schemas and
//! SQL; only the driver row APIs differ. This module holds the row-access
//! bridge trait plus the pure timestamp helpers the embedded pair shares.

#[cfg(any(feature = "sqlite", feature = "duckdb"))]
use chrono::{DateTime, Utc};

use crate::Result;
#[cfg(any(feature = "sqlite", feature = "duckdb", feature = "postgres"))]
use crate::StorageError;

/// Typed row access shared by the embedded backends. Accessors delegate to
/// the driver's own decoding and map driver errors into [`StorageError`] at
/// the boundary, so shared mappers can be written once as
/// `fn(&dyn EmbeddedRow) -> Result<T>`. The one genuinely divergent decode —
/// `snapshot_time` columns — lives in the per-driver `get_time_micros`.
pub(crate) trait EmbeddedRow {
    fn get_i64(&self, idx: usize) -> Result<i64>;
    fn get_opt_i64(&self, idx: usize) -> Result<Option<i64>>;
    fn get_string(&self, idx: usize) -> Result<String>;
    fn get_opt_string(&self, idx: usize) -> Result<Option<String>>;
    fn get_bool(&self, idx: usize) -> Result<bool>;
    fn get_opt_bool(&self, idx: usize) -> Result<Option<bool>>;
    fn get_time_micros(&self, idx: usize) -> Result<i64>;
}

impl<T> EmbeddedRow for &T
where
    T: EmbeddedRow + ?Sized,
{
    fn get_i64(&self, idx: usize) -> Result<i64> {
        (**self).get_i64(idx)
    }

    fn get_opt_i64(&self, idx: usize) -> Result<Option<i64>> {
        (**self).get_opt_i64(idx)
    }

    fn get_string(&self, idx: usize) -> Result<String> {
        (**self).get_string(idx)
    }

    fn get_opt_string(&self, idx: usize) -> Result<Option<String>> {
        (**self).get_opt_string(idx)
    }

    fn get_bool(&self, idx: usize) -> Result<bool> {
        (**self).get_bool(idx)
    }

    fn get_opt_bool(&self, idx: usize) -> Result<Option<bool>> {
        (**self).get_opt_bool(idx)
    }

    fn get_time_micros(&self, idx: usize) -> Result<i64> {
        (**self).get_time_micros(idx)
    }
}

#[cfg(any(feature = "postgres", feature = "quack", test))]
pub(crate) fn collect_embedded_rows<T, R, Rows, Map>(rows: Rows, mut map: Map) -> Result<Vec<T>>
where
    Rows: IntoIterator<Item = R>,
    R: EmbeddedRow,
    Map: FnMut(&dyn EmbeddedRow) -> Result<T>,
{
    let iterator = rows.into_iter();
    let (lower, _) = iterator.size_hint();
    let mut out = Vec::with_capacity(lower);
    for row in iterator {
        out.push(map(&row)?);
    }
    Ok(out)
}

#[cfg(feature = "sqlite")]
impl EmbeddedRow for rusqlite::Row<'_> {
    fn get_i64(&self, idx: usize) -> Result<i64> {
        self.get(idx).map_err(sqlite_row_err)
    }

    fn get_opt_i64(&self, idx: usize) -> Result<Option<i64>> {
        self.get(idx).map_err(sqlite_row_err)
    }

    fn get_string(&self, idx: usize) -> Result<String> {
        self.get(idx).map_err(sqlite_row_err)
    }

    fn get_opt_string(&self, idx: usize) -> Result<Option<String>> {
        self.get(idx).map_err(sqlite_row_err)
    }

    fn get_bool(&self, idx: usize) -> Result<bool> {
        self.get(idx).map_err(sqlite_row_err)
    }

    fn get_opt_bool(&self, idx: usize) -> Result<Option<bool>> {
        self.get(idx).map_err(sqlite_row_err)
    }

    fn get_time_micros(&self, idx: usize) -> Result<i64> {
        use rusqlite::types::ValueRef;

        match self.get_ref(idx).map_err(sqlite_row_err)? {
            ValueRef::Null => Ok(0),
            ValueRef::Integer(value) => Ok(value),
            ValueRef::Real(value) => Ok(value as i64),
            ValueRef::Text(value) => {
                let raw = std::str::from_utf8(value).unwrap_or("");
                if let Ok(micros) = raw.trim().parse::<i64>() {
                    return Ok(micros);
                }
                parse_timestamp_micros(raw)
                    .ok_or_else(|| StorageError::Backend("invalid DuckLake snapshot_time".into()))
            }
            ValueRef::Blob(_) => Err(StorageError::Backend(
                "invalid column type Blob for snapshot_time".into(),
            )),
        }
    }
}

#[cfg(feature = "sqlite")]
fn sqlite_row_err(err: rusqlite::Error) -> StorageError {
    StorageError::Backend(err.to_string())
}

#[cfg(feature = "duckdb")]
impl EmbeddedRow for duckdb::Row<'_> {
    fn get_i64(&self, idx: usize) -> Result<i64> {
        self.get(idx).map_err(duckdb_row_err)
    }

    fn get_opt_i64(&self, idx: usize) -> Result<Option<i64>> {
        self.get(idx).map_err(duckdb_row_err)
    }

    fn get_string(&self, idx: usize) -> Result<String> {
        self.get(idx).map_err(duckdb_row_err)
    }

    fn get_opt_string(&self, idx: usize) -> Result<Option<String>> {
        self.get(idx).map_err(duckdb_row_err)
    }

    fn get_bool(&self, idx: usize) -> Result<bool> {
        self.get(idx).map_err(duckdb_row_err)
    }

    fn get_opt_bool(&self, idx: usize) -> Result<Option<bool>> {
        self.get(idx).map_err(duckdb_row_err)
    }

    fn get_time_micros(&self, idx: usize) -> Result<i64> {
        use duckdb::types::ValueRef;

        match self.get_ref(idx).map_err(duckdb_row_err)? {
            ValueRef::Null => Ok(0),
            ValueRef::TinyInt(value) => Ok(i64::from(value)),
            ValueRef::SmallInt(value) => Ok(i64::from(value)),
            ValueRef::Int(value) => Ok(i64::from(value)),
            ValueRef::BigInt(value) => Ok(value),
            ValueRef::HugeInt(value) => i64::try_from(value)
                .map_err(|_| StorageError::Backend("DuckLake snapshot_time out of range".into())),
            ValueRef::UTinyInt(value) => Ok(i64::from(value)),
            ValueRef::USmallInt(value) => Ok(i64::from(value)),
            ValueRef::UInt(value) => Ok(i64::from(value)),
            ValueRef::UBigInt(value) => i64::try_from(value)
                .map_err(|_| StorageError::Backend("DuckLake snapshot_time out of range".into())),
            ValueRef::Float(value) => Ok(value as i64),
            ValueRef::Double(value) => Ok(value as i64),
            ValueRef::Timestamp(unit, value) => Ok(unit.to_micros(value)),
            ValueRef::Text(value) => {
                parse_timestamp_micros(std::str::from_utf8(value).unwrap_or(""))
                    .ok_or_else(|| StorageError::Backend("invalid DuckLake snapshot_time".into()))
            }
            other => Err(StorageError::Backend(format!(
                "invalid column type {} for snapshot_time",
                other.data_type()
            ))),
        }
    }
}

#[cfg(feature = "duckdb")]
fn duckdb_row_err(err: duckdb::Error) -> StorageError {
    StorageError::Backend(err.to_string())
}

#[cfg(feature = "postgres")]
impl EmbeddedRow for tokio_postgres::Row {
    fn get_i64(&self, idx: usize) -> Result<i64> {
        self.try_get(idx).map_err(postgres_row_err)
    }

    fn get_opt_i64(&self, idx: usize) -> Result<Option<i64>> {
        self.try_get(idx).map_err(postgres_row_err)
    }

    fn get_string(&self, idx: usize) -> Result<String> {
        self.try_get(idx).map_err(postgres_row_err)
    }

    fn get_opt_string(&self, idx: usize) -> Result<Option<String>> {
        self.try_get(idx).map_err(postgres_row_err)
    }

    fn get_bool(&self, idx: usize) -> Result<bool> {
        self.try_get(idx).map_err(postgres_row_err)
    }

    fn get_opt_bool(&self, idx: usize) -> Result<Option<bool>> {
        self.try_get(idx).map_err(postgres_row_err)
    }

    fn get_time_micros(&self, idx: usize) -> Result<i64> {
        self.try_get(idx).map_err(postgres_row_err)
    }
}

#[cfg(feature = "postgres")]
fn postgres_row_err(err: tokio_postgres::Error) -> StorageError {
    StorageError::Backend(err.to_string())
}

#[cfg(feature = "quack")]
impl EmbeddedRow for crate::quack_protocol::QuackRow<'_> {
    fn get_i64(&self, idx: usize) -> Result<i64> {
        crate::quack_protocol::QuackRow::get_i64(self, idx)
    }

    fn get_opt_i64(&self, idx: usize) -> Result<Option<i64>> {
        self.opt_i64(idx)
    }

    fn get_string(&self, idx: usize) -> Result<String> {
        crate::quack_protocol::QuackRow::get_string(self, idx)
    }

    fn get_opt_string(&self, idx: usize) -> Result<Option<String>> {
        self.opt_string(idx)
    }

    fn get_bool(&self, idx: usize) -> Result<bool> {
        crate::quack_protocol::QuackRow::get_bool(self, idx)
    }

    fn get_opt_bool(&self, idx: usize) -> Result<Option<bool>> {
        self.opt_bool(idx)
    }

    fn get_time_micros(&self, idx: usize) -> Result<i64> {
        self.get_timestamp_micros(idx)
    }
}

#[cfg(any(feature = "sqlite", feature = "duckdb"))]
pub(crate) fn format_timestamp_micros(micros: i64) -> Result<String> {
    let seconds = micros.div_euclid(1_000_000);
    let subsecond_micros = micros.rem_euclid(1_000_000);
    let nanos = u32::try_from(subsecond_micros)
        .map_err(|_| StorageError::Backend("snapshot_time is out of range".into()))?
        .checked_mul(1_000)
        .ok_or_else(|| StorageError::Backend("snapshot_time is out of range".into()))?;
    let timestamp = DateTime::<Utc>::from_timestamp(seconds, nanos)
        .ok_or_else(|| StorageError::Backend("snapshot_time is out of range".into()))?;
    Ok(timestamp.format("%Y-%m-%d %H:%M:%S%.6f+00").to_string())
}

#[cfg(any(feature = "sqlite", feature = "duckdb"))]
pub(crate) fn parse_timestamp_micros(raw: &str) -> Option<i64> {
    let value = raw.trim().replace('T', " ");
    if value.len() < 19 {
        return None;
    }

    let year = value.get(0..4)?.parse::<i64>().ok()?;
    let month = value.get(5..7)?.parse::<i64>().ok()?;
    let day = value.get(8..10)?.parse::<i64>().ok()?;
    let hour = value.get(11..13)?.parse::<i64>().ok()?;
    let minute = value.get(14..16)?.parse::<i64>().ok()?;
    let second = value.get(17..19)?.parse::<i64>().ok()?;

    let mut rest = value.get(19..).unwrap_or("").trim();
    let mut micros = 0;
    if let Some(fraction) = rest.strip_prefix('.') {
        let end = fraction
            .find(|ch: char| !ch.is_ascii_digit())
            .unwrap_or(fraction.len());
        let digits = &fraction[..end];
        let padded = format!("{digits:0<6}");
        micros = padded.get(..6)?.parse::<i64>().ok()?;
        rest = fraction[end..].trim();
    }

    let offset_seconds = parse_timezone_offset_seconds(rest).unwrap_or(0);
    let days = days_from_civil(year, month, day);
    let local_seconds = days * 86_400 + hour * 3_600 + minute * 60 + second;
    Some((local_seconds - offset_seconds) * 1_000_000 + micros)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn parse_timezone_offset_seconds(raw: &str) -> Option<i64> {
    let value = raw.trim();
    if value.is_empty() || value.eq_ignore_ascii_case("z") {
        return Some(0);
    }

    let sign = match value.as_bytes().first()? {
        b'+' => 1,
        b'-' => -1,
        _ => return None,
    };
    let body = &value[1..];
    let hour = body.get(0..2)?.parse::<i64>().ok()?;
    let minute = if body.len() >= 5 && body.as_bytes().get(2) == Some(&b':') {
        body.get(3..5)?.parse::<i64>().ok()?
    } else if body.len() >= 4 {
        body.get(2..4)?.parse::<i64>().ok()?
    } else {
        0
    };
    Some(sign * (hour * 3_600 + minute * 60))
}

#[cfg(any(feature = "sqlite", feature = "duckdb"))]
fn days_from_civil(year: i64, month: i64, day: i64) -> i64 {
    let year = year - i64::from(month <= 2);
    let era = if year >= 0 { year } else { year - 399 } / 400;
    let year_of_era = year - era * 400;
    let month_prime = month + if month > 2 { -3 } else { 9 };
    let day_of_year = (153 * month_prime + 2) / 5 + day - 1;
    let day_of_era = year_of_era * 365 + year_of_era / 4 - year_of_era / 100 + day_of_year;
    era * 146_097 + day_of_era - 719_468
}
