package com.easyagents.federation.sql.api;

import com.easyagents.federation.sql.adapter.AdapterCompatibility;
import com.easyagents.federation.sql.adapter.AdapterCompatibilityStatus;
import com.easyagents.federation.sql.compile.FederationFragmentExplain;
import com.easyagents.federation.sql.compile.SqlCompileRequest;
import com.easyagents.federation.sql.compile.SqlExplainLevel;
import com.easyagents.federation.sql.compile.SqlExplainRequest;
import com.easyagents.federation.sql.compile.SqlExplainResult;
import com.easyagents.federation.sql.execute.FederationColumn;
import com.easyagents.federation.sql.execute.FederationFragmentMetrics;
import com.easyagents.federation.sql.execute.FederationPhysicalExplain;
import com.easyagents.federation.sql.execute.FederationQueryMetricsSnapshot;
import com.easyagents.federation.sql.execute.QueryId;
import com.easyagents.federation.sql.execute.SqlParameter;
import com.easyagents.federation.sql.federation.FederationExecutionPolicy;
import com.easyagents.federation.sql.federation.FederationQueryMode;
import com.easyagents.federation.sql.federation.FederationQueryScopeDefinition;
import com.easyagents.federation.sql.federation.FederationSourceBindingDefinition;
import com.easyagents.federation.sql.source.SourceId;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.sql.Types;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.junit.Assert;
import org.junit.Test;

/**
 * 跨节点查询命令序列化边界测试。
 */
public class SqlQueryCommandSerializationTest {

    /**
     * 验证命令可以序列化并在目标节点恢复。
     *
     * @throws Exception 序列化失败
     */
    @Test
    public void shouldRoundTripSerializableCommand() throws Exception {
        SqlQueryCommand command = SqlQueryCommand.of(
            "SELECT name FROM person WHERE id = ?",
            new SourceId("main"),
            3,
            List.of(SqlParameter.of(7))
        );
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        try (ObjectOutputStream output = new ObjectOutputStream(bytes)) {
            output.writeObject(command);
        }
        SqlQueryCommand restored;
        try (ObjectInputStream input = new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) {
            restored = (SqlQueryCommand) input.readObject();
        }
        Assert.assertEquals(command, restored);
    }

    /**
     * 验证非序列化参数在进入跨节点命令前被拒绝。
     */
    @Test(expected = IllegalArgumentException.class)
    public void shouldRejectNonSerializableParameterValue() {
        SqlParameter.of(new Object());
    }

    /**
     * 验证仅带 Serializable 标记的任意对象不能进入跨节点参数边界。
     */
    @Test(expected = IllegalArgumentException.class)
    public void shouldRejectUnsupportedSerializableObjectGraph() {
        SqlParameter.of(new BrokenSerializableValue());
    }

    /**
     * 验证常见浮点、时间和时区时间值映射为精确 JDBC 类型。
     */
    @Test
    public void shouldInferPreciseJdbcScalarTypes() {
        Assert.assertEquals(Types.REAL, SqlParameter.of(1.5F).jdbcType());
        Assert.assertEquals(Types.DOUBLE, SqlParameter.of(1.5D).jdbcType());
        Assert.assertEquals(
            Types.TIME,
            SqlParameter.of(java.time.LocalTime.NOON).jdbcType()
        );
        Assert.assertEquals(
            Types.TIME_WITH_TIMEZONE,
            SqlParameter.of(java.time.OffsetTime.parse("12:00:00+08:00")).jdbcType()
        );
        Assert.assertEquals(
            Types.TIMESTAMP_WITH_TIMEZONE,
            SqlParameter.of(java.time.OffsetDateTime.parse(
                "2026-08-21T12:00:00+08:00"
            )).jdbcType()
        );
        Assert.assertEquals(
            Types.OTHER,
            SqlParameter.of(java.util.UUID.randomUUID()).jdbcType()
        );
    }

    /**
     * 验证虚拟查询范围可跨节点序列化，且 Binding 顺序不影响稳定摘要。
     *
     * @throws Exception 序列化失败
     */
    @Test
    public void shouldSerializeVirtualScopeWithStableChecksum() throws Exception {
        SourceId first = new SourceId("first");
        SourceId second = new SourceId("second");
        FederationSourceBindingDefinition firstBinding =
            FederationSourceBindingDefinition.of(first, 2, Map.of("APP", "MAIN"));
        FederationSourceBindingDefinition secondBinding =
            FederationSourceBindingDefinition.of(second, 4);
        FederationQueryScopeDefinition scope = FederationQueryScopeDefinition.virtual(
            "virtual",
            9,
            Map.of("A", firstBinding, "B", secondBinding),
            "A",
            FederationExecutionPolicy.basic()
        );
        FederationQueryScopeDefinition reordered = FederationQueryScopeDefinition.virtual(
            "virtual",
            9,
            new java.util.LinkedHashMap<>(Map.of("B", secondBinding, "A", firstBinding)),
            "A",
            FederationExecutionPolicy.basic()
        );
        Assert.assertEquals(scope.checksum(), reordered.checksum());

        SqlQueryCommand command = SqlQueryCommand.of(
            "SELECT * FROM A.APP.T JOIN B.APP.T USING (ID)",
            scope,
            List.of()
        );
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        try (ObjectOutputStream output = new ObjectOutputStream(bytes)) {
            output.writeObject(command);
        }
        try (ObjectInputStream input = new ObjectInputStream(
            new ByteArrayInputStream(bytes.toByteArray()))) {
            SqlQueryCommand restored = (SqlQueryCommand) input.readObject();
            Assert.assertEquals(scope, restored.queryScope());
            Assert.assertEquals(scope.checksum(), restored.queryScope().checksum());
        }
    }

    /**
     * 验证 Binding 与 Schema Mapping 的层级边界会进入稳定摘要。
     */
    @Test
    public void shouldSeparateBindingAndSchemaMappingChecksumStructure() {
        FederationQueryScopeDefinition mappingScope = FederationQueryScopeDefinition.virtual(
            "collision-regression",
            1,
            Map.of("0", FederationSourceBindingDefinition.of(
                new SourceId("s1"),
                1,
                Map.of("1", "s2", "2", "b3", "3", "4")
            )),
            "0",
            FederationExecutionPolicy.basic()
        );
        FederationQueryScopeDefinition bindingScope = FederationQueryScopeDefinition.virtual(
            "collision-regression",
            1,
            Map.of(
                "0", FederationSourceBindingDefinition.of(new SourceId("s1"), 1),
                "1", FederationSourceBindingDefinition.of(new SourceId("s2"), 2),
                "b3", FederationSourceBindingDefinition.of(new SourceId("3"), 4)
            ),
            "0",
            FederationExecutionPolicy.basic()
        );

        Assert.assertNotEquals(mappingScope.checksum(), bindingScope.checksum());
    }

    /**
     * 验证未加引号标识符的大小写歧义在定义阶段直接拒绝。
     */
    @Test
    public void shouldRejectCaseInsensitiveScopeNameAmbiguity() {
        LinkedHashMap<String, FederationSourceBindingDefinition> bindings = new LinkedHashMap<>();
        bindings.put("Sales", FederationSourceBindingDefinition.of(new SourceId("first"), 1));
        bindings.put("SALES", FederationSourceBindingDefinition.of(new SourceId("second"), 1));
        try {
            FederationQueryScopeDefinition.virtual(
                "ambiguous-bindings",
                1,
                bindings,
                "Sales",
                FederationExecutionPolicy.basic()
            );
            Assert.fail("case-insensitive duplicate bindings should be rejected");
        } catch (IllegalArgumentException expected) {
            Assert.assertTrue(expected.getMessage().contains("binding names"));
        }

        LinkedHashMap<String, String> mappings = new LinkedHashMap<>();
        mappings.put("App", "MAIN");
        mappings.put("APP", "OTHER");
        try {
            FederationSourceBindingDefinition.of(new SourceId("first"), 1, mappings);
            Assert.fail("case-insensitive duplicate schema mappings should be rejected");
        } catch (IllegalArgumentException expected) {
            Assert.assertTrue(expected.getMessage().contains("query schema names"));
        }
    }

    /**
     * 验证 Explain 与指标纯数据视图可完整跨节点序列化。
     *
     * @throws Exception 序列化失败
     */
    @Test
    public void shouldSerializeExplainAndMetricsValueGraphs() throws Exception {
        SourceId sourceId = new SourceId("main");
        FederationColumn column = new FederationColumn(1, "ID", Types.INTEGER, "INTEGER", false);
        FederationPhysicalExplain physical = new FederationPhysicalExplain(
            true,
            "index scan",
            "Index Scan",
            "index",
            List.of("IDX_ID"),
            "IDX_ID",
            1L,
            null,
            "safe"
        );
        FederationFragmentExplain fragment = new FederationFragmentExplain(
            "fragment-1",
            "main",
            sourceId,
            "jdbc",
            "SELECT ID FROM T",
            List.of(),
            List.of(column),
            physical
        );
        AdapterCompatibility compatibility = new AdapterCompatibility(
            AdapterCompatibilityStatus.CODE_SUPPORTED_UNVERIFIED,
            "H2",
            "2",
            "H2 JDBC Driver",
            "2",
            "supported"
        );
        SqlExplainResult explain = new SqlExplainResult(
            SqlExplainLevel.PHYSICAL,
            FederationQueryMode.SINGLE_SOURCE,
            "SELECT ID FROM T",
            "SELECT ID FROM T",
            "LogicalProject",
            "JdbcProject",
            List.of(fragment),
            Set.of(sourceId),
            compatibility,
            true,
            false,
            "executable"
        );
        Assert.assertEquals(explain, roundTrip(explain));

        FederationQueryMetricsSnapshot metrics = new FederationQueryMetricsSnapshot(
            QueryId.create(),
            FederationQueryMode.FEDERATED,
            true,
            10,
            20,
            5,
            1,
            8,
            2,
            16,
            true,
            false,
            List.of(new FederationFragmentMetrics("fragment-1", sourceId, 2, 16, 12, true))
        );
        Assert.assertEquals(metrics, roundTrip(metrics));
    }

    /**
     * 验证物理 Explain 拒绝实际参数值，避免原生计划泄露敏感常量。
     */
    @Test
    public void shouldRejectActualExplainParameterValues() {
        SqlCompileRequest compileRequest = new SqlCompileRequest(
            "SELECT NAME FROM PERSON WHERE ID = ?",
            new SourceId("main"),
            1,
            List.of(Types.INTEGER),
            "policy-v1"
        );
        try {
            new SqlExplainRequest(
                compileRequest,
                SqlExplainLevel.PHYSICAL,
                List.of(new SqlParameter(Types.INTEGER, 7))
            );
            Assert.fail("Explain parameter values should be rejected");
        } catch (IllegalArgumentException expected) {
            Assert.assertTrue(expected.getMessage().contains("does not accept parameter values"));
        }
    }

    private static <T extends Serializable> T roundTrip(T value) throws Exception {
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        try (ObjectOutputStream output = new ObjectOutputStream(bytes)) {
            output.writeObject(value);
        }
        try (ObjectInputStream input = new ObjectInputStream(
            new ByteArrayInputStream(bytes.toByteArray()))) {
            @SuppressWarnings("unchecked")
            T restored = (T) input.readObject();
            return restored;
        }
    }

    private static final class BrokenSerializableValue implements Serializable {

        private final Object nested = new Object();
    }
}
