use std::time::Instant;

use crate::diagnostic::DiagnosticSeverity;
use crate::{
    DiagnosticSet, Error, MssqlProfile, MssqlType, PlanOptions, PlanOutcome, Result, SchemaMapping,
};

use super::{
    SCHEMA_PLANNING_COMPLETED_EVENT, SCHEMA_PLANNING_FAILED_EVENT, SCHEMA_PLANNING_PHASE,
    SCHEMA_PLANNING_SPAN, SCHEMA_PLANNING_STARTED_EVENT, TRACE_TARGET, append_debug_name,
    append_text, append_unique_text, duration_micros_u64, usize_to_u64_saturating,
};

pub(crate) struct SchemaPlanningTrace {
    span: tracing::Span,
    field_count: usize,
    started: Instant,
}

impl SchemaPlanningTrace {
    pub(crate) fn start(field_count: usize, profile: MssqlProfile, options: PlanOptions) -> Self {
        let span = tracing::info_span!(
            target: TRACE_TARGET,
            SCHEMA_PLANNING_SPAN,
            phase = SCHEMA_PLANNING_PHASE,
            arrow_field_count = usize_to_u64_saturating(field_count),
            mssql_version = ?profile.version(),
            compatibility_level = u64::from(profile.compatibility_level().as_u16()),
            string_policy = ?options.string_policy,
            binary_policy = ?options.binary_policy,
            timezone_policy = ?options.timezone_policy,
            nanosecond_policy = ?options.nanosecond_policy,
            uint64_policy = ?options.uint64_policy,
            decimal_policy = ?options.decimal_policy,
            decimal256_policy = ?options.decimal256_policy,
            float_policy = ?options.float_policy,
            date64_policy = ?options.date64_policy,
        );

        span.in_scope(|| {
            tracing::info!(
                target: TRACE_TARGET,
                phase = SCHEMA_PLANNING_PHASE,
                telemetry_event = SCHEMA_PLANNING_STARTED_EVENT,
                arrow_field_count = usize_to_u64_saturating(field_count)
            );
        });

        Self {
            span,
            field_count,
            started: Instant::now(),
        }
    }

    pub(crate) fn trace_result(
        &self,
        result: Result<PlanOutcome<Vec<SchemaMapping>>>,
    ) -> Result<PlanOutcome<Vec<SchemaMapping>>> {
        match &result {
            Ok(outcome) => self.completed(outcome.value(), outcome.diagnostics()),
            Err(Error::Planning { diagnostics }) => self.failed(diagnostics),
            Err(_) => {}
        }
        result
    }

    pub(crate) fn completed(&self, mappings: &[SchemaMapping], diagnostics: &DiagnosticSet) {
        let summary = DiagnosticTraceSummary::from_diagnostics(diagnostics);
        let shape = PlanningShapeTraceSummary::from_mappings(mappings);
        self.span.in_scope(|| {
            tracing::info!(
                target: TRACE_TARGET,
                phase = SCHEMA_PLANNING_PHASE,
                telemetry_event = SCHEMA_PLANNING_COMPLETED_EVENT,
                arrow_field_count = usize_to_u64_saturating(self.field_count),
                planned_mapping_count = usize_to_u64_saturating(mappings.len()),
                arrow_data_type_families = %shape.arrow_data_type_families,
                mssql_type_families = %shape.mssql_type_families,
                diagnostic_count = usize_to_u64_saturating(summary.total_count),
                error_diagnostic_count = usize_to_u64_saturating(summary.error_count),
                warning_diagnostic_count = usize_to_u64_saturating(summary.warning_count),
                diagnostic_codes = %summary.codes,
                diagnostic_field_names = %summary.field_names,
                elapsed_us = duration_micros_u64(self.started.elapsed())
            );
        });
    }

    pub(crate) fn failed(&self, diagnostics: &DiagnosticSet) {
        let summary = DiagnosticTraceSummary::from_diagnostics(diagnostics);
        self.span.in_scope(|| {
            tracing::error!(
                target: TRACE_TARGET,
                phase = SCHEMA_PLANNING_PHASE,
                telemetry_event = SCHEMA_PLANNING_FAILED_EVENT,
                arrow_field_count = usize_to_u64_saturating(self.field_count),
                diagnostic_count = usize_to_u64_saturating(summary.total_count),
                error_diagnostic_count = usize_to_u64_saturating(summary.error_count),
                warning_diagnostic_count = usize_to_u64_saturating(summary.warning_count),
                diagnostic_codes = %summary.codes,
                diagnostic_field_names = %summary.field_names,
                error_summary = "schema planning failed with diagnostics",
                elapsed_us = duration_micros_u64(self.started.elapsed())
            );
        });
    }
}

#[derive(Debug, Default, PartialEq, Eq)]
struct PlanningShapeTraceSummary {
    arrow_data_type_families: String,
    mssql_type_families: String,
}

impl PlanningShapeTraceSummary {
    fn from_mappings(mappings: &[SchemaMapping]) -> Self {
        let mut summary = Self::default();

        for mapping in mappings {
            append_unique_text(
                &mut summary.arrow_data_type_families,
                arrow_data_type_family(mapping.arrow().data_type()),
            );
            append_unique_text(
                &mut summary.mssql_type_families,
                mssql_type_family(mapping.mssql().ty()),
            );
        }

        summary
    }
}

#[derive(Debug, Default, PartialEq, Eq)]
struct DiagnosticTraceSummary {
    total_count: usize,
    error_count: usize,
    warning_count: usize,
    codes: String,
    field_names: String,
}

impl DiagnosticTraceSummary {
    fn from_diagnostics(diagnostics: &DiagnosticSet) -> Self {
        let mut summary = Self::default();

        for diagnostic in diagnostics.all() {
            summary.total_count += 1;
            match diagnostic.severity() {
                DiagnosticSeverity::Warning => summary.warning_count += 1,
                DiagnosticSeverity::Error => summary.error_count += 1,
            }

            append_debug_name(&mut summary.codes, diagnostic.code());
            if let Some(field) = diagnostic.field() {
                append_text(&mut summary.field_names, field.name());
            }
        }

        summary
    }
}

fn arrow_data_type_family(data_type: &arrow_schema::DataType) -> &'static str {
    match data_type {
        arrow_schema::DataType::Null => "null",
        arrow_schema::DataType::Boolean => "boolean",
        arrow_schema::DataType::Int8 => "int8",
        arrow_schema::DataType::Int16 => "int16",
        arrow_schema::DataType::Int32 => "int32",
        arrow_schema::DataType::Int64 => "int64",
        arrow_schema::DataType::UInt8 => "uint8",
        arrow_schema::DataType::UInt16 => "uint16",
        arrow_schema::DataType::UInt32 => "uint32",
        arrow_schema::DataType::UInt64 => "uint64",
        arrow_schema::DataType::Float16 => "float16",
        arrow_schema::DataType::Float32 => "float32",
        arrow_schema::DataType::Float64 => "float64",
        arrow_schema::DataType::Utf8 => "utf8",
        arrow_schema::DataType::LargeUtf8 => "large_utf8",
        arrow_schema::DataType::Binary => "binary",
        arrow_schema::DataType::LargeBinary => "large_binary",
        arrow_schema::DataType::FixedSizeBinary(_) => "fixed_size_binary",
        arrow_schema::DataType::Decimal32(_, _) => "decimal32",
        arrow_schema::DataType::Decimal64(_, _) => "decimal64",
        arrow_schema::DataType::Decimal128(_, _) => "decimal128",
        arrow_schema::DataType::Decimal256(_, _) => "decimal256",
        arrow_schema::DataType::Date32 => "date32",
        arrow_schema::DataType::Date64 => "date64",
        arrow_schema::DataType::Time32(_) => "time32",
        arrow_schema::DataType::Time64(_) => "time64",
        arrow_schema::DataType::Timestamp(_, _) => "timestamp",
        arrow_schema::DataType::Duration(_) => "duration",
        arrow_schema::DataType::Interval(_) => "interval",
        arrow_schema::DataType::List(_)
        | arrow_schema::DataType::ListView(_)
        | arrow_schema::DataType::FixedSizeList(_, _)
        | arrow_schema::DataType::LargeList(_)
        | arrow_schema::DataType::LargeListView(_)
        | arrow_schema::DataType::Struct(_)
        | arrow_schema::DataType::Map(_, _)
        | arrow_schema::DataType::Union(_, _) => "nested",
        arrow_schema::DataType::Dictionary(_, _) | arrow_schema::DataType::RunEndEncoded(_, _) => {
            "encoded"
        }
        arrow_schema::DataType::BinaryView => "binary_view",
        arrow_schema::DataType::Utf8View => "utf8_view",
    }
}

fn mssql_type_family(ty: &MssqlType) -> &'static str {
    match ty {
        MssqlType::Bit => "bit",
        MssqlType::TinyInt => "tinyint",
        MssqlType::SmallInt => "smallint",
        MssqlType::Int => "int",
        MssqlType::BigInt => "bigint",
        MssqlType::Real => "real",
        MssqlType::Float { .. } => "float",
        MssqlType::NVarChar(_) => "nvarchar",
        MssqlType::VarBinary(_) => "varbinary",
        MssqlType::Binary(_) => "binary",
        MssqlType::Decimal { .. } => "decimal",
        MssqlType::Date => "date",
        MssqlType::Time(_) => "time",
        MssqlType::DateTime2 { .. } => "datetime2",
        MssqlType::DateTimeOffset { .. } => "datetimeoffset",
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex, MutexGuard};

    use arrow_schema::{DataType, Field, Schema, TimeUnit};
    use tracing::Level;

    use crate::{
        Diagnostic, DiagnosticCode, DiagnosticSet, Error, FieldRef, MssqlProfile, PlanOptions,
        TimezonePolicy, plan_arrow_schema_to_mssql_mappings,
    };

    use super::{
        DiagnosticTraceSummary, SCHEMA_PLANNING_COMPLETED_EVENT, SCHEMA_PLANNING_FAILED_EVENT,
        SCHEMA_PLANNING_PHASE, SCHEMA_PLANNING_SPAN, SCHEMA_PLANNING_STARTED_EVENT, TRACE_TARGET,
    };
    use crate::observability::test_support::{CapturedTraceKind, capture_traces};

    static SCHEMA_TRACE_TEST_LOCK: Mutex<()> = Mutex::new(());

    fn schema_trace_test_guard() -> MutexGuard<'static, ()> {
        match SCHEMA_TRACE_TEST_LOCK.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        }
    }

    #[test]
    fn diagnostic_trace_summary_counts_warning_codes() {
        let diagnostics = DiagnosticSet::from(vec![
            Diagnostic::warning(DiagnosticCode::PolicyApplied, "policy applied")
                .with_field(FieldRef::new(0, "amount")),
            Diagnostic::error(
                DiagnosticCode::ProfileDependentConversion,
                "policy required",
            )
            .with_field(FieldRef::new(1, "unsigned_huge")),
        ]);

        let summary = DiagnosticTraceSummary::from_diagnostics(&diagnostics);

        assert_eq!(summary.total_count, 2);
        assert_eq!(summary.warning_count, 1);
        assert_eq!(summary.error_count, 1);
        assert_eq!(summary.codes, "PolicyApplied,ProfileDependentConversion");
        assert_eq!(summary.field_names, "amount,unsigned_huge");
    }

    #[test]
    fn successful_schema_planning_emits_structured_trace() -> Result<(), String> {
        let _trace_guard = schema_trace_test_guard();
        let schema = Schema::new(vec![
            Field::new("is_active", DataType::Boolean, false),
            Field::new("quantity", DataType::Int32, true),
        ]);

        let (outcome, traces) = capture_traces(|| {
            plan_arrow_schema_to_mssql_mappings(
                Arc::new(schema),
                MssqlProfile::sql_server_2016_compat_100(),
                PlanOptions::default(),
            )
        });
        let outcome = outcome.map_err(|err| err.to_string())?;
        assert_eq!(outcome.value().len(), 2);

        let records = traces.records()?;
        assert!(
            records.iter().any(|record| {
                record.kind() == CapturedTraceKind::Span
                    && record.name() == SCHEMA_PLANNING_SPAN
                    && record.target() == TRACE_TARGET
                    && record.level() == Level::INFO
                    && record
                        .fields()
                        .get("phase")
                        .is_some_and(|value| value == SCHEMA_PLANNING_PHASE)
                    && record
                        .fields()
                        .get("arrow_field_count")
                        .is_some_and(|value| value == "2")
                    && record
                        .fields()
                        .get("mssql_version")
                        .is_some_and(|value| value == "SqlServer2016")
                    && record
                        .fields()
                        .get("compatibility_level")
                        .is_some_and(|value| value == "100")
                    && record
                        .fields()
                        .get("uint64_policy")
                        .is_some_and(|value| value == "Reject")
            }),
            "captured records: {records:#?}"
        );

        assert!(
            records.iter().any(|record| {
                record.kind() == CapturedTraceKind::Event
                    && record.target() == TRACE_TARGET
                    && record.level() == Level::INFO
                    && record.span_name() == Some(SCHEMA_PLANNING_SPAN)
                    && record
                        .fields()
                        .get("telemetry_event")
                        .is_some_and(|value| value == SCHEMA_PLANNING_STARTED_EVENT)
                    && record
                        .fields()
                        .get("arrow_field_count")
                        .is_some_and(|value| value == "2")
            }),
            "captured records: {records:#?}"
        );

        assert!(
            records.iter().any(|record| {
                record.kind() == CapturedTraceKind::Event
                    && record.target() == TRACE_TARGET
                    && record.level() == Level::INFO
                    && record.span_name() == Some(SCHEMA_PLANNING_SPAN)
                    && record
                        .fields()
                        .get("telemetry_event")
                        .is_some_and(|value| value == SCHEMA_PLANNING_COMPLETED_EVENT)
                    && record
                        .fields()
                        .get("planned_mapping_count")
                        .is_some_and(|value| value == "2")
                    && record
                        .fields()
                        .get("arrow_data_type_families")
                        .is_some_and(|value| value == "boolean,int32")
                    && record
                        .fields()
                        .get("mssql_type_families")
                        .is_some_and(|value| value == "bit,int")
                    && record
                        .fields()
                        .get("diagnostic_count")
                        .is_some_and(|value| value == "0")
                    && record
                        .fields()
                        .get("error_diagnostic_count")
                        .is_some_and(|value| value == "0")
                    && record
                        .fields()
                        .get("warning_diagnostic_count")
                        .is_some_and(|value| value == "0")
                    && record.fields().contains_key("elapsed_us")
            }),
            "captured records: {records:#?}"
        );

        Ok(())
    }

    #[test]
    fn failed_schema_planning_emits_sanitized_diagnostic_trace() -> Result<(), String> {
        let _trace_guard = schema_trace_test_guard();
        let schema = Schema::new(vec![
            Field::new("ok", DataType::Int32, false),
            Field::new("unsigned_huge", DataType::UInt64, true),
            Field::new("items", DataType::new_list(DataType::Int64, true), true),
        ]);

        let (outcome, traces) = capture_traces(|| {
            plan_arrow_schema_to_mssql_mappings(
                Arc::new(schema),
                MssqlProfile::sql_server_2016_compat_100(),
                PlanOptions::default(),
            )
        });
        let err = outcome.expect_err("unsupported planning should fail");
        let Error::Planning { diagnostics } = err else {
            panic!("expected planning error");
        };
        assert_eq!(diagnostics.len(), 2);

        let records = traces.records()?;
        assert!(
            records.iter().any(|record| {
                record.kind() == CapturedTraceKind::Event
                    && record.target() == TRACE_TARGET
                    && record.level() == Level::ERROR
                    && record.span_name() == Some(SCHEMA_PLANNING_SPAN)
                    && record
                        .fields()
                        .get("telemetry_event")
                        .is_some_and(|value| value == SCHEMA_PLANNING_FAILED_EVENT)
                    && record
                        .fields()
                        .get("arrow_field_count")
                        .is_some_and(|value| value == "3")
                    && record
                        .fields()
                        .get("diagnostic_count")
                        .is_some_and(|value| value == "2")
                    && record
                        .fields()
                        .get("error_diagnostic_count")
                        .is_some_and(|value| value == "2")
                    && record
                        .fields()
                        .get("warning_diagnostic_count")
                        .is_some_and(|value| value == "0")
                    && record
                        .fields()
                        .get("diagnostic_codes")
                        .is_some_and(|value| {
                            value.contains("ProfileDependentConversion")
                                && value.contains("UnsupportedArrowType")
                        })
                    && record
                        .fields()
                        .get("diagnostic_field_names")
                        .is_some_and(|value| {
                            value.contains("unsigned_huge") && value.contains("items")
                        })
                    && record
                        .fields()
                        .get("error_summary")
                        .is_some_and(|value| value == "schema planning failed with diagnostics")
                    && record.fields().contains_key("elapsed_us")
            }),
            "captured records: {records:#?}"
        );

        Ok(())
    }

    #[test]
    fn schema_planning_trace_does_not_emit_field_metadata_values() -> Result<(), String> {
        let _trace_guard = schema_trace_test_guard();
        let schema = Schema::new(vec![
            Field::new("id", DataType::Int32, false).with_metadata(
                [(
                    "connection_hint".to_owned(),
                    "server=tcp:sql.example.com;password=secret".to_owned(),
                )]
                .into_iter()
                .collect(),
            ),
            Field::new("label", DataType::Utf8, true).with_metadata(
                [("token".to_owned(), "access_token=abc123".to_owned())]
                    .into_iter()
                    .collect(),
            ),
            Field::new(
                "created_at",
                DataType::Timestamp(TimeUnit::Second, Some("password=timezone-secret".into())),
                true,
            ),
        ]);

        let (outcome, traces) = capture_traces(|| {
            plan_arrow_schema_to_mssql_mappings(
                Arc::new(schema),
                MssqlProfile::sql_server_2016_compat_100(),
                PlanOptions {
                    timezone_policy: TimezonePolicy::DateTimeOffset,
                    ..PlanOptions::default()
                },
            )
        });
        outcome.map_err(|err| err.to_string())?;

        traces.assert_no_forbidden_text(&[
            "server=tcp:sql.example.com",
            "password=secret",
            "access_token=abc123",
            "password=timezone-secret",
        ])?;

        Ok(())
    }
}
