package gudusoft.gsqlparser.resolver2.expansion;

import gudusoft.gsqlparser.nodes.TObjectName;
import gudusoft.gsqlparser.nodes.TResultColumn;
import gudusoft.gsqlparser.nodes.TResultColumnList;
import gudusoft.gsqlparser.resolver2.ScopeBuildResult;
import gudusoft.gsqlparser.resolver2.model.ScopeChild;
import gudusoft.gsqlparser.resolver2.namespace.INamespace;
import gudusoft.gsqlparser.resolver2.scope.IScope;
import gudusoft.gsqlparser.resolver2.scope.SelectScope;
import gudusoft.gsqlparser.resolver2.scope.FromScope;
import gudusoft.gsqlparser.resolver2.scope.CTEScope;
import gudusoft.gsqlparser.resolver2.matcher.INameMatcher;
import gudusoft.gsqlparser.resolver2.matcher.DefaultNameMatcher;
import gudusoft.gsqlparser.stmt.TSelectSqlStatement;

import java.util.*;

/**
 * Resolves star columns by pushing down column references from outer queries.
 *
 * <p>This is the GSP Java implementation of the star column push-down mechanism.
 * When an outer query references columns from a CTE or subquery that uses SELECT *,
 * we need to "push down" those column references to infer which columns are needed.
 *
 * <p>Example scenario:
 * <pre>
 * WITH my_cte AS (SELECT * FROM Employees)
 * SELECT id, name FROM my_cte;
 * </pre>
 *
 * <p>Push-down process:
 * 1. my_cte has SELECT * (star column)
 * 2. Outer query references id, name from my_cte
 * 3. Push down: id and name are "required" columns from the star
 * 4. These columns are inferred to come from Employees table
 *
 * <p>Key features:
 * - Works with ScopeBuildResult from ScopeBuilder
 * - Multi-level push-down (nested CTEs/subqueries)
 * - Iterative resolution (may need multiple passes)
 * - Confidence scoring based on reference type
 */
public class StarPushDownResolver {

    /**
     * Represents a column reference that needs to be pushed down.
     */
    public static class PushDownRequest {
        /** The column name being referenced */
        public final String columnName;

        /** The column reference node */
        public final TObjectName columnReference;

        /** The namespace containing the star column */
        public final INamespace starNamespace;

        /** The star column (TResultColumn with *) */
        public final TResultColumn starColumn;

        /** Confidence that this push-down is correct */
        public final double confidence;

        /** Evidence for this push-down */
        public final String evidence;

        public PushDownRequest(
                String columnName,
                TObjectName columnReference,
                INamespace starNamespace,
                TResultColumn starColumn,
                double confidence,
                String evidence) {
            this.columnName = columnName;
            this.columnReference = columnReference;
            this.starNamespace = starNamespace;
            this.starColumn = starColumn;
            this.confidence = confidence;
            this.evidence = evidence;
        }

        @Override
        public String toString() {
            return String.format("PushDown[%s -> %s, conf=%.2f]",
                columnName, starNamespace.getDisplayName(), confidence);
        }
    }

    /**
     * Represents the result of push-down resolution.
     */
    public static class PushDownResult {
        /** Namespace -> list of columns pushed down to it */
        private final Map<INamespace, Set<String>> pushedDownColumns = new LinkedHashMap<>();

        /** Namespace -> list of column references from outer queries */
        private final Map<INamespace, List<TObjectName>> upLevelReferences = new LinkedHashMap<>();

        /** All processed requests */
        private final List<PushDownRequest> processedRequests = new ArrayList<>();

        /** Number of iterations performed */
        private int iterations = 0;

        public void addPushedDownColumn(INamespace namespace, String columnName) {
            pushedDownColumns.computeIfAbsent(namespace, k -> new LinkedHashSet<>()).add(columnName);
        }

        public void addUpLevelReference(INamespace namespace, TObjectName reference) {
            upLevelReferences.computeIfAbsent(namespace, k -> new ArrayList<>()).add(reference);
        }

        public void addProcessedRequest(PushDownRequest request) {
            processedRequests.add(request);
        }

        public Set<String> getPushedDownColumns(INamespace namespace) {
            return pushedDownColumns.getOrDefault(namespace, Collections.emptySet());
        }

        public List<TObjectName> getUpLevelReferences(INamespace namespace) {
            return upLevelReferences.getOrDefault(namespace, Collections.emptyList());
        }

        public int getTotalPushedDownColumns() {
            int count = 0;
            for (Set<String> columns : pushedDownColumns.values()) {
                count += columns.size();
            }
            return count;
        }

        public int getStarNamespaceCount() {
            return pushedDownColumns.size();
        }

        public int getIterations() {
            return iterations;
        }

        void setIterations(int iterations) {
            this.iterations = iterations;
        }

        public List<PushDownRequest> getProcessedRequests() {
            return Collections.unmodifiableList(processedRequests);
        }

        @Override
        public String toString() {
            return String.format(
                "PushDownResult[namespaces=%d, columns=%d, iterations=%d]",
                pushedDownColumns.size(),
                getTotalPushedDownColumns(),
                iterations
            );
        }
    }

    /** Maximum number of iterations for push-down resolution */
    private final int maxIterations;

    /** Whether to enable debug logging */
    private final boolean debugLogging;

    /** Name matcher for column comparison */
    private final INameMatcher nameMatcher;

    public StarPushDownResolver() {
        this(10, false, new DefaultNameMatcher());
    }

    public StarPushDownResolver(int maxIterations, boolean debugLogging) {
        this(maxIterations, debugLogging, new DefaultNameMatcher());
    }

    public StarPushDownResolver(int maxIterations, boolean debugLogging, INameMatcher nameMatcher) {
        this.maxIterations = maxIterations;
        this.debugLogging = debugLogging;
        this.nameMatcher = nameMatcher != null ? nameMatcher : new DefaultNameMatcher();
    }

    /**
     * Resolve star columns by pushing down column references.
     *
     * @param scopeBuildResult the scope build result from ScopeBuilder
     * @return push-down resolution result
     */
    public PushDownResult resolve(ScopeBuildResult scopeBuildResult) {
        PushDownResult result = new PushDownResult();

        if (scopeBuildResult == null || scopeBuildResult.getGlobalScope() == null) {
            return result;
        }

        // Step 1: Collect all namespaces with star columns
        List<StarNamespaceInfo> starNamespaces = collectStarNamespaces(scopeBuildResult);
        log("Found " + starNamespaces.size() + " namespaces with star columns");

        if (starNamespaces.isEmpty()) {
            return result;
        }

        // Step 2: Iteratively collect push-down requests and process them
        int iteration = 0;
        int newColumnsFound;

        do {
            newColumnsFound = 0;
            iteration++;

            // Collect all column references that target star namespaces
            List<PushDownRequest> requests = collectPushDownRequests(
                scopeBuildResult, starNamespaces, result);

            log("Iteration " + iteration + ": Found " + requests.size() + " push-down requests");

            // Process each request
            for (PushDownRequest request : requests) {
                if (processPushDownRequest(request, result)) {
                    newColumnsFound++;
                }
            }

            log("Iteration " + iteration + ": Added " + newColumnsFound + " new columns");

        } while (newColumnsFound > 0 && iteration < maxIterations);

        result.setIterations(iteration);

        // Step 3: Apply inferred columns to dynamic namespaces
        applyInferredColumns(result);

        log("Push-down complete: " + result);
        return result;
    }

    /**
     * Information about a namespace with star column.
     */
    private static class StarNamespaceInfo {
        final INamespace namespace;
        final TResultColumn starColumn;
        final IScope containingScope;
        final String alias;

        StarNamespaceInfo(INamespace namespace, TResultColumn starColumn,
                         IScope containingScope, String alias) {
            this.namespace = namespace;
            this.starColumn = starColumn;
            this.containingScope = containingScope;
            this.alias = alias;
        }
    }

    /**
     * Collect all namespaces that have star columns.
     */
    private List<StarNamespaceInfo> collectStarNamespaces(ScopeBuildResult scopeBuildResult) {
        List<StarNamespaceInfo> result = new ArrayList<>();

        // Traverse all scopes
        collectStarNamespacesFromScope(scopeBuildResult.getGlobalScope(), result);

        // Also check SelectScopes from statementScopeMap
        for (SelectScope selectScope : scopeBuildResult.getStatementScopeMap().values()) {
            if (selectScope.getFromScope() != null) {
                collectStarNamespacesFromFromScope(selectScope.getFromScope(), selectScope, result);
            }
        }

        return result;
    }

    /**
     * Recursively collect star namespaces from a scope.
     */
    private void collectStarNamespacesFromScope(IScope scope, List<StarNamespaceInfo> result) {
        if (scope == null) {
            return;
        }

        // Check children of this scope
        for (ScopeChild child : scope.getChildren()) {
            INamespace ns = child.getNamespace();
            if (ns != null && ns.hasStarColumn()) {
                TResultColumn starCol = findStarColumn(ns);
                result.add(new StarNamespaceInfo(ns, starCol, scope, child.getAlias()));
            }
        }

        // Recurse into child scopes
        if (scope instanceof SelectScope) {
            SelectScope selectScope = (SelectScope) scope;
            if (selectScope.getFromScope() != null) {
                collectStarNamespacesFromScope(selectScope.getFromScope(), result);
            }
        }
    }

    /**
     * Collect star namespaces from a FromScope.
     */
    private void collectStarNamespacesFromFromScope(FromScope fromScope,
                                                     SelectScope parentSelect,
                                                     List<StarNamespaceInfo> result) {
        for (ScopeChild child : fromScope.getChildren()) {
            INamespace ns = child.getNamespace();
            if (ns != null && ns.hasStarColumn()) {
                // Avoid duplicates
                boolean exists = result.stream()
                    .anyMatch(info -> info.namespace == ns);
                if (!exists) {
                    TResultColumn starCol = findStarColumn(ns);
                    result.add(new StarNamespaceInfo(ns, starCol, parentSelect, child.getAlias()));
                }
            }
        }
    }

    /**
     * Find the star column in a namespace's SELECT statement.
     */
    private TResultColumn findStarColumn(INamespace namespace) {
        TSelectSqlStatement select = namespace.getSelectStatement();
        if (select == null || select.getResultColumnList() == null) {
            return null;
        }

        TResultColumnList resultList = select.getResultColumnList();
        for (int i = 0; i < resultList.size(); i++) {
            TResultColumn rc = resultList.getResultColumn(i);
            if (rc != null && rc.toString().endsWith("*")) {
                return rc;
            }
        }
        return null;
    }

    /**
     * Collect push-down requests by finding column references that target star namespaces.
     */
    private List<PushDownRequest> collectPushDownRequests(
            ScopeBuildResult scopeBuildResult,
            List<StarNamespaceInfo> starNamespaces,
            PushDownResult currentResult) {

        List<PushDownRequest> requests = new ArrayList<>();

        // For each column reference in the build result
        for (TObjectName colRef : scopeBuildResult.getAllColumnReferences()) {
            // Get the scope where this column appears
            IScope colScope = scopeBuildResult.getScopeForColumn(colRef);
            if (colScope == null) {
                continue;
            }

            // Extract table prefix and column name
            String prefix = getTablePrefix(colRef);
            String columnName = colRef.getColumnNameOnly();

            if (columnName == null || columnName.isEmpty() || columnName.equals("*")) {
                continue;  // Skip star columns themselves
            }

            // Check if this column reference targets any star namespace
            for (StarNamespaceInfo starInfo : starNamespaces) {
                if (matchesStarNamespace(colRef, prefix, starInfo, colScope)) {
                    // Check if already processed
                    Set<String> existing = currentResult.getPushedDownColumns(starInfo.namespace);
                    boolean alreadyExists = existing.stream()
                        .anyMatch(c -> nameMatcher.matches(c, columnName));

                    if (!alreadyExists) {
                        double confidence = calculateConfidence(colRef, prefix, starInfo);
                        String evidence = buildEvidence(colRef, starInfo);

                        requests.add(new PushDownRequest(
                            columnName,
                            colRef,
                            starInfo.namespace,
                            starInfo.starColumn,
                            confidence,
                            evidence
                        ));
                    }
                }
            }
        }

        return requests;
    }

    /**
     * Get the table prefix from a column reference (e.g., "t" from "t.col").
     */
    private String getTablePrefix(TObjectName colRef) {
        if (colRef == null) {
            return null;
        }

        // Try to get the table/schema part
        String fullName = colRef.toString();
        String columnOnly = colRef.getColumnNameOnly();

        if (fullName != null && columnOnly != null && fullName.contains(".")) {
            int lastDot = fullName.lastIndexOf('.');
            if (lastDot > 0) {
                return fullName.substring(0, lastDot);
            }
        }

        return null;
    }

    /**
     * Check if a column reference matches a star namespace.
     */
    private boolean matchesStarNamespace(TObjectName colRef, String prefix,
                                         StarNamespaceInfo starInfo, IScope colScope) {
        // If column has a prefix, it must match the namespace alias/name
        if (prefix != null && !prefix.isEmpty()) {
            String nsName = starInfo.alias;
            if (nsName == null) {
                nsName = starInfo.namespace.getDisplayName();
            }

            // Check if prefix matches namespace name (case-insensitive)
            if (!nameMatcher.matches(prefix, nsName)) {
                // Also check without schema prefix (e.g., "schema.table" vs "table")
                if (prefix.contains(".")) {
                    String shortPrefix = prefix.substring(prefix.lastIndexOf('.') + 1);
                    if (!nameMatcher.matches(shortPrefix, nsName)) {
                        return false;
                    }
                } else {
                    return false;
                }
            }
        }

        // Verify scope relationship: column's scope should be able to "see" the star namespace
        return canAccessNamespace(colScope, starInfo);
    }

    /**
     * Check if a scope can access a star namespace.
     */
    private boolean canAccessNamespace(IScope colScope, StarNamespaceInfo starInfo) {
        // Walk up the scope chain from colScope
        IScope current = colScope;
        while (current != null) {
            // Check if this scope contains the star namespace
            if (current == starInfo.containingScope) {
                return true;
            }

            // Check if this is a SelectScope with FromScope containing the namespace
            if (current instanceof SelectScope) {
                SelectScope selectScope = (SelectScope) current;
                if (selectScope.getFromScope() != null) {
                    for (ScopeChild child : selectScope.getFromScope().getChildren()) {
                        if (child.getNamespace() == starInfo.namespace) {
                            return true;
                        }
                    }
                }
            }

            // Check CTE scopes
            if (current instanceof CTEScope) {
                CTEScope cteScope = (CTEScope) current;
                // CTEs are accessible from this scope and below
                for (ScopeChild child : cteScope.getChildren()) {
                    if (child.getNamespace() == starInfo.namespace) {
                        return true;
                    }
                }
            }

            current = current.getParent();
        }

        return false;
    }

    /**
     * Calculate confidence score for a push-down request.
     */
    private double calculateConfidence(TObjectName colRef, String prefix, StarNamespaceInfo starInfo) {
        // Higher confidence for qualified references (t.col)
        if (prefix != null && !prefix.isEmpty()) {
            return 0.95;
        }

        // Lower confidence for unqualified references
        return 0.7;
    }

    /**
     * Build evidence string for a push-down request.
     */
    private String buildEvidence(TObjectName colRef, StarNamespaceInfo starInfo) {
        return String.format("Column '%s' referenced from outer query, targets '%s' with SELECT *",
            colRef.toString(), starInfo.namespace.getDisplayName());
    }

    /**
     * Process a single push-down request.
     *
     * @param request the push-down request
     * @param result the result to update
     * @return true if a new column was added, false otherwise
     */
    private boolean processPushDownRequest(PushDownRequest request, PushDownResult result) {
        if (request == null || request.starNamespace == null) {
            return false;
        }

        Set<String> existingColumns = result.getPushedDownColumns(request.starNamespace);

        // Check if this column is already pushed down (case-insensitive)
        for (String existing : existingColumns) {
            if (nameMatcher.matches(existing, request.columnName)) {
                return false;  // Already have this column
            }
        }

        // Add new pushed-down column
        result.addPushedDownColumn(request.starNamespace, request.columnName);
        result.addUpLevelReference(request.starNamespace, request.columnReference);
        result.addProcessedRequest(request);

        log("Pushed down: " + request.columnName + " -> " + request.starNamespace.getDisplayName());

        return true;
    }

    /**
     * Apply inferred columns to dynamic namespaces.
     */
    private void applyInferredColumns(PushDownResult result) {
        for (Map.Entry<INamespace, Set<String>> entry : result.pushedDownColumns.entrySet()) {
            INamespace namespace = entry.getKey();
            Set<String> columns = entry.getValue();

            // If namespace supports dynamic inference, add inferred columns
            if (namespace.supportsDynamicInference()) {
                for (String columnName : columns) {
                    namespace.addInferredColumn(columnName, 0.9, "star_push_down");
                }
                log("Applied " + columns.size() + " inferred columns to " + namespace.getDisplayName());
            }
        }
    }

    /**
     * Get statistics about push-down resolution.
     */
    public Map<String, Object> getStatistics(PushDownResult result) {
        Map<String, Object> stats = new LinkedHashMap<>();
        stats.put("star_namespaces", result.getStarNamespaceCount());
        stats.put("total_pushed_down", result.getTotalPushedDownColumns());
        stats.put("iterations", result.getIterations());
        stats.put("requests_processed", result.getProcessedRequests().size());
        return stats;
    }

    private void log(String message) {
        if (debugLogging) {
            System.out.println("[StarPushDownResolver] " + message);
        }
    }
}