package gudusoft.gsqlparser.ir.semantic.diff;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Pure comparison: two {@link CanonicalLineageModel}s in, one
 * {@link DivergenceReport} out. Pass order matters and is documented
 * inline below — see the slice-7 plan §"Comparison passes".
 */
public final class DivergenceReporter {

    private DivergenceReporter() {}

    /**
     * @param sqlName  human-readable name (e.g. "01_alias_collision")
     * @param ir       result of projecting the Semantic IR
     * @param dlineage result of projecting the dlineage XML
     */
    public static DivergenceReport report(String sqlName, ProjectorResult ir, ProjectorResult dlineage) {
        if (sqlName == null || sqlName.isEmpty()) {
            throw new IllegalArgumentException("sqlName must be non-empty");
        }
        if (ir == null || dlineage == null) {
            throw new IllegalArgumentException("ir and dlineage results must not be null");
        }

        // Unsupported-side short-circuit. We still produce both models in the
        // report (empty for the failing side) so the JSON shape is uniform.
        List<Divergence> divs = new ArrayList<>();
        if (!ir.isSupported()) {
            divs.add(new Divergence(DivergenceClass.UNSUPPORTED_BY_IR, Divergence.QUERY_WIDE,
                    detailFor(ir)));
        }
        if (!dlineage.isSupported()) {
            divs.add(new Divergence(DivergenceClass.UNSUPPORTED_BY_DLINEAGE, Divergence.QUERY_WIDE,
                    detailFor(dlineage)));
        }
        if (!ir.isSupported() || !dlineage.isSupported()) {
            // When either side is unsupported the per-edge passes don't make
            // sense; the query-wide divergence is the only one reported.
            return new DivergenceReport(sqlName, ir.getModel(), dlineage.getModel(),
                    sortDivergences(divs));
        }

        CanonicalLineageModel a = ir.getModel();
        CanonicalLineageModel b = dlineage.getModel();

        // Pass 1: output-presence symmetric difference.
        // Drop SELECT edges for one-sided outputs so later passes don't double-count.
        Set<String> aOnly = new LinkedHashSet<>(a.getOutputNames());
        aOnly.removeAll(b.getOutputNames());
        Set<String> bOnly = new LinkedHashSet<>(b.getOutputNames());
        bOnly.removeAll(a.getOutputNames());
        for (String n : aOnly) {
            divs.add(new Divergence(DivergenceClass.IR_EXTRA_DEPENDENCY, n,
                    Divergence.DETAIL_OUTPUT_PRESENT));
        }
        for (String n : bOnly) {
            divs.add(new Divergence(DivergenceClass.IR_MISSING_DEPENDENCY, n,
                    Divergence.DETAIL_OUTPUT_PRESENT));
        }

        Set<String> commonOutputs = new HashSet<>(a.getOutputNames());
        commonOutputs.retainAll(b.getOutputNames());

        Set<CanonicalLineageEdge> aEdges = filterEdgesForCommonOutputs(a.getEdges(), commonOutputs);
        Set<CanonicalLineageEdge> bEdges = filterEdgesForCommonOutputs(b.getEdges(), commonOutputs);

        // Pass 2: aggregation pass over common outputs.
        for (String name : new java.util.TreeSet<>(commonOutputs)) {
            Boolean ia = a.getAggregateByOutput().get(name);
            Boolean da = b.getAggregateByOutput().get(name);
            if (ia == null || da == null) continue;
            if (!ia.equals(da)) {
                divs.add(new Divergence(DivergenceClass.AGGREGATION_MISMATCH, name,
                        "ir=" + ia + ",dlineage=" + da));
            }
        }

        // Working sets we'll whittle down with binding/role passes.
        Set<CanonicalLineageEdge> aRemaining = new LinkedHashSet<>(aEdges);
        Set<CanonicalLineageEdge> bRemaining = new LinkedHashSet<>(bEdges);

        // Pass 3: binding mismatch.
        // Group by (role, outputName, baseColumn). If one side has exactly
        // one edge and the other has exactly one edge with a different
        // baseTable, that's a single binding mismatch. Drop both edges.
        Map<String, List<CanonicalLineageEdge>> aByBindKey = groupBy(aRemaining, DivergenceReporter::bindingKey);
        Map<String, List<CanonicalLineageEdge>> bByBindKey = groupBy(bRemaining, DivergenceReporter::bindingKey);
        for (String key : new java.util.TreeSet<>(union(aByBindKey.keySet(), bByBindKey.keySet()))) {
            List<CanonicalLineageEdge> al = aByBindKey.getOrDefault(key, Collections.emptyList());
            List<CanonicalLineageEdge> bl = bByBindKey.getOrDefault(key, Collections.emptyList());
            if (al.size() == 1 && bl.size() == 1) {
                CanonicalLineageEdge ae = al.get(0);
                CanonicalLineageEdge be = bl.get(0);
                if (!ae.getBaseTable().equals(be.getBaseTable())) {
                    divs.add(new Divergence(DivergenceClass.BINDING_MISMATCH,
                            ae.getOutputName(),
                            ae.getRole() + " " + ae.getBaseColumn()
                                    + " ir=" + ae.getBaseTable()
                                    + ",dlineage=" + be.getBaseTable()));
                    aRemaining.remove(ae);
                    bRemaining.remove(be);
                }
            }
        }

        // Pass 4: role mismatch (FILTER vs JOIN).
        // Group by (outputName, baseTable, baseColumn) without role.
        Map<String, List<CanonicalLineageEdge>> aByRoleKey = groupBy(aRemaining, DivergenceReporter::roleAgnosticKey);
        Map<String, List<CanonicalLineageEdge>> bByRoleKey = groupBy(bRemaining, DivergenceReporter::roleAgnosticKey);
        for (String key : new java.util.TreeSet<>(union(aByRoleKey.keySet(), bByRoleKey.keySet()))) {
            List<CanonicalLineageEdge> al = aByRoleKey.getOrDefault(key, Collections.emptyList());
            List<CanonicalLineageEdge> bl = bByRoleKey.getOrDefault(key, Collections.emptyList());
            if (al.size() == 1 && bl.size() == 1) {
                CanonicalLineageEdge ae = al.get(0);
                CanonicalLineageEdge be = bl.get(0);
                if (ae.getRole() != be.getRole()) {
                    divs.add(new Divergence(DivergenceClass.FILTER_OR_JOIN_SCOPE_MISMATCH,
                            ae.getOutputName(),
                            ae.getBaseTable() + "." + ae.getBaseColumn()
                                    + " ir=" + ae.getRole() + ",dlineage=" + be.getRole()));
                    aRemaining.remove(ae);
                    bRemaining.remove(be);
                }
            }
        }

        // Pass 4b (US-010): semantic-role comparison over identity-matched
        // edges. Equality ignores the fine-grained semanticRole, so an edge
        // present on both sides may still disagree about it. Only fires when
        // BOTH sides carry a non-null role — projectors without role
        // information (slice-7 SemanticIRProjector / DlineageXmlProjector)
        // can never trigger it. Matched edges are not removed from the
        // remaining sets; pass 5's symmetric difference drops them anyway.
        Map<CanonicalLineageEdge, CanonicalLineageEdge> bByIdentity = new HashMap<>();
        for (CanonicalLineageEdge be : bRemaining) {
            bByIdentity.put(be, be);
        }
        List<CanonicalLineageEdge> aMatchOrder = new ArrayList<>(aRemaining);
        Collections.sort(aMatchOrder, CanonicalLineageEdge.ORDER);
        for (CanonicalLineageEdge ae : aMatchOrder) {
            CanonicalLineageEdge be = bByIdentity.get(ae);
            if (be == null) continue;
            String ar = ae.getSemanticRole();
            String br = be.getSemanticRole();
            if (ar != null && br != null && !ar.equals(br)) {
                divs.add(new Divergence(DivergenceClass.ROLE_MISMATCH,
                        ae.getOutputName(),
                        ae.getRole() + " " + ae.getBaseTable() + "." + ae.getBaseColumn()
                                + " ir=" + ar + ",dlineage=" + br));
            }
        }

        // Pass 5: symmetric difference. Whatever is left is a flat
        // missing/extra divergence. Sort by canonical edge order so
        // emission is deterministic.
        List<CanonicalLineageEdge> aLeft = new ArrayList<>(aRemaining);
        aLeft.removeAll(bRemaining);
        List<CanonicalLineageEdge> bLeft = new ArrayList<>(bRemaining);
        bLeft.removeAll(aRemaining);
        Collections.sort(aLeft, CanonicalLineageEdge.ORDER);
        Collections.sort(bLeft, CanonicalLineageEdge.ORDER);
        for (CanonicalLineageEdge e : aLeft) {
            divs.add(new Divergence(DivergenceClass.IR_EXTRA_DEPENDENCY,
                    e.getOutputName(), edgeDetail(e)));
        }
        for (CanonicalLineageEdge e : bLeft) {
            divs.add(new Divergence(DivergenceClass.IR_MISSING_DEPENDENCY,
                    e.getOutputName(), edgeDetail(e)));
        }

        return new DivergenceReport(sqlName, a, b, sortDivergences(divs));
    }

    private static String detailFor(ProjectorResult r) {
        StringBuilder sb = new StringBuilder(r.getReason().name());
        if (r.getDetail() != null && !r.getDetail().isEmpty()) {
            sb.append(": ").append(r.getDetail());
        }
        return sb.toString();
    }

    private static Set<CanonicalLineageEdge> filterEdgesForCommonOutputs(
            Set<CanonicalLineageEdge> edges, Set<String> commonOutputs) {
        // SELECT edges keyed by an output not in both sides → drop (already
        // accounted for by output-presence pass). FILTER/JOIN edges have null
        // outputName and always pass through.
        Set<CanonicalLineageEdge> out = new LinkedHashSet<>();
        for (CanonicalLineageEdge e : edges) {
            if (e.getRole() == EdgeRole.SELECT && !commonOutputs.contains(e.getOutputName())) continue;
            out.add(e);
        }
        return out;
    }

    private static String edgeDetail(CanonicalLineageEdge e) {
        return e.getRole() + " " + e.getBaseTable() + "." + e.getBaseColumn();
    }

    private static String bindingKey(CanonicalLineageEdge e) {
        // Include role so SELECT bindings don't merge with FILTER bindings
        // for the same base column.
        return e.getRole().name() + "
"
                + (e.getOutputName() == null ? "" : e.getOutputName()) + "
"
                + e.getBaseColumn();
    }

    private static String roleAgnosticKey(CanonicalLineageEdge e) {
        return (e.getOutputName() == null ? "" : e.getOutputName()) + "
"
                + e.getBaseTable() + "
"
                + e.getBaseColumn();
    }

    private static Map<String, List<CanonicalLineageEdge>> groupBy(
            Set<CanonicalLineageEdge> edges,
            java.util.function.Function<CanonicalLineageEdge, String> keyFn) {
        Map<String, List<CanonicalLineageEdge>> out = new HashMap<>();
        for (CanonicalLineageEdge e : edges) {
            out.computeIfAbsent(keyFn.apply(e), k -> new ArrayList<>()).add(e);
        }
        return out;
    }

    private static Set<String> union(Set<String> a, Set<String> b) {
        Set<String> out = new HashSet<>(a);
        out.addAll(b);
        return out;
    }

    private static List<Divergence> sortDivergences(List<Divergence> divs) {
        List<Divergence> sorted = new ArrayList<>(divs);
        Collections.sort(sorted, Divergence.ORDER);
        return sorted;
    }
}