package com.agent.management.rag.structured;

import com.agent.management.model.entity.DataSource;
import com.agent.management.rag.capability.SqlSchemaSnapshot;
import com.agent.management.service.DataSourceService;
import com.agent.management.service.DynamicJdbcService;
import com.agent.management.service.SchemaExplorerService;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;

@Service
@RequiredArgsConstructor
public class StructuredRelationshipDiscoveryService {
    private static final int VALIDATION_SAMPLE_ROWS = 500;
    private static final int MAX_TARGETS_PER_SOURCE_COLUMN = 6;
    private static final int MAX_DISCRIMINATOR_VALUES = 20;

    private final DataSourceService dataSources;
    private final DynamicJdbcService jdbc;
    private final SchemaExplorerService explorer;

    public DiscoveryResult discover(Long dataSourceId, String catalog, String schema,
                                    SqlSchemaSnapshot snapshot,
                                    Map<String, List<SchemaExplorerService.ColumnSpec>> columnsByTable) {
        List<String> warnings = new ArrayList<>();
        List<StructuredRelationship> result = new ArrayList<>(physicalRelationships(snapshot));
        Metadata metadata;
        try {
            metadata = metadata(dataSourceId, catalog, schema, snapshot);
        } catch (Exception error) {
            warnings.add("relationship metadata unavailable: " + shortMessage(error));
            return new DiscoveryResult(result, warnings);
        }

        List<TargetColumn> targets = targetColumns(snapshot, columnsByTable, metadata.uniqueColumns());
        for (SqlSchemaSnapshot.TableSchema sourceTable : snapshot.tables()) {
            List<SchemaExplorerService.ColumnSpec> sourceColumns = columnsByTable.getOrDefault(sourceTable.name(), List.of());
            List<SchemaExplorerService.ColumnSpec> discriminators = sourceColumns.stream()
                    .filter(column -> isDiscriminator(column.name()) && isShortValueType(column.type())).limit(3).toList();
            for (SchemaExplorerService.ColumnSpec sourceColumn : sourceColumns) {
                if (!isRelationColumn(sourceColumn.name())) continue;
                List<SchemaExplorerService.ColumnSpec> matchingDiscriminators = discriminators.stream()
                        .filter(discriminator -> discriminatorStem(discriminator.name()).equals(stem(normalize(sourceColumn.name()))))
                        .toList();
                List<Candidate> candidates = targets.stream()
                        .filter(target -> !target.table().equals(sourceTable.name()))
                        .filter(target -> compatible(sourceColumn.type(), target.column().type()))
                        .map(target -> new Candidate(sourceTable.name(), sourceColumn, target,
                                nameScore(sourceTable.name(), sourceColumn, target)))
                        .filter(candidate -> candidate.nameScore() >= 0.12 || !matchingDiscriminators.isEmpty())
                        .sorted(Comparator.comparingDouble(Candidate::nameScore).reversed()
                                .thenComparing(candidate -> candidate.target().table()))
                        .limit(MAX_TARGETS_PER_SOURCE_COLUMN).toList();
                List<ScoredCandidate> scored = new ArrayList<>();
                for (Candidate candidate : candidates) {
                    try {
                        Coverage coverage = validate(dataSourceId, candidate, null, null);
                        double confidence = score(candidate.nameScore(), coverage);
                        if (confidence >= 0.58) scored.add(new ScoredCandidate(candidate, coverage, confidence));
                    } catch (Exception error) {
                        warnings.add("relationship validation " + sourceTable.name() + "." + sourceColumn.name()
                                + " failed: " + shortMessage(error));
                    }
                }
                markAmbiguity(scored);
                for (ScoredCandidate candidate : scored) {
                    boolean ambiguous = candidate.ambiguous();
                    if (!ambiguous && candidate.candidate().nameScore() < 0.20 && !matchingDiscriminators.isEmpty()) {
                        List<StructuredRelationship> conditional = conditionalRelationships(
                                dataSourceId, candidate, matchingDiscriminators, warnings);
                        if (!conditional.isEmpty()) {
                            result.addAll(conditional);
                            continue;
                        }
                    }
                    result.add(toRelationship(candidate, ambiguous
                            ? StructuredRelationship.RelationshipType.AMBIGUOUS
                            : StructuredRelationship.RelationshipType.INFERRED));
                }
            }
        }
        return new DiscoveryResult(deduplicate(result), warnings);
    }

    private Metadata metadata(Long dataSourceId, String catalog, String schema, SqlSchemaSnapshot snapshot) throws Exception {
        DataSource source = dataSources.getDecrypted(dataSourceId);
        Map<String, Set<String>> unique = new LinkedHashMap<>();
        try (Connection connection = jdbc.getConnection(source)) {
            DatabaseMetaData meta = connection.getMetaData();
            for (SqlSchemaSnapshot.TableSchema table : snapshot.tables()) {
                Set<String> columns = new LinkedHashSet<>();
                Map<String, List<String>> uniqueGroups = new LinkedHashMap<>();
                try (var keys = meta.getPrimaryKeys(emptyToNull(catalog), emptyToNull(schema), table.name())) {
                    while (keys.next()) uniqueGroups.computeIfAbsent(
                            "PK:" + value(keys.getString("PK_NAME"), table.name()), ignored -> new ArrayList<>())
                            .add(keys.getString("COLUMN_NAME"));
                }
                try (var indexes = meta.getIndexInfo(emptyToNull(catalog), emptyToNull(schema), table.name(), true, false)) {
                    while (indexes.next()) {
                        String column = indexes.getString("COLUMN_NAME");
                        String index = indexes.getString("INDEX_NAME");
                        if (column != null && index != null) uniqueGroups
                                .computeIfAbsent("UK:" + index, ignored -> new ArrayList<>()).add(column);
                    }
                }
                uniqueGroups.values().stream().filter(group -> group.size() == 1)
                        .map(group -> group.get(0)).forEach(columns::add);
                unique.put(table.name(), columns);
            }
        }
        return new Metadata(unique);
    }

    private static List<TargetColumn> targetColumns(SqlSchemaSnapshot snapshot,
                                                     Map<String, List<SchemaExplorerService.ColumnSpec>> columnsByTable,
                                                     Map<String, Set<String>> uniqueColumns) {
        List<TargetColumn> result = new ArrayList<>();
        for (SqlSchemaSnapshot.TableSchema table : snapshot.tables()) {
            Map<String, SchemaExplorerService.ColumnSpec> byName = new HashMap<>();
            columnsByTable.getOrDefault(table.name(), List.of()).forEach(column -> byName.put(column.name(), column));
            for (String name : uniqueColumns.getOrDefault(table.name(), Set.of())) {
                SchemaExplorerService.ColumnSpec column = byName.get(name);
                if (column != null && isRelationColumn(name)) result.add(new TargetColumn(table.name(), column));
            }
        }
        return result;
    }

    private Coverage validate(Long dataSourceId, Candidate candidate, String discriminator, Object value) {
        String sourceAlias = "s", targetAlias = "t";
        StringBuilder where = new StringBuilder(quote(candidate.sourceColumn().name())).append(" IS NOT NULL");
        if (discriminator != null) where.append(" AND ").append(quote(discriminator)).append(" = ").append(literal(value));
        String sql = "SELECT COUNT(*), COUNT(DISTINCT " + sourceAlias + "." + quote(candidate.sourceColumn().name())
                + "), SUM(CASE WHEN " + targetAlias + "." + quote(candidate.target().column().name())
                + " IS NULL THEN 1 ELSE 0 END) FROM (SELECT " + quote(candidate.sourceColumn().name())
                + " FROM " + quote(candidate.sourceTable()) + " WHERE " + where + " LIMIT " + VALIDATION_SAMPLE_ROWS
                + ") " + sourceAlias + " LEFT JOIN " + quote(candidate.target().table()) + " " + targetAlias
                + " ON " + sourceAlias + "." + quote(candidate.sourceColumn().name()) + " = " + targetAlias + "."
                + quote(candidate.target().column().name());
        SchemaExplorerService.QueryResult query = explorer.executeQuery(dataSourceId, sql, 1);
        if (query.rows().isEmpty()) return new Coverage(0, 0, 0);
        List<Object> row = query.rows().get(0);
        return new Coverage(number(row, 0), number(row, 1), number(row, 2));
    }

    private List<StructuredRelationship> conditionalRelationships(
            Long dataSourceId, ScoredCandidate candidate,
            List<SchemaExplorerService.ColumnSpec> discriminators, List<String> warnings) {
        for (SchemaExplorerService.ColumnSpec discriminator : discriminators) {
            try {
                String sql = "SELECT DISTINCT " + quote(discriminator.name()) + " FROM "
                        + quote(candidate.candidate().sourceTable()) + " WHERE " + quote(discriminator.name())
                        + " IS NOT NULL LIMIT " + (MAX_DISCRIMINATOR_VALUES + 1);
                SchemaExplorerService.QueryResult values = explorer.executeQuery(
                        dataSourceId, sql, MAX_DISCRIMINATOR_VALUES + 1);
                if (values.rows().size() > MAX_DISCRIMINATOR_VALUES) continue;
                List<StructuredRelationship> relations = new ArrayList<>();
                for (List<Object> row : values.rows()) {
                    if (row.isEmpty() || row.get(0) == null) continue;
                    Coverage coverage = validate(dataSourceId, candidate.candidate(), discriminator.name(), row.get(0));
                    double confidence = Math.min(1, score(candidate.candidate().nameScore(), coverage) + 0.08);
                    if (confidence < 0.78) continue;
                    relations.add(new StructuredRelationship(
                            StructuredRelationship.RelationshipType.CONDITIONAL,
                            candidate.candidate().sourceTable(), candidate.candidate().sourceColumn().name(),
                            candidate.candidate().target().table(), candidate.candidate().target().column().name(),
                            confidence,
                            List.of(new StructuredRelationship.Condition(candidate.candidate().sourceTable(),
                                    discriminator.name(), "=", row.get(0))),
                            evidence(candidate.candidate(), coverage, "discriminator=" + discriminator.name()), List.of()));
                }
                if (!relations.isEmpty()) return relations;
            } catch (Exception error) {
                warnings.add("conditional relationship validation failed: " + shortMessage(error));
            }
        }
        return List.of();
    }

    private static void markAmbiguity(List<ScoredCandidate> candidates) {
        candidates.sort(Comparator.comparingDouble(ScoredCandidate::confidence).reversed());
        for (int i = 0; i < candidates.size(); i++) {
            for (int j = i + 1; j < candidates.size(); j++) {
                ScoredCandidate left = candidates.get(i), right = candidates.get(j);
                if (!left.candidate().sourceTable().equals(right.candidate().sourceTable())
                        || !left.candidate().sourceColumn().name().equals(right.candidate().sourceColumn().name())) continue;
                if (Math.abs(left.confidence() - right.confidence()) <= 0.05) {
                    left.setAmbiguous(true); right.setAmbiguous(true);
                }
            }
        }
    }

    private static StructuredRelationship toRelationship(ScoredCandidate candidate,
                                                           StructuredRelationship.RelationshipType type) {
        Candidate value = candidate.candidate();
        return new StructuredRelationship(type, value.sourceTable(), value.sourceColumn().name(),
                value.target().table(), value.target().column().name(), candidate.confidence(), List.of(),
                evidence(value, candidate.coverage(), "targetUnique=true"),
                type == StructuredRelationship.RelationshipType.AMBIGUOUS
                        ? List.of("multiple target columns have similar evidence") : List.of());
    }

    private static List<StructuredRelationship> physicalRelationships(SqlSchemaSnapshot snapshot) {
        return snapshot.foreignKeys().stream().map(key -> new StructuredRelationship(
                StructuredRelationship.RelationshipType.PHYSICAL_FK,
                key.sourceTable(), key.sourceColumn(), key.targetTable(), key.targetColumn(), 1,
                List.of(), List.of("JDBC imported foreign key: " + String.valueOf(key.name())), List.of())).toList();
    }

    private static List<String> evidence(Candidate candidate, Coverage coverage, String extra) {
        return List.of("nameScore=" + round(candidate.nameScore()), "sampleRows=" + coverage.sourceCount(),
                "coverage=" + round(coverage.coverage()), "orphanRate=" + round(coverage.orphanRate()),
                "distinctRatio=" + round(coverage.distinctRatio()), extra);
    }

    private static double score(double nameScore, Coverage coverage) {
        if (coverage.sourceCount() == 0) return 0;
        return Math.max(0, Math.min(1, nameScore + 0.20 + 0.35 * coverage.coverage()
                + 0.10 * coverage.distinctRatio() - 0.20 * coverage.orphanRate()));
    }

    private static double nameScore(String sourceTable, SchemaExplorerService.ColumnSpec source,
                                    TargetColumn target) {
        String sourceName = normalize(source.name()), targetName = normalize(target.column().name());
        String sourceStem = stem(sourceName), targetTable = singular(normalize(target.table()));
        double score = 0;
        if (sourceName.equals(targetName)) score += 0.35;
        else if (sourceStem.equals(targetTable)) score += 0.30;
        else if (sourceName.contains(targetTable) || targetTable.contains(sourceStem)) score += 0.20;
        else if (suffix(sourceName).equals(suffix(targetName))) score += 0.08;
        if (commentOverlap(source.remarks(), target.column().remarks())) score += 0.05;
        if (singular(normalize(sourceTable)).equals(targetTable)) score -= 0.10;
        return Math.max(0, score);
    }

    private static boolean compatible(String left, String right) {
        return typeFamily(left).equals(typeFamily(right)) && !typeFamily(left).equals("OTHER");
    }

    private static String typeFamily(String type) {
        String value = type == null ? "" : type.toUpperCase(Locale.ROOT);
        if (value.contains("INT") || value.contains("DECIMAL") || value.contains("NUMERIC")) return "NUMBER";
        if (value.contains("CHAR") || value.contains("TEXT") || value.contains("ENUM")) return "TEXT";
        if (value.contains("UUID")) return "TEXT";
        return "OTHER";
    }

    private static boolean isRelationColumn(String name) {
        String value = normalize(name);
        return value.equals("id") || value.equals("code") || value.endsWith("_id") || value.endsWith("_code");
    }

    private static boolean isDiscriminator(String name) {
        String value = normalize(name);
        return value.equals("type") || value.equals("category") || value.equals("kind")
                || value.endsWith("_type") || value.endsWith("_category") || value.endsWith("_kind");
    }

    private static boolean isShortValueType(String type) { return "TEXT".equals(typeFamily(type)); }

    private static boolean commentOverlap(String left, String right) {
        if (left == null || right == null || left.isBlank() || right.isBlank()) return false;
        Set<String> tokens = new HashSet<>(List.of(left.toLowerCase(Locale.ROOT).split("[^\\p{L}\\p{N}]+")));
        for (String token : right.toLowerCase(Locale.ROOT).split("[^\\p{L}\\p{N}]+")) {
            if (token.length() > 1 && tokens.contains(token)) return true;
        }
        return false;
    }

    private static List<StructuredRelationship> deduplicate(List<StructuredRelationship> relationships) {
        Map<String, StructuredRelationship> result = new LinkedHashMap<>();
        relationships.forEach(relation -> result.merge(relation.id(), relation,
                (left, right) -> left.confidence() >= right.confidence() ? left : right));
        return List.copyOf(result.values());
    }

    private static long number(List<Object> row, int index) {
        if (index >= row.size() || row.get(index) == null) return 0;
        Object value = row.get(index);
        return value instanceof Number number ? number.longValue() : Long.parseLong(String.valueOf(value));
    }

    private static String quote(String identifier) { return "`" + identifier.replace("`", "") + "`"; }
    private static String literal(Object value) { return "'" + String.valueOf(value).replace("'", "''") + "'"; }
    private static String normalize(String value) { return value == null ? "" : value.toLowerCase(Locale.ROOT); }
    private static String suffix(String value) { int index = value.lastIndexOf('_'); return index < 0 ? value : value.substring(index + 1); }
    private static String stem(String value) { return value.endsWith("_id") ? value.substring(0, value.length() - 3)
            : value.endsWith("_code") ? value.substring(0, value.length() - 5) : value; }
    private static String discriminatorStem(String value) {
        String normalized = normalize(value);
        for (String suffix : List.of("_type", "_category", "_kind")) {
            if (normalized.endsWith(suffix)) return normalized.substring(0, normalized.length() - suffix.length());
        }
        return normalized.equals("type") || normalized.equals("category") || normalized.equals("kind") ? "id" : normalized;
    }
    private static String value(String value, String fallback) {
        return value == null || value.isBlank() ? fallback : value;
    }
    private static String singular(String value) { return value.endsWith("s") && value.length() > 1
            ? value.substring(0, value.length() - 1) : value; }
    private static String emptyToNull(String value) { return value == null || value.isBlank() ? null : value; }
    private static double round(double value) { return Math.round(value * 1000d) / 1000d; }
    private static String shortMessage(Exception error) { String message = error.getMessage(); return message == null
            ? error.getClass().getSimpleName() : message.substring(0, Math.min(200, message.length())); }

    private record Metadata(Map<String, Set<String>> uniqueColumns) {}
    private record TargetColumn(String table, SchemaExplorerService.ColumnSpec column) {}
    private record Candidate(String sourceTable, SchemaExplorerService.ColumnSpec sourceColumn,
                             TargetColumn target, double nameScore) {}
    private record Coverage(long sourceCount, long distinctCount, long orphanCount) {
        double coverage() { return sourceCount == 0 ? 0 : Math.max(0, (double) (sourceCount - orphanCount) / sourceCount); }
        double orphanRate() { return sourceCount == 0 ? 1 : Math.min(1, (double) orphanCount / sourceCount); }
        double distinctRatio() { return sourceCount == 0 ? 0 : Math.min(1, (double) distinctCount / sourceCount); }
    }
    private static final class ScoredCandidate {
        private final Candidate candidate; private final Coverage coverage; private final double confidence;
        private boolean ambiguous;
        private ScoredCandidate(Candidate candidate, Coverage coverage, double confidence) {
            this.candidate = candidate; this.coverage = coverage; this.confidence = confidence;
        }
        Candidate candidate() { return candidate; } Coverage coverage() { return coverage; }
        double confidence() { return confidence; } boolean ambiguous() { return ambiguous; }
        void setAmbiguous(boolean value) { ambiguous = value; }
    }
    public record DiscoveryResult(List<StructuredRelationship> relationships, List<String> warnings) {}
}
