package com.easyagents.federation.sql.runtime;

import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.api.SqlCompletionItem;
import com.easyagents.federation.sql.api.SqlCompletionKind;
import com.easyagents.federation.sql.api.SqlCompletionRequest;
import com.easyagents.federation.sql.api.SqlCompletionResult;
import com.easyagents.federation.sql.federation.FederationLogicalTableDefinition;
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
import com.easyagents.federation.sql.source.FederationSchemaDefinition;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.calcite.config.CalciteConnectionConfigImpl;
import org.apache.calcite.config.CalciteConnectionProperty;
import org.apache.calcite.jdbc.CalciteSchema;
import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
import org.apache.calcite.prepare.CalciteCatalogReader;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.sql.SqlFunction;
import org.apache.calcite.sql.SqlOperatorTable;
import org.apache.calcite.sql.advise.SqlAdvisor;
import org.apache.calcite.sql.advise.SqlAdvisorValidator;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.sql.parser.SqlParser;
import org.apache.calcite.sql.util.SqlOperatorTables;
import org.apache.calcite.sql.validate.SqlMoniker;
import org.apache.calcite.sql.validate.SqlMonikerImpl;
import org.apache.calcite.sql.validate.SqlMonikerType;
import org.apache.calcite.sql.validate.SqlValidator;

/**
 * 基于 Calcite Advisor 的请求级 SQL 补全器。
 */
final class CalciteSqlCompleter {

    /**
     * 在当前 Runtime 快照内生成上下文补全候选。
     *
     * @param request 补全请求
     * @param snapshot 查询范围 Runtime 快照
     * @return 替换区间与候选列表
     */
    SqlCompletionResult complete(
        SqlCompletionRequest request,
        FederationQueryScopeSnapshot snapshot
    ) {
        try {
            CompletionCatalog catalog = buildCatalog(request.queryScope(), snapshot);
            SourceRuntime plannerRuntime = plannerRuntime(request.queryScope(), snapshot);
            SqlParser.Config parserConfig = plannerRuntime.adapter()
                .parserConfig(plannerRuntime.dialect());
            JavaTypeFactoryImpl typeFactory = new JavaTypeFactoryImpl(
                plannerRuntime.adapter().typeSystem()
            );
            CalciteCatalogReader catalogReader = new CalciteCatalogReader(
                CalciteSchema.from(catalog.root()),
                catalog.defaultPath(),
                typeFactory,
                connectionConfig(parserConfig)
            );
            SqlOperatorTable operatorTable = operatorTable(snapshot);
            SqlAdvisorValidator validator = new SqlAdvisorValidator(
                operatorTable,
                catalogReader,
                typeFactory,
                SqlValidator.Config.DEFAULT
                    .withConformance(parserConfig.conformance())
                    .withIdentifierExpansion(true)
                    .withLenientOperatorLookup(true)
            );
            SqlAdvisor advisor = new SqlAdvisor(validator, parserConfig);
            String[] replaced = {""};
            List<SqlMoniker> hints = new ArrayList<>(advisor.getCompletionHints(
                request.sql(),
                request.cursorOffset(),
                replaced
            ));
            // Advisor 在语句首部仅有半截关键字时可能没有候选，直接复用当前
            // Calcite Parser 的期望 Token 补齐，避免维护一份易漂移的关键字表。
            hints.addAll(statementStartHints(
                advisor,
                request.sql(),
                request.cursorOffset(),
                replaced[0]
            ));
            // SqlAdvisor 对末尾三段 Schema 路径可能不返回候选，继续使用同一 Calcite
            // CatalogReader 补齐该路径，避免实现第二套 SQL 元数据目录。
            hints.addAll(qualifiedCatalogHints(
                catalogReader,
                request.sql(),
                request.cursorOffset(),
                replaced[0]
            ));
            int replaceStart = request.cursorOffset() - replaced[0].length();
            return new SqlCompletionResult(
                replaceStart,
                request.cursorOffset(),
                completionItems(advisor, operatorTable, hints, replaced[0])
            );
        } catch (FederationSqlException exception) {
            throw exception;
        } catch (RuntimeException exception) {
            throw new FederationSqlException(
                FederationSqlErrorCode.SQL_COMPLETION_FAILED,
                "Calcite SQL completion failed",
                exception
            );
        }
    }

    private static CompletionCatalog buildCatalog(
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        return scope.logicalTables().isEmpty()
            ? buildPhysicalCatalog(scope, snapshot)
            : buildLogicalCatalog(scope, snapshot);
    }

    private static CompletionCatalog buildLogicalCatalog(
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        SchemaPlus root = CalciteSchema.createRootSchema(true, false).plus();
        Map<String, SchemaPlus> bindingSchemas = new LinkedHashMap<>();
        Map<String, SchemaPlus> logicalSchemas = new LinkedHashMap<>();
        for (FederationLogicalTableDefinition definition : scope.logicalTables()) {
            SourceRuntime runtime = snapshot.runtime(definition.bindingName());
            Table sourceTable = sourceTable(scope, runtime, definition);
            SchemaPlus bindingSchema = bindingSchemas.computeIfAbsent(
                definition.bindingName(),
                name -> root.add(name, new AbstractSchema())
            );
            String schemaKey = definition.bindingName() + '\u0000' + definition.schemaName();
            SchemaPlus logicalSchema = logicalSchemas.computeIfAbsent(
                schemaKey,
                ignored -> bindingSchema.add(definition.schemaName(), new AbstractSchema())
            );
            // 逻辑目录只挂载显式授权别名，避免 Advisor 泄露同 Schema 的其他物理表。
            root.add(definition.logicalName(), sourceTable);
            logicalSchema.add(definition.logicalName(), sourceTable);
        }
        return new CompletionCatalog(root, List.of());
    }

    private static CompletionCatalog buildPhysicalCatalog(
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        SchemaPlus root = CalciteSchema.createRootSchema(true, false).plus();
        List<String> defaultPath = List.of();
        for (Map.Entry<String, SourceRuntime> entry : snapshot.runtimesByBinding().entrySet()) {
            String bindingName = entry.getKey();
            SourceRuntime runtime = entry.getValue();
            FederationSourceBindingDefinition binding = scope.bindings().get(bindingName);
            SchemaPlus bindingSchema = root.add(bindingName, new AbstractSchema());
            Map<String, String> mappings = schemaMappings(binding, runtime);
            for (Map.Entry<String, String> mapping : mappings.entrySet()) {
                SchemaPlus sourceSchema = sourceSchema(runtime, mapping.getValue());
                Schema mounted = sourceSchema.unwrap(Schema.class);
                bindingSchema.add(mapping.getKey(), mounted);
            }
            if (bindingName.equals(scope.defaultBinding())) {
                defaultPath = mappings.size() == 1
                    ? List.of(bindingName, mappings.keySet().iterator().next())
                    : List.of(bindingName);
            }
        }
        return new CompletionCatalog(root, defaultPath);
    }

    private static Table sourceTable(
        FederationQueryScopeDefinition scope,
        SourceRuntime runtime,
        FederationLogicalTableDefinition definition
    ) {
        FederationSourceBindingDefinition binding = scope.bindings()
            .get(definition.bindingName());
        String physicalSchema = mappedSchema(binding, definition.schemaName());
        SchemaPlus schema = sourceSchema(runtime, physicalSchema);
        Table table = schema.getTable(definition.sourceTableName());
        if (table == null) {
            String actualName = schema.getTableNames().stream()
                .filter(name -> name.equalsIgnoreCase(definition.sourceTableName()))
                .findFirst()
                .orElse(null);
            table = actualName == null ? null : schema.getTable(actualName);
        }
        if (table == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_QUERY_SCOPE,
                "logical table maps an unknown source table: " + definition.logicalName()
            );
        }
        return table;
    }

    private static SchemaPlus sourceSchema(SourceRuntime runtime, String schemaName) {
        SchemaPlus sourceRoot = runtime.rootSchema()
            .getSubSchema(runtime.definition().sourceId().value());
        SchemaPlus sourceSchema = sourceRoot == null ? null : sourceRoot.getSubSchema(schemaName);
        if (sourceSchema == null && sourceRoot != null) {
            String actualName = sourceRoot.getSubSchemaNames().stream()
                .filter(name -> name.equalsIgnoreCase(schemaName))
                .findFirst()
                .orElse(null);
            sourceSchema = actualName == null ? null : sourceRoot.getSubSchema(actualName);
        }
        if (sourceSchema == null) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_QUERY_SCOPE,
                "query scope maps an unknown source schema: " + schemaName
            );
        }
        return sourceSchema;
    }

    private static String mappedSchema(
        FederationSourceBindingDefinition binding,
        String logicalSchema
    ) {
        if (binding == null || binding.schemaMappings().isEmpty()) {
            return logicalSchema;
        }
        return binding.schemaMappings().entrySet().stream()
            .filter(entry -> entry.getKey().equalsIgnoreCase(logicalSchema))
            .map(Map.Entry::getValue)
            .findFirst()
            .orElseThrow(() -> new FederationSqlException(
                FederationSqlErrorCode.INVALID_QUERY_SCOPE,
                "logical table references an unknown mapped schema: " + logicalSchema
            ));
    }

    private static Map<String, String> schemaMappings(
        FederationSourceBindingDefinition binding,
        SourceRuntime runtime
    ) {
        if (binding != null && !binding.schemaMappings().isEmpty()) {
            return binding.schemaMappings();
        }
        LinkedHashMap<String, String> mappings = new LinkedHashMap<>();
        for (FederationSchemaDefinition definition : runtime.definition().schemas()) {
            mappings.put(definition.logicalName(), definition.logicalName());
        }
        return mappings;
    }

    private static CalciteConnectionConfigImpl connectionConfig(
        SqlParser.Config parserConfig
    ) {
        return new CalciteConnectionConfigImpl(new Properties())
            .set(
                CalciteConnectionProperty.CASE_SENSITIVE,
                Boolean.toString(parserConfig.caseSensitive())
            )
            .set(CalciteConnectionProperty.QUOTING, parserConfig.quoting().name())
            .set(
                CalciteConnectionProperty.UNQUOTED_CASING,
                parserConfig.unquotedCasing().name()
            )
            .set(
                CalciteConnectionProperty.QUOTED_CASING,
                parserConfig.quotedCasing().name()
            )
            .set(
                CalciteConnectionProperty.CONFORMANCE,
                parserConfig.conformance().toString()
            );
    }

    private static SqlOperatorTable operatorTable(FederationQueryScopeSnapshot snapshot) {
        List<SqlOperatorTable> adapterTables = snapshot.runtimesByBinding().values().stream()
            .map(runtime -> runtime.adapter().operatorTable())
            .distinct()
            .toList();
        List<SqlOperatorTable> tables = new ArrayList<>(adapterTables.size() + 1);
        // Adapter 只声明厂商扩展；标准 SQL 函数始终由 Calcite 标准表提供。
        tables.add(SqlStdOperatorTable.instance());
        tables.addAll(adapterTables);
        return SqlOperatorTables.chain(tables);
    }

    private static SourceRuntime plannerRuntime(
        FederationQueryScopeDefinition scope,
        FederationQueryScopeSnapshot snapshot
    ) {
        SourceRuntime runtime = snapshot.runtimesByBinding().get(scope.defaultBinding());
        return runtime == null
            ? snapshot.runtimesByBinding().values().iterator().next()
            : runtime;
    }

    private static List<SqlCompletionItem> completionItems(
        SqlAdvisor advisor,
        SqlOperatorTable operatorTable,
        List<SqlMoniker> hints,
        String replacedWord
    ) {
        LinkedHashMap<String, SqlCompletionItem> distinct = new LinkedHashMap<>();
        for (SqlMoniker hint : hints) {
            List<String> qualifiedName = hint.getFullyQualifiedNames();
            String label = qualifiedName.isEmpty()
                ? hint.id()
                : qualifiedName.get(qualifiedName.size() - 1);
            String insertText = advisor.getReplacement(hint, replacedWord);
            SqlCompletionKind kind = completionKind(
                hint.getType(),
                label,
                operatorTable
            );
            SqlCompletionItem item = new SqlCompletionItem(
                label,
                insertText,
                kind,
                qualifiedName
            );
            distinct.putIfAbsent(kind + "\u0000" + insertText + "\u0000" + qualifiedName, item);
        }
        return new ArrayList<>(distinct.values());
    }

    private static List<SqlMoniker> qualifiedCatalogHints(
        CalciteCatalogReader catalogReader,
        String sql,
        int cursorOffset,
        String replacedWord
    ) {
        int tokenStart = cursorOffset;
        while (tokenStart > 0) {
            char current = sql.charAt(tokenStart - 1);
            if (current != '.' && !Character.isJavaIdentifierPart(current)) {
                break;
            }
            tokenStart--;
        }
        String token = sql.substring(tokenStart, cursorOffset);
        int lastDot = token.lastIndexOf('.');
        if (lastDot <= 0) {
            return List.of();
        }
        String qualifier = token.substring(0, lastDot);
        List<String> path = List.of(qualifier.split("\\."));
        return catalogReader.getAllSchemaObjectNames(path).stream()
            .filter(moniker -> {
                List<String> names = moniker.getFullyQualifiedNames();
                String name = names.isEmpty() ? moniker.id() : names.get(names.size() - 1);
                return replacedWord.isEmpty()
                    || name.regionMatches(true, 0, replacedWord, 0, replacedWord.length());
            })
            .toList();
    }

    private static List<SqlMoniker> statementStartHints(
        SqlAdvisor advisor,
        String sql,
        int cursorOffset,
        String replacedWord
    ) {
        int wordStart = cursorOffset - replacedWord.length();
        if (wordStart < 0 || !sql.substring(0, wordStart).isBlank()) {
            return List.of();
        }
        return advisor.getReservedAndKeyWords().stream()
                .filter(keyword -> keyword.regionMatches(
                    true,
                    0,
                    replacedWord,
                    0,
                    replacedWord.length()
                ))
                .map(keyword -> (SqlMoniker) new SqlMonikerImpl(
                    List.of(keyword),
                    SqlMonikerType.KEYWORD
                ))
                .distinct()
                .toList();
    }

    private static SqlCompletionKind completionKind(
        SqlMonikerType type,
        String label,
        SqlOperatorTable operatorTable
    ) {
        // SqlAdvisor 会把 COUNT 等可直接输入的标准函数标记为 KEYWORD，继续以
        // Calcite OperatorTable 校正分类，避免前端维护函数名称清单。
        if (type == SqlMonikerType.FUNCTION || operatorTable.getOperatorList().stream()
            .anyMatch(operator -> operator instanceof SqlFunction
                && operator.getName().equalsIgnoreCase(label))) {
            return SqlCompletionKind.FUNCTION;
        }
        if (type == SqlMonikerType.KEYWORD) {
            return SqlCompletionKind.KEYWORD;
        }
        if (type == SqlMonikerType.TABLE) {
            return SqlCompletionKind.TABLE;
        }
        if (type == SqlMonikerType.VIEW) {
            return SqlCompletionKind.VIEW;
        }
        if (type == SqlMonikerType.SCHEMA) {
            return SqlCompletionKind.SCHEMA;
        }
        if (type == SqlMonikerType.CATALOG) {
            return SqlCompletionKind.CATALOG;
        }
        if (type == SqlMonikerType.COLUMN) {
            return SqlCompletionKind.COLUMN;
        }
        return SqlCompletionKind.OTHER;
    }

    private record CompletionCatalog(SchemaPlus root, List<String> defaultPath) {
    }
}
