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

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.adapter.jdbc.JdbcTable;
import org.apache.calcite.config.CalciteConnectionConfig;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rel.type.RelRecordType;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaVersion;
import org.apache.calcite.schema.Statistic;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.TranslatableTable;
import org.apache.calcite.schema.Wrapper;
import org.apache.calcite.schema.impl.DelegatingSchema;
import org.apache.calcite.schema.lookup.IgnoreCaseLookup;
import org.apache.calcite.schema.lookup.LikePattern;
import org.apache.calcite.schema.lookup.Lookup;
import org.apache.calcite.sql.SqlCall;
import org.apache.calcite.sql.SqlNode;

/**
 * 保留 MySQL 表名精确匹配，同时让列名遵循 MySQL 的大小写不敏感语义。
 */
final class MysqlCaseInsensitiveColumnSchema extends DelegatingSchema {

    private final Lookup<Table> tableLookup;

    /**
     * 创建 MySQL 列名语义包装器。
     *
     * @param schema 原始 JDBC Schema
     */
    MysqlCaseInsensitiveColumnSchema(Schema schema) {
        super(Objects.requireNonNull(schema, "schema"));
        Lookup<Table> sourceLookup = schema.tables();
        if (schema instanceof JdbcSchema jdbcSchema) {
            sourceLookup = new ExactJdbcTableLookup(jdbcSchema, sourceLookup);
        }
        this.tableLookup = sourceLookup.map((table, ignoredName) -> wrap(table));
    }

    /**
     * 返回保持原始表名 Lookup 规则的包装表集合。
     *
     * @return 包装后的表 Lookup
     */
    @Override
    public Lookup<Table> tables() {
        return tableLookup;
    }

    /**
     * 按原始 Schema 规则精确获取表，再包装列类型。
     *
     * @param name 表名
     * @return 包装表；不存在时返回 null
     */
    @Override
    public Table getTable(String name) {
        return tableLookup.get(name);
    }

    /**
     * 为 Schema 快照保留相同的列名语义。
     *
     * @param version Schema 版本
     * @return 包装后的快照
     */
    @Override
    public Schema snapshot(SchemaVersion version) {
        return new MysqlCaseInsensitiveColumnSchema(schema.snapshot(version));
    }

    private static Table wrap(Table table) {
        return table instanceof MysqlCaseInsensitiveColumnTable
            ? table
            : new MysqlCaseInsensitiveColumnTable(table);
    }

    /**
     * 将 Calcite 的表名查找收紧为 JDBC 元数据层面的精确查找。
     */
    private static final class ExactJdbcTableLookup extends IgnoreCaseLookup<Table> {

        private final JdbcSchema jdbcSchema;
        private final Lookup<Table> delegate;
        private volatile boolean searchEscapeLoaded;
        private String searchEscape;

        private ExactJdbcTableLookup(JdbcSchema jdbcSchema, Lookup<Table> delegate) {
            this.jdbcSchema = Objects.requireNonNull(jdbcSchema, "jdbcSchema");
            this.delegate = Objects.requireNonNull(delegate, "delegate");
        }

        /**
         * 转义 JDBC LIKE 通配字符后读取，并校验驱动返回的真实物理表名。
         *
         * @param name 精确表名
         * @return 精确匹配的表；不存在时返回 null
         */
        @Override
        public Table get(String name) {
            Table table = delegate.get(escapePattern(name));
            if (table == null) {
                return null;
            }
            JdbcTable jdbcTable = table instanceof JdbcTable direct
                ? direct
                : table instanceof Wrapper wrapper
                    ? wrapper.unwrap(JdbcTable.class)
                    : null;
            return jdbcTable != null && name.equals(jdbcTable.jdbcTableName) ? table : null;
        }

        /**
         * 返回符合 Calcite LIKE 语义的表名，过滤 JDBC 对下划线的额外通配匹配。
         *
         * @param pattern 表名模式
         * @return 匹配名称集合
         */
        @Override
        public Set<String> getNames(LikePattern pattern) {
            return delegate.getNames(pattern).stream()
                .filter(pattern.matcher()::apply)
                .collect(Collectors.toUnmodifiableSet());
        }

        private String escapePattern(String name) {
            String escape = searchEscape();
            if (escape == null || escape.isEmpty()) {
                if (name.indexOf('_') >= 0 || name.indexOf('%') >= 0) {
                    throw new IllegalStateException(
                        "MySQL JDBC driver does not expose a metadata search escape"
                    );
                }
                return name;
            }
            return name
                .replace(escape, escape + escape)
                .replace("_", escape + "_")
                .replace("%", escape + "%");
        }

        private String searchEscape() {
            if (searchEscapeLoaded) {
                return searchEscape;
            }
            synchronized (this) {
                if (!searchEscapeLoaded) {
                    try (Connection connection = jdbcSchema.getDataSource().getConnection()) {
                        searchEscape = connection.getMetaData().getSearchStringEscape();
                        searchEscapeLoaded = true;
                    } catch (SQLException exception) {
                        throw new IllegalStateException(
                            "Failed to read MySQL JDBC metadata search escape",
                            exception
                        );
                    }
                }
                return searchEscape;
            }
        }
    }

    /**
     * 仅调整行类型的字段查找规则，关系转换继续交由原始 JDBC Table 完成。
     */
    private static final class MysqlCaseInsensitiveColumnTable
        implements TranslatableTable, Wrapper {

        private final Table delegate;

        private MysqlCaseInsensitiveColumnTable(Table delegate) {
            this.delegate = Objects.requireNonNull(delegate, "delegate");
        }

        /**
         * 返回列名大小写不敏感的结构类型。
         *
         * @param typeFactory Calcite 类型工厂
         * @return 包装后的结构类型
         */
        @Override
        public RelDataType getRowType(RelDataTypeFactory typeFactory) {
            return new CaseInsensitiveRelRecordType(delegate.getRowType(typeFactory));
        }

        /**
         * 复用原始表统计信息。
         *
         * @return 表统计信息
         */
        @Override
        public Statistic getStatistic() {
            return delegate.getStatistic();
        }

        /**
         * 复用原始 JDBC 表类型。
         *
         * @return JDBC 表类型
         */
        @Override
        public Schema.TableType getJdbcTableType() {
            return delegate.getJdbcTableType();
        }

        /**
         * 判断列是否为预聚合列。
         *
         * @param column 列名
         * @return 原始表判断结果
         */
        @Override
        public boolean isRolledUp(String column) {
            return delegate.isRolledUp(column);
        }

        /**
         * 判断预聚合列能否用于聚合表达式。
         *
         * @param column 列名
         * @param call SQL 调用
         * @param parent 父节点
         * @param config Calcite 连接配置
         * @return 原始表判断结果
         */
        @Override
        public boolean rolledUpColumnValidInsideAgg(
            String column,
            SqlCall call,
            SqlNode parent,
            CalciteConnectionConfig config
        ) {
            return delegate.rolledUpColumnValidInsideAgg(column, call, parent, config);
        }

        /**
         * 交由原始 JDBC Table 生成关系节点，保留 JDBC Convention 与 SQL 下推。
         *
         * @param context 关系转换上下文
         * @param relOptTable 规划器表
         * @return 关系节点
         * @throws IllegalStateException 原始表不支持关系转换
         */
        @Override
        public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) {
            if (!(delegate instanceof TranslatableTable translatableTable)) {
                throw new IllegalStateException("MySQL JDBC table does not support relational translation");
            }
            return translatableTable.toRel(context, relOptTable);
        }

        /**
         * 解包包装器或原始 JDBC Table 能力。
         *
         * @param type 目标类型
         * @param <C> 目标类型参数
         * @return 匹配实例；不存在时返回 null
         */
        @Override
        public <C> C unwrap(Class<C> type) {
            if (type.isInstance(this)) {
                return type.cast(this);
            }
            if (type.isInstance(delegate)) {
                return type.cast(delegate);
            }
            return delegate instanceof Wrapper wrapper ? wrapper.unwrap(type) : null;
        }
    }

    /**
     * 始终以大小写不敏感方式解析 MySQL 列名的记录类型。
     */
    private static final class CaseInsensitiveRelRecordType extends RelRecordType {

        private CaseInsensitiveRelRecordType(RelDataType delegate) {
            super(delegate.getStructKind(), delegate.getFieldList(), delegate.isNullable());
        }

        /**
         * 按 MySQL 规则查找字段。
         *
         * @param fieldName 字段名
         * @param caseSensitive Calcite 请求的匹配规则；MySQL 列名语义下忽略
         * @param elideRecord 是否递归省略嵌套记录层级
         * @return 匹配字段；不存在时返回 null
         */
        @Override
        public RelDataTypeField getField(
            String fieldName,
            boolean caseSensitive,
            boolean elideRecord
        ) {
            return super.getField(fieldName, false, elideRecord);
        }
    }
}
