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

import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.util.List;
import java.util.Map;
import javax.sql.DataSource;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.adapter.jdbc.JdbcTable;
import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.Wrapper;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.calcite.sql.dialect.MysqlSqlDialect;
import org.apache.calcite.sql.type.SqlTypeName;
import org.junit.Assert;
import org.junit.Test;

/**
 * MySQL 表名与列名大小写语义包装器测试。
 */
public class MysqlCaseInsensitiveColumnSchemaTest {

    /**
     * 验证 JDBC 元数据 LIKE 模式不会把下划线表名解析到近似表。
     */
    @Test
    public void shouldEscapeJdbcMetadataPatternAndRequireExactPhysicalTable() {
        DataSource dataSource = metadataDataSource();
        JdbcSchema jdbcSchema = new JdbcSchema(
            dataSource,
            MysqlSqlDialect.DEFAULT,
            null,
            "app",
            null
        );
        JdbcTable rawTable = ((Wrapper) jdbcSchema.tables().get("order_item"))
            .unwrap(JdbcTable.class);
        Assert.assertEquals("order0item", rawTable.jdbcTableName);

        Schema schema = new MysqlCaseInsensitiveColumnSchema(jdbcSchema);
        JdbcTable exactTable = ((Wrapper) schema.getTable("order_item"))
            .unwrap(JdbcTable.class);
        Assert.assertEquals("order_item", exactTable.jdbcTableName);
    }

    /**
     * 验证大小写不同的表保持独立，同时列名可忽略大小写查找。
     */
    @Test
    public void shouldKeepExactTableNamesAndMatchColumnsIgnoringCase() {
        Schema schema = new MysqlCaseInsensitiveColumnSchema(new AbstractSchema() {
            @Override
            protected Map<String, Table> getTableMap() {
                return Map.of(
                    "orders", table("id"),
                    "Orders", table("different_column")
                );
            }
        });
        RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();

        Table lowerCaseTable = schema.getTable("orders");
        Table upperCaseTable = schema.getTable("Orders");
        Assert.assertNotNull(lowerCaseTable);
        Assert.assertNotNull(upperCaseTable);
        Assert.assertNull(schema.getTable("ORDERS"));

        RelDataType lowerCaseRow = lowerCaseTable.getRowType(typeFactory);
        RelDataType upperCaseRow = upperCaseTable.getRowType(typeFactory);
        Assert.assertNotNull(lowerCaseRow.getField("ID", true, false));
        Assert.assertEquals("id", lowerCaseRow.getField("ID", true, false).getName());
        Assert.assertNotNull(upperCaseRow.getField("DIFFERENT_COLUMN", true, false));
        Assert.assertNull(upperCaseRow.getField("ID", true, false));
    }

    private static Table table(String columnName) {
        return new AbstractTable() {
            @Override
            public RelDataType getRowType(RelDataTypeFactory typeFactory) {
                return typeFactory.builder()
                    .add(columnName, SqlTypeName.INTEGER)
                    .build();
            }
        };
    }

    private static DataSource metadataDataSource() {
        DatabaseMetaData metadata = (DatabaseMetaData) Proxy.newProxyInstance(
            MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
            new Class<?>[] {DatabaseMetaData.class},
            (proxy, method, arguments) -> switch (method.getName()) {
                case "getSearchStringEscape" -> "\\";
                case "getJDBCMajorVersion" -> 4;
                case "getJDBCMinorVersion" -> 2;
                case "getDatabaseProductName" -> "MySQL";
                case "getTables" -> tableResultSet((String) arguments[2]);
                default -> defaultValue(method.getReturnType());
            }
        );
        Connection connection = (Connection) Proxy.newProxyInstance(
            MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
            new Class<?>[] {Connection.class},
            (proxy, method, arguments) -> switch (method.getName()) {
                case "getMetaData" -> metadata;
                case "getCatalog" -> "app";
                case "getSchema" -> null;
                case "close" -> null;
                case "isClosed" -> false;
                default -> defaultValue(method.getReturnType());
            }
        );
        return (DataSource) Proxy.newProxyInstance(
            MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
            new Class<?>[] {DataSource.class},
            (proxy, method, arguments) -> switch (method.getName()) {
                case "getConnection" -> connection;
                default -> defaultValue(method.getReturnType());
            }
        );
    }

    private static ResultSet tableResultSet(String pattern) {
        List<String> tableNames = switch (pattern) {
            case "order_item" -> List.of("order0item", "order_item");
            case "order\\_item" -> List.of("order_item");
            case "%" -> List.of("order0item", "order_item");
            default -> List.of();
        };
        int[] cursor = {-1};
        return (ResultSet) Proxy.newProxyInstance(
            MysqlCaseInsensitiveColumnSchemaTest.class.getClassLoader(),
            new Class<?>[] {ResultSet.class},
            (proxy, method, arguments) -> switch (method.getName()) {
                case "next" -> ++cursor[0] < tableNames.size();
                case "getString" -> switch ((Integer) arguments[0]) {
                    case 1 -> "app";
                    case 2 -> null;
                    case 3 -> tableNames.get(cursor[0]);
                    case 4 -> "TABLE";
                    default -> null;
                };
                case "close" -> null;
                default -> defaultValue(method.getReturnType());
            }
        );
    }

    private static Object defaultValue(Class<?> type) {
        if (!type.isPrimitive()) {
            return null;
        }
        if (type == boolean.class) {
            return false;
        }
        if (type == int.class) {
            return 0;
        }
        if (type == long.class) {
            return 0L;
        }
        if (type == short.class) {
            return (short) 0;
        }
        if (type == byte.class) {
            return (byte) 0;
        }
        if (type == float.class) {
            return 0F;
        }
        if (type == double.class) {
            return 0D;
        }
        if (type == char.class) {
            return '\0';
        }
        return null;
    }
}
