use super::{
    AutoPrefixSqlNode, CaseSqlNode, EvaluateSqlNode, FinalMeasureSqlNode,
    FinalPreAggregationMeasureSqlNode, GeoDimensionSqlNode, MaskedSqlNode, MeasureFilterSqlNode,
    MeasureRenderModifierSqlNode, MultiStageRankNode, MultiStageWindowNode, ParenthesizeSqlNode,
    RenderReferencesSqlNode, RenderReferencesType, RollingWindowNode, RootSqlNode,
    SegmentDimensionSqlNode, SqlNode, TimeDimensionNode, TimeShiftSqlNode, UngroupedMeasureSqlNode,
    UngroupedQueryFinalMeasureSqlNode,
};
use crate::physical_plan::cube_ref_evaluator::CubeRefEvaluator;
use crate::physical_plan::sql_nodes::calendar_time_shift::CalendarTimeShiftSqlNode;
use crate::physical_plan::sql_nodes::RenderReferences;
use crate::planner::planners::multi_stage::TimeShiftState;
use crate::planner::query_tools::QueryTools;
use crate::planner::symbols::CalendarDimensionTimeShift;
use std::collections::{HashMap, HashSet};
use std::rc::Rc;

/// Builds the SQL-node chain for a query. Carries the reference maps
/// the query needs (time shifts, render references, pre-aggregation
/// refs, cube aliases) and assembles them into a layered `SqlNode`
/// via `default_node_processor`.
#[derive(Clone, Default)]
pub struct SqlNodesFactory {
    time_shifts: TimeShiftState,
    calendar_time_shifts: HashMap<String, CalendarDimensionTimeShift>,
    render_references: RenderReferences,
    pre_aggregation_dimensions_references: RenderReferences,
    pre_aggregation_measures_references: RenderReferences,
    ungrouped_measure_references: RenderReferences,
    cube_name_references: HashMap<String, String>,
    use_local_tz_in_date_range: bool,
    original_sql_pre_aggregations: HashMap<String, String>,
    // Full names of the members present in the query GROUP BY. Used by
    // MaskedSqlNode to decide whether conditional masking can be applied to an
    // aggregate measure.
    group_by_members: HashSet<String>,
}

impl SqlNodesFactory {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn set_time_shifts(&mut self, time_shifts: TimeShiftState) {
        self.time_shifts = time_shifts;
    }

    pub fn time_shifts(&self) -> &TimeShiftState {
        &self.time_shifts
    }

    pub fn set_calendar_time_shifts(
        &mut self,
        calendar_time_shifts: HashMap<String, CalendarDimensionTimeShift>,
    ) {
        self.calendar_time_shifts = calendar_time_shifts;
    }

    pub fn set_use_local_tz_in_date_range(&mut self, value: bool) {
        self.use_local_tz_in_date_range = value;
    }

    pub fn set_group_by_members(&mut self, value: HashSet<String>) {
        self.group_by_members = value;
    }

    pub fn use_local_tz_in_date_range(&self) -> bool {
        self.use_local_tz_in_date_range
    }

    pub fn reading_pre_aggregation(&self) -> bool {
        !self.pre_aggregation_dimensions_references.is_empty()
    }

    pub fn add_render_reference<T: Into<RenderReferencesType>>(&mut self, name: String, value: T) {
        self.render_references.insert(name, value);
    }

    pub fn render_references(&self) -> &RenderReferences {
        &self.render_references
    }

    pub fn clear_render_references(&mut self) {
        self.render_references = RenderReferences::default();
    }

    pub fn render_references_mut(&mut self) -> &mut RenderReferences {
        &mut self.render_references
    }

    pub fn add_pre_aggregation_dimension_reference<T: Into<RenderReferencesType>>(
        &mut self,
        name: String,
        value: T,
    ) {
        self.pre_aggregation_dimensions_references
            .insert(name, value);
    }

    pub fn set_original_sql_pre_aggregations(&mut self, value: HashMap<String, String>) {
        self.original_sql_pre_aggregations = value;
    }

    pub fn add_pre_aggregation_measure_reference<T: Into<RenderReferencesType>>(
        &mut self,
        name: String,
        value: T,
    ) {
        self.pre_aggregation_measures_references.insert(name, value);
    }

    pub fn add_ungrouped_measure_reference<T: Into<RenderReferencesType>>(
        &mut self,
        name: String,
        value: T,
    ) {
        self.ungrouped_measure_references.insert(name, value);
    }

    pub fn set_cube_name_references(&mut self, value: HashMap<String, String>) {
        self.cube_name_references = value;
    }

    pub fn add_cube_name_reference(&mut self, key: String, value: String) {
        self.cube_name_references.insert(key, value);
    }

    pub fn cube_ref_evaluator(&self) -> CubeRefEvaluator {
        CubeRefEvaluator::new(
            self.cube_name_references.clone(),
            self.original_sql_pre_aggregations.clone(),
        )
    }

    /// Assembles the full SQL-node chain for the configured options.
    /// Three sub-chains hang off a `RootSqlNode` keyed by member
    /// kind: a dimension chain (geo / case / time-shift / calendar
    /// time-shift wraps), a time-dimension chain, and a measure
    /// chain (case → measure filter → render-modifier dispatch over
    /// the final-measure / rolling-merge / ungrouped chains → mask →
    /// multi-stage window and rank wraps). The whole tree is then
    /// wrapped in a top-level `RenderReferencesSqlNode` for
    /// query-wide reference substitution.
    pub fn default_node_processor(&self, query_tools: &QueryTools) -> Rc<dyn SqlNode> {
        // Build an "unmasked" copy of the tree (masking disabled, but still
        // dispatching by member kind) only when the query has masked members. It
        // is handed to the masked nodes so they can render mask-filter member
        // references through it — routing dimensions through the dimension chain
        // and avoiding mask recursion. For queries with no masked members
        // `MaskedSqlNode::resolve_mask` short-circuits and never dereferences the
        // unmasked root, so building it would be pure waste; passing `None` falls
        // back to the masked node's own input (see masked.rs).
        let unmasked_root = if query_tools.has_masked_members() {
            Some(self.build_node_processor(true, None))
        } else {
            None
        };
        self.build_node_processor(false, unmasked_root)
    }

    fn build_node_processor(
        &self,
        skip_masking: bool,
        unmasked_root: Option<Rc<dyn SqlNode>>,
    ) -> Rc<dyn SqlNode> {
        let evaluate_sql_processor = MaskedSqlNode::new(
            EvaluateSqlNode::new(),
            false,
            self.group_by_members.clone(),
            skip_masking,
            unmasked_root.clone(),
        );
        let auto_prefix_processor = AutoPrefixSqlNode::new(
            evaluate_sql_processor.clone(),
            self.cube_name_references.clone(),
        );
        let parenthesize_processor: Rc<dyn SqlNode> =
            ParenthesizeSqlNode::new(auto_prefix_processor.clone());

        let measure_filter_processor = MeasureFilterSqlNode::new(parenthesize_processor.clone());
        let measure_processor = CaseSqlNode::new(measure_filter_processor.clone());

        let measure_processor = self.add_ungrouped_measure_reference_if_needed(measure_processor);
        let measure_processor = self.final_measure_node_processor(measure_processor);
        // Wrap the entire measure chain with MaskedSqlNode so masked measures
        // are intercepted before aggregation/ungrouped wrapping.
        let measure_processor = MaskedSqlNode::new(
            measure_processor,
            true,
            self.group_by_members.clone(),
            skip_masking,
            unmasked_root.clone(),
        );
        let measure_processor: Rc<dyn SqlNode> =
            MultiStageWindowNode::new(measure_filter_processor.clone(), measure_processor);
        let measure_processor: Rc<dyn SqlNode> = MultiStageRankNode::new(measure_processor);

        let default_processor: Rc<dyn SqlNode> =
            if !self.pre_aggregation_dimensions_references.is_empty() {
                // Reading from a pre-aggregation: members are plain column refs,
                // so a segment is already a stored column — no wrapping.
                RenderReferencesSqlNode::new(
                    evaluate_sql_processor.clone(),
                    self.pre_aggregation_dimensions_references.clone(),
                )
            } else {
                // Building/evaluating the expression: wrap segment dimensions per
                // dialect so a boolean is a valid projected value.
                SegmentDimensionSqlNode::new(evaluate_sql_processor.clone())
            };
        let default_processor: Rc<dyn SqlNode> = ParenthesizeSqlNode::new(default_processor);

        let root_node = RootSqlNode::new(
            self.dimension_processor(evaluate_sql_processor.clone()),
            self.time_dimension_processor(ParenthesizeSqlNode::new(evaluate_sql_processor.clone())),
            measure_processor.clone(),
            default_processor,
        );
        RenderReferencesSqlNode::new(root_node, self.render_references.clone())
    }

    /// When an ungrouped query reads from a pre-aggregation, a measure must
    /// resolve to its stored rollup column. Unlike the grouped case, the column
    /// already holds the aggregated value and is returned as-is, so it's a plain
    /// column reference with no `sum()` wrap. Wrapping the ungrouped node keeps
    /// the reference outermost so the measure is intercepted before it would
    /// otherwise re-render its base-table SQL.
    fn wrap_ungrouped_pre_aggregation_measure(&self, node: Rc<dyn SqlNode>) -> Rc<dyn SqlNode> {
        if !self.pre_aggregation_measures_references.is_empty() {
            RenderReferencesSqlNode::new(node, self.pre_aggregation_measures_references.clone())
        } else {
            node
        }
    }

    fn add_ungrouped_measure_reference_if_needed(
        &self,
        default: Rc<dyn SqlNode>,
    ) -> Rc<dyn SqlNode> {
        if !self.ungrouped_measure_references.is_empty() {
            RenderReferencesSqlNode::new(default, self.ungrouped_measure_references.clone())
        } else {
            default
        }
    }

    fn final_measure_node_processor(&self, input: Rc<dyn SqlNode>) -> Rc<dyn SqlNode> {
        let aggregated: Rc<dyn SqlNode> = {
            let final_processor: Rc<dyn SqlNode> = FinalMeasureSqlNode::new(input.clone());
            if !self.pre_aggregation_measures_references.is_empty() {
                FinalPreAggregationMeasureSqlNode::new(
                    final_processor,
                    self.pre_aggregation_measures_references.clone(),
                )
            } else {
                final_processor
            }
        };
        let rolling_merge = RollingWindowNode::new(input.clone(), aggregated.clone());
        let raw_value = self
            .wrap_ungrouped_pre_aggregation_measure(UngroupedMeasureSqlNode::new(input.clone()));
        let ungrouped_final = self
            .wrap_ungrouped_pre_aggregation_measure(UngroupedQueryFinalMeasureSqlNode::new(input));
        MeasureRenderModifierSqlNode::new(aggregated, rolling_merge, raw_value, ungrouped_final)
    }

    fn dimension_processor(&self, input: Rc<dyn SqlNode>) -> Rc<dyn SqlNode> {
        let input = if !self.pre_aggregation_dimensions_references.is_empty() {
            RenderReferencesSqlNode::new(input, self.pre_aggregation_dimensions_references.clone())
        } else {
            let input: Rc<dyn SqlNode> = GeoDimensionSqlNode::new(input);
            let input: Rc<dyn SqlNode> = CaseSqlNode::new(input);
            input
        };

        let input: Rc<dyn SqlNode> =
            AutoPrefixSqlNode::new(input, self.cube_name_references.clone());

        let input: Rc<dyn SqlNode> = ParenthesizeSqlNode::new(input);

        let input: Rc<dyn SqlNode> = TimeDimensionNode::new(input);

        let input = if !self.calendar_time_shifts.is_empty() {
            CalendarTimeShiftSqlNode::new(self.calendar_time_shifts.clone(), input)
        } else {
            input
        };

        let input = if !self.time_shifts.is_empty() {
            TimeShiftSqlNode::new(
                self.time_shifts.clone(),
                self.pre_aggregation_dimensions_references.clone(),
                input,
            )
        } else {
            input
        };

        input
    }

    fn time_dimension_processor(&self, input: Rc<dyn SqlNode>) -> Rc<dyn SqlNode> {
        let input: Rc<dyn SqlNode> = TimeDimensionNode::new(input);

        input
    }
}
