use std::{ops::RangeInclusive, sync::Arc};

use crate::err::{TemplateSubstitutionError, TranslationError};
use derive_more::derive::From;

use crate::{
    tree::{
        ast::{
            clause::{
                Assignment, AssignmentClause, FromClause, Selection, TableAlias, TableReference,
                TableReferenceTarget, TemplatedTablePath,
            },
            dataset_identifier::DatasetIdentifier,
            identifier::{Identifier, ParsedIdentifier, ParsedSimpleIdentifier, SimpleIdentifier},
            node::Spannable,
        },
        expand_templated_table_path,
        typed_ast::{context::ExpressionTranslationContext, environment::TypeEnvironment},
    },
    types::{matcher::ExactMatcher, struct_type::Struct, BOOLEAN},
};

use super::{context::StatementTranslationContext, expression::TypedExpression};

/// Result of checking for identifier conflicts within an assignment list.
pub struct IdentifierConflict {
    /// Span of the earlier conflicting identifier (for "previously defined here" annotation)
    pub existing_span: RangeInclusive<usize>,
    /// Explanation of the conflict (shown as a note)
    pub note: String,
}

/// Check if a new identifier conflicts with any already-seen identifiers.
///
/// Detects:
/// - Exact duplicate: same identifier appears twice
/// - Prefix conflict: one identifier is a strict prefix of another
///   (e.g., `x` and `x.y` — `x` can't be both assigned directly and traversed into)
pub fn find_identifier_conflict(
    seen: &[(Identifier, RangeInclusive<usize>)],
    new: &Identifier,
) -> Option<IdentifierConflict> {
    for (existing, span) in seen.iter() {
        if existing == new {
            return Some(IdentifierConflict {
                existing_span: span.clone(),
                note: format!("'{}' is assigned twice in the same command", new),
            });
        }
        if new.has_prefix(existing) {
            return Some(IdentifierConflict {
                existing_span: span.clone(),
                note: format!(
                    "'{}' is a child of '{}' — '{}' can't be both assigned directly and traversed into",
                    new, existing, existing
                ),
            });
        }
        if existing.has_prefix(new) {
            return Some(IdentifierConflict {
                existing_span: span.clone(),
                note: format!(
                    "'{}' is a child of '{}' — '{}' can't be both assigned directly and traversed into",
                    existing, new, new
                ),
            });
        }
    }
    None
}

#[derive(Debug, Clone)]
pub struct Projections {
    pub assignments: Vec<TypedAssignment>,
}

#[derive(Debug, Clone)]
pub struct TypedAssignment {
    pub identifier: ParsedIdentifier,
    pub expression: Arc<TypedExpression>,
}

impl Spannable for TypedAssignment {
    fn span(&self) -> Option<RangeInclusive<usize>> {
        // Try to span from start of identifier to end of expression
        // Fall back to just expression span if identifier has no span (e.g., inferred from struct literal)
        match (self.identifier.span(), self.expression.ast.span()) {
            (Some(id_span), Some(expr_span)) => Some(*id_span.start()..=*expr_span.end()),
            (None, Some(expr_span)) => Some(expr_span),
            (Some(id_span), None) => Some(id_span),
            (None, None) => None,
        }
    }
}

impl Projections {
    pub fn from_assignment_clauses_with_context(
        clauses: &Vec<Arc<AssignmentClause>>,
        ctx: &mut ExpressionTranslationContext,
    ) -> Self {
        let assignments = clauses
            .iter()
            .filter_map(|clause| match clause.as_ref() {
                AssignmentClause::Assignment(assignment) => Some(TypedAssignment {
                    identifier: assignment.identifier.clone(),
                    expression: Arc::new(TypedExpression::from_ast_with_context(
                        assignment.expression.clone(),
                        ctx,
                    )),
                }),
                AssignmentClause::Error(_) => None,
            })
            .collect();

        Self { assignments }
    }

    pub fn from_assignments_with_context(
        assignments: &Vec<Arc<Assignment>>,
        ctx: &mut ExpressionTranslationContext,
    ) -> Self {
        let assignments = assignments
            .iter()
            .map(|assignment| TypedAssignment {
                identifier: assignment.identifier.clone(),
                expression: Arc::new(TypedExpression::from_ast_with_context(
                    assignment.expression.clone(),
                    ctx,
                )),
            })
            .collect();

        Self { assignments }
    }

    /// Check for identifier conflicts within this projection list and optionally
    /// across additional projection lists. Returns the set of valid (non-conflicting)
    /// identifiers. Conflicting assignments are reported as errors on `ctx`.
    pub fn check_conflicts(
        &self,
        ctx: &mut StatementTranslationContext,
        additional: &[&Projections],
    ) -> Vec<Identifier> {
        let all_assignments: Vec<&TypedAssignment> = self
            .assignments
            .iter()
            .chain(additional.iter().flat_map(|p| p.assignments.iter()))
            .collect();

        let mut seen: Vec<(Identifier, RangeInclusive<usize>)> = Vec::new();
        for typed_assignment in &all_assignments {
            if let Ok(identifier) = typed_assignment.identifier.valid_ref() {
                if let Some(conflict) = find_identifier_conflict(&seen, identifier) {
                    ctx.error(format!("conflicting definition for '{}'", identifier))
                        .at(&typed_assignment.identifier)
                        .add_context(conflict.existing_span, "previously defined here")
                        .with_note(conflict.note)
                        .emit();
                } else {
                    seen.push((
                        identifier.clone(),
                        typed_assignment.identifier.span().unwrap_or(0..=0),
                    ));
                }
            }
        }
        seen.into_iter().map(|(id, _)| id).collect()
    }

    pub fn types(&self) -> Struct {
        let mut new_fields = Struct::default();
        for typed_assignment in &self.assignments {
            // Extract valid identifier for struct field
            if let Ok(identifier) = typed_assignment.identifier.clone().valid() {
                new_fields.insert_arc(
                    identifier,
                    typed_assignment.expression.resolved_type.clone(),
                );
            }
            // Skip error identifiers - they don't contribute to the type environment
        }

        new_fields
    }

    pub fn lookup(&self, id: &Identifier) -> Option<&TypedAssignment> {
        self.assignments
            .iter()
            .find(|assignment| assignment.identifier.valid_ref().ok() == Some(id))
    }
}

// ============================================================================
// ResolvedEnvironment - Catalog Resolution Result
// ============================================================================

/// Result of resolving a table reference from the catalog
///
/// Separates resolution concerns from aliasing concerns.
/// Resolution errors (table not found) are stored here.
#[derive(Debug, Clone, From)]
pub enum ResolvedEnvironment {
    /// Table was found in catalog
    Resolved(Arc<TypeEnvironment>),

    /// Table not found in catalog
    Error(Arc<TranslationError>),
}

impl ResolvedEnvironment {
    /// Get the environment, or empty if error
    pub fn environment(&self) -> Arc<TypeEnvironment> {
        match self {
            ResolvedEnvironment::Resolved(env) => env.clone(),
            ResolvedEnvironment::Error(_) => Arc::new(TypeEnvironment::default()),
        }
    }
}

// ============================================================================
// TypedFromClause - Type-checked FROM clause
// ============================================================================

/// A type-checked FROM clause
///
/// Three-variant enum that models the AST structure:
/// - Reference: Plain table reference (FROM events)
/// - Alias: Aliased table reference (FROM x = events)
/// - Error: Clause-level structural errors (e.g., schema merge failure)
#[derive(Debug, Clone, From)]
pub enum TypedFromClause {
    Reference(TypedTableReference),
    Alias(TypedTableAlias),
    Error(Arc<TranslationError>),
}

impl TypedFromClause {
    /// Type-check a FROM clause by resolving it from the catalog
    pub fn from_ast_with_context(
        clause: Arc<FromClause>,
        ctx: &mut StatementTranslationContext,
    ) -> Self {
        match clause.as_ref() {
            FromClause::TableReference(table_ref) => {
                match TypedTableReference::try_from_ast_with_context(table_ref.clone(), ctx) {
                    Ok(typed_ref) => typed_ref.into(),
                    Err(err) => err.into(),
                }
            }
            FromClause::TableAlias(table_alias) => {
                match TypedTableAlias::try_from_ast_with_context(table_alias.clone(), ctx) {
                    Ok(typed_alias) => typed_alias.into(),
                    Err(err) => err.into(),
                }
            }
        }
    }

    /// Get the resolved environment (or empty if error)
    pub fn environment(&self) -> Arc<TypeEnvironment> {
        match self {
            TypedFromClause::Reference(r) => r.resolved.environment(),
            TypedFromClause::Alias(a) => a.resolved.environment(),
            TypedFromClause::Error(_) => Arc::new(TypeEnvironment::default()),
        }
    }

    /// Get the alias if this is an Alias variant
    pub fn alias(&self) -> Option<&ParsedSimpleIdentifier> {
        match self {
            TypedFromClause::Alias(a) => Some(&a.alias),
            _ => None,
        }
    }

    /// Get the table reference
    pub fn reference(&self) -> Result<&TableReference, &Arc<TranslationError>> {
        match self {
            TypedFromClause::Reference(r) => Ok(&r.ast),
            TypedFromClause::Alias(a) => Ok(&a.ast.table),
            TypedFromClause::Error(translation_error) => Err(translation_error),
        }
    }
}

// ============================================================================
// TypedTableReference - Plain table reference without alias
// ============================================================================

/// A type-checked table reference without explicit alias
///
/// Models "FROM events" syntax - no alias, fields merged at top level.
#[derive(Debug, Clone)]
pub struct TypedTableReference {
    /// Reference to the AST table reference
    pub ast: Arc<TableReference>,

    /// The resolved environment (or error if table not found)
    pub resolved: ResolvedEnvironment,
}

impl TypedTableReference {
    /// Type-check a table reference by resolving it from the catalog
    /// Returns Err if the identifier itself is invalid
    pub fn try_from_ast_with_context(
        table_ref: Arc<TableReference>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<Self, Arc<TranslationError>> {
        let resolved =
            Self::resolve_table_reference_target(&table_ref.target, ctx, table_ref.as_ref())?;

        Ok(Self {
            ast: table_ref,
            resolved,
        })
    }

    fn resolve_table_reference_target(
        target: &TableReferenceTarget,
        ctx: &mut StatementTranslationContext,
        error_node: &impl Spannable,
    ) -> Result<ResolvedEnvironment, Arc<TranslationError>> {
        match target {
            TableReferenceTarget::Static(id) => {
                let dataset = id.valid_ref().map_err(|e| e.clone())?;
                Ok(Self::resolve_table_schema(dataset, ctx, error_node))
            }
            TableReferenceTarget::Templated(path) => {
                Ok(Self::resolve_templated_table_schema(path, ctx, error_node))
            }
        }
    }

    /// Resolve a table schema from CTEs first, then fall back to catalog provider
    ///
    /// Returns the resolved environment or an error if not found.
    fn resolve_table_schema(
        dataset: &DatasetIdentifier,
        ctx: &mut StatementTranslationContext,
        error_node: &impl Spannable,
    ) -> ResolvedEnvironment {
        match Self::reflect_struct(dataset, ctx, error_node) {
            Ok(s) => ResolvedEnvironment::Resolved(Arc::new(s.into())),
            Err(e) => ResolvedEnvironment::Error(e),
        }
    }

    fn reflect_struct(
        dataset: &DatasetIdentifier,
        ctx: &mut StatementTranslationContext,
        error_node: &impl Spannable,
    ) -> Result<Struct, Arc<TranslationError>> {
        if let DatasetIdentifier::Unqualified(unqualified) = dataset {
            if unqualified.namespace.is_empty() {
                let cte_id: Identifier = unqualified.table.clone().into();
                if let Some(cte_schema) = ctx.lookup_cte(&cte_id) {
                    return Ok(cte_schema.as_struct().clone());
                }
            }
        }

        let qualified = dataset
            .clone()
            .qualify(ctx.default_space())
            .map_err(|e| ctx.error(e.to_string()).at(error_node).emit())?;

        ctx.provider
            .reflect_columns(&qualified)
            .map_err(|e| ctx.error(e.to_string()).at(error_node).emit())
    }

    /// Resolve a templated table reference by enumerating its concrete identifiers and intersecting
    /// their schemas. Mirrors `resolve_table_schema` for the static case: any failure (missing
    /// declaration, kind mismatch, expansion blow-up, reflection error) is embedded as
    /// `ResolvedEnvironment::Error` rather than bubbled, so type-checking can continue past the
    /// reference.
    fn resolve_templated_table_schema(
        path: &TemplatedTablePath,
        ctx: &mut StatementTranslationContext,
        error_node: &impl Spannable,
    ) -> ResolvedEnvironment {
        // Domain lookup against the typecheck context. Failure modes are surfaced as
        // `TemplateSubstitutionError` so the underlying expander can stay free of context types.
        let lookup = |p: &SimpleIdentifier| -> Result<Arc<[String]>, TemplateSubstitutionError> {
            if let Some(d) = ctx.lookup_template_param_enum(p) {
                Ok(d)
            } else if ctx.lookup_template_param_type(p).is_some() {
                Err(
                    TemplateSubstitutionError::TablePathParameterNeedsIdentifierFragment(
                        p.to_string(),
                    ),
                )
            } else {
                Err(TemplateSubstitutionError::UndeclaredTemplateParameter(
                    p.to_string(),
                ))
            }
        };
        let identifiers = match expand_templated_table_path(path, lookup) {
            Ok(ids) => ids,
            Err(e) => {
                let err = ctx.error(e.to_string()).at(error_node).emit();
                return ResolvedEnvironment::Error(err);
            }
        };

        let mut schemas: Vec<Struct> = Vec::with_capacity(identifiers.len());
        for ident in identifiers {
            match Self::reflect_struct(&ident, ctx, error_node) {
                Ok(s) => schemas.push(s),
                Err(e) => return ResolvedEnvironment::Error(e),
            }
        }

        let intersected = Struct::intersect_equal(&schemas);
        ResolvedEnvironment::Resolved(Arc::new(intersected.into()))
    }
}

// ============================================================================
// TypedTableAlias - Aliased table reference
// ============================================================================

/// A type-checked table reference with explicit alias
///
/// Models "FROM x = events" syntax - alias is required and stored.
#[derive(Debug, Clone)]
pub struct TypedTableAlias {
    /// Reference to the AST table alias
    pub ast: Arc<TableAlias>,

    /// The alias identifier (simple identifier, matches grammar)
    pub alias: ParsedSimpleIdentifier,

    /// The resolved environment (or error if table not found)
    pub resolved: ResolvedEnvironment,
}

impl TypedTableAlias {
    /// Type-check a table alias by resolving it from the catalog
    /// Returns Err if the identifier itself is invalid
    fn try_from_ast_with_context(
        table_alias: Arc<TableAlias>,
        ctx: &mut StatementTranslationContext,
    ) -> Result<Self, Arc<TranslationError>> {
        let resolved = TypedTableReference::resolve_table_reference_target(
            &table_alias.table.target,
            ctx,
            table_alias.as_ref(),
        )?;

        Ok(Self {
            ast: table_alias.clone(),
            alias: table_alias.alias.clone(),
            resolved,
        })
    }
}

// ============================================================================
// TypedConstraint - WHEN clause constraints
// ============================================================================

/// A type-checked constraint from the WHEN clause
#[derive(Debug, Clone)]
pub struct TypedConstraint {
    /// Reference to the AST assignment
    pub ast: Arc<Assignment>,

    /// The pattern name (left side of assignment)
    pub identifier: ParsedIdentifier,

    /// The constraint expression (right side of assignment)
    pub expression: Arc<TypedExpression>,
}

impl TypedConstraint {
    /// Type-check a constraint from an Assignment
    /// Returns Err if the identifier is invalid
    pub fn try_from_ast_with_context(
        assignment: Arc<Assignment>,
        ctx: &mut ExpressionTranslationContext,
    ) -> Result<Self, Arc<TranslationError>> {
        // Validate that identifier is not a parse error (propagate error if it is)
        assignment.identifier.valid_ref()?;

        // Type-check the constraint expression (must be BOOLEAN)
        let expression = TypedExpression::from_ast_with_expected_type(
            assignment.expression.clone(),
            ctx,
            &ExactMatcher::of(BOOLEAN),
        );

        Ok(Self {
            ast: assignment.clone(),
            identifier: assignment.identifier.clone(),
            expression: Arc::new(expression),
        })
    }
}

// ============================================================================
// TypedSelection - Field selections (used in DISTINCT BY, DROP, etc.)
// ============================================================================

/// A type-checked selection (field identifier with schema validation)
#[derive(Debug, Clone)]
pub struct TypedSelection {
    /// Reference to the AST selection
    pub ast: Arc<Selection>,

    /// The resolved identifier (validated against schema) or error
    pub resolved: ResolvedSelection,
}

/// Result of resolving a selection against a schema
#[derive(Debug, Clone, From)]
pub enum ResolvedSelection {
    /// Successfully resolved - identifier exists in merged schema.
    Resolved(Identifier),

    /// Resolution failed - identifier not found or invalid
    Error(Arc<TranslationError>),
}

impl TypedSelection {
    /// Type-check a selection against an environment
    ///
    /// Validates that the identifier exists in the schema.
    pub fn from_ast_with_context(
        selection: Arc<Selection>,
        input_schema: &Arc<TypeEnvironment>,
        ctx: &mut StatementTranslationContext,
    ) -> Self {
        let resolved = match selection.identifier.clone().valid() {
            Ok(identifier) => {
                // Validate that the field exists in the input schema
                if input_schema.lookup(&identifier).is_some() {
                    ResolvedSelection::Resolved(identifier.clone())
                } else {
                    let err = ctx
                        .error(format!("Field '{:?}' not found", identifier))
                        .at(selection.as_ref())
                        .emit();
                    ResolvedSelection::Error(err)
                }
            }
            Err(err) => ResolvedSelection::Error(err),
        };

        Self {
            ast: selection,
            resolved,
        }
    }
}
