package com.easyagents.federation.sql.adapter.jdbc;

import com.easyagents.federation.sql.adapter.AdapterCompatibility;
import com.easyagents.federation.sql.adapter.AdapterCompatibilityStatus;
import com.easyagents.federation.sql.adapter.AdapterDialectContext;
import com.easyagents.federation.sql.adapter.AdapterHints;
import com.easyagents.federation.sql.adapter.AdapterSchemaContext;
import com.easyagents.federation.sql.adapter.FederationSqlAdapterProvider;
import com.easyagents.federation.sql.adapter.FederationStatisticsCollector;
import com.easyagents.federation.sql.api.FederationSqlErrorCode;
import com.easyagents.federation.sql.api.FederationSqlException;
import com.easyagents.federation.sql.execute.FederationFragmentExecutor;
import com.easyagents.federation.sql.execute.FederationFragmentExplainer;
import java.sql.DatabaseMetaData;
import java.sql.SQLException;
import java.util.Locale;
import java.util.Optional;
import java.util.Set;
import org.apache.calcite.adapter.jdbc.JdbcConvention;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.Schemas;
import org.apache.calcite.sql.SqlDialect;
import org.apache.calcite.sql.SqlDialectFactoryImpl;
import org.apache.calcite.sql.dialect.AnsiSqlDialect;
import org.apache.calcite.sql.dialect.MysqlSqlDialect;

/**
 * MySQL、PostgreSQL、Oracle 及显式实验 ANSI 数据库的默认 JDBC Adapter。
 */
public final class JdbcFederationSqlAdapterProvider implements FederationSqlAdapterProvider {

    /** 默认 JDBC Adapter 标识。 */
    public static final String ADAPTER_ID = "jdbc";
    /** 允许未知数据库采用实验 ANSI 方言的 Definition 选项。 */
    public static final String EXPERIMENTAL_ANSI_OPTION = "experimentalAnsi";

    private static final Set<String> SUPPORTED_PRODUCTS = Set.of(
        "mysql",
        "postgresql",
        "oracle",
        "h2"
    );

    private final FederationFragmentExecutor executor = new JdbcFederationFragmentExecutor();
    private final FederationFragmentExplainer explainer = new JdbcFederationFragmentExplainer();
    private final FederationStatisticsCollector statisticsCollector =
        new JdbcFederationStatisticsCollector();

    /**
     * 创建默认 JDBC Adapter Provider。
     */
    public JdbcFederationSqlAdapterProvider() {
    }

    /**
     * 返回默认 Adapter 标识。
     *
     * @return {@value #ADAPTER_ID}
     */
    @Override
    public String adapterId() {
        return ADAPTER_ID;
    }

    /**
     * 基于数据库产品名判断内建或实验 ANSI 支持。
     *
     * @param metadata JDBC 元数据
     * @param hints Adapter 提示
     * @return 是否支持
     * @throws SQLException 元数据读取失败
     */
    @Override
    public boolean supports(DatabaseMetaData metadata, AdapterHints hints) throws SQLException {
        return SUPPORTED_PRODUCTS.contains(normalize(metadata.getDatabaseProductName()))
            || hints.enabled(EXPERIMENTAL_ANSI_OPTION);
    }

    /**
     * 返回与实际验证证据一致的兼容性状态。
     *
     * @param metadata JDBC 元数据
     * @param hints Adapter 提示
     * @return 兼容性说明
     * @throws SQLException 元数据读取失败
     */
    @Override
    public AdapterCompatibility compatibility(DatabaseMetaData metadata, AdapterHints hints) throws SQLException {
        String product = metadata.getDatabaseProductName();
        String normalized = normalize(product);
        AdapterCompatibilityStatus status;
        String diagnostic;
        if ("h2".equals(normalized)) {
            status = AdapterCompatibilityStatus.VERIFIED;
            diagnostic = "verified by module-level H2 integration tests";
        } else if (SUPPORTED_PRODUCTS.contains(normalized)) {
            status = AdapterCompatibilityStatus.CODE_SUPPORTED_UNVERIFIED;
            diagnostic = "dialect is supported by code; verify against the target database version before production";
        } else if (hints.enabled(EXPERIMENTAL_ANSI_OPTION)) {
            status = AdapterCompatibilityStatus.CODE_SUPPORTED_UNVERIFIED;
            diagnostic = "experimental ANSI mode is enabled for an unrecognized database";
        } else {
            status = AdapterCompatibilityStatus.UNSUPPORTED;
            diagnostic = "database product is not recognized";
        }
        return new AdapterCompatibility(
            status,
            product,
            metadata.getDatabaseProductVersion(),
            metadata.getDriverName(),
            metadata.getDriverVersion(),
            diagnostic
        );
    }

    /**
     * 创建复用已探测 Dialect 和调用方 DataSource 的 JdbcSchema。
     *
     * @param context Schema 上下文
     * @return Calcite JdbcSchema
     */
    @Override
    public Schema createSchema(AdapterSchemaContext context) {
        if (!(context.schemaDefinition() instanceof JdbcSchemaDefinition definition)) {
            throw new FederationSqlException(
                FederationSqlErrorCode.INVALID_ARGUMENT,
                "jdbc adapter requires JdbcSchemaDefinition"
            );
        }
        JdbcConvention convention = JdbcConvention.of(
            context.dialect(),
            Schemas.subSchemaExpression(
                context.parentSchema(),
                definition.logicalName(),
                JdbcSchema.class
            ),
            context.sourceDefinition().sourceId().value() + "." + definition.logicalName()
        );
        Schema schema = new JdbcSchema(
            context.handle().dataSource(),
            context.dialect(),
            convention,
            definition.catalog(),
            definition.physicalSchema()
        );
        // MySQL 表名可区分大小写而列名始终不区分大小写，需分别建模。
        return context.dialect() instanceof MysqlSqlDialect
            ? new MysqlCaseInsensitiveColumnSchema(schema)
            : schema;
    }

    /**
     * 使用 Calcite 官方 DialectFactory 选择方言，未知数据库仅在显式 ANSI 模式下放行。
     *
     * @param context 方言上下文
     * @return SqlDialect
     * @throws SQLException 元数据读取失败
     */
    @Override
    public SqlDialect createDialect(AdapterDialectContext context) throws SQLException {
        String product = normalize(context.metadata().getDatabaseProductName());
        if (!SUPPORTED_PRODUCTS.contains(product)) {
            AdapterHints hints = new AdapterHints(context.sourceDefinition().adapterOptions());
            if (hints.enabled(EXPERIMENTAL_ANSI_OPTION)) {
                return AnsiSqlDialect.DEFAULT;
            }
            throw new FederationSqlException(
                FederationSqlErrorCode.ADAPTER_UNSUPPORTED,
                "database product is not supported by jdbc adapter: "
                    + context.metadata().getDatabaseProductName()
            );
        }
        return SqlDialectFactoryImpl.INSTANCE.create(context.metadata());
    }

    /**
     * 返回直接 JDBC 流式执行器。
     *
     * @return Fragment 执行器
     */
    @Override
    public FederationFragmentExecutor fragmentExecutor() {
        return executor;
    }

    /**
     * 返回 MySQL、PostgreSQL 和 H2 的显式物理 Explain 实现。
     *
     * @return JDBC 物理 Explain SPI
     */
    @Override
    public Optional<FederationFragmentExplainer> fragmentExplainer() {
        return Optional.of(explainer);
    }

    /**
     * 返回 MySQL 与 PostgreSQL 的内建目录统计采集器。
     *
     * <p>Oracle、H2 和实验 ANSI 数据库当前返回空统计，由引擎使用默认成本估算。</p>
     *
     * @return JDBC 统计采集 SPI
     */
    @Override
    public Optional<FederationStatisticsCollector> statisticsCollector() {
        return Optional.of(statisticsCollector);
    }

    private static String normalize(String productName) {
        return productName == null ? "" : productName.trim().toLowerCase(Locale.ROOT);
    }
}
