package tech.easyflow.datacenter.execution.service.impl;

import com.mybatisflex.core.row.Row;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
import tech.easyflow.datacenter.connector.DatacenterConnector;
import tech.easyflow.datacenter.connector.DatacenterConnectorRegistry;
import tech.easyflow.datacenter.connector.support.AbstractInternalTableConnector;
import tech.easyflow.datacenter.entity.DatacenterTable;
import tech.easyflow.datacenter.execution.model.DatacenterSqlQueryRequest;
import tech.easyflow.datacenter.execution.model.DatasetRef;
import tech.easyflow.datacenter.mapper.DatacenterCatalogMapper;
import tech.easyflow.datacenter.meta.entity.DatacenterCatalog;
import tech.easyflow.datacenter.meta.entity.DatacenterSource;
import tech.easyflow.datacenter.meta.service.DatacenterDatasetRegistryService;

import java.lang.reflect.Field;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;

/**
 * {@link DatacenterDatasetQueryServiceImpl} 分页 SQL 读取回归测试。
 */
public class DatacenterDatasetQueryServiceImplTest {

    /**
     * 验证惰性迭代器逐页读取并保持原始行顺序。
     *
     * @throws Exception 测试依赖注入失败时抛出
     */
    @Test
    public void consumeBySqlShouldStreamSingleQueryInOrder()
            throws Exception {
        BigInteger sourceId = BigInteger.valueOf(1001L);
        DatacenterSource source = new DatacenterSource();
        source.setId(sourceId);
        source.setSourceType("MYSQL");
        DatacenterTable table = new DatacenterTable();
        table.setId(BigInteger.valueOf(2001L));
        table.setSourceId(sourceId);
        table.setTableName("orders");
        table.setActualTable("orders_actual");

        DatacenterDatasetRegistryService registry =
                Mockito.mock(
                        DatacenterDatasetRegistryService.class);
        Mockito.when(registry.getSourceRequired(sourceId))
                .thenReturn(source);
        Mockito.when(registry.listManagedTables(
                        sourceId, null))
                .thenReturn(List.of(table));
        DatacenterConnector connector =
                Mockito.mock(DatacenterConnector.class);
        Mockito.doAnswer(invocation -> {
                    @SuppressWarnings("unchecked")
                    Consumer<Row> consumer =
                            invocation.getArgument(3);
                    consumer.accept(row(1));
                    consumer.accept(row(2));
                    consumer.accept(row(3));
                    return null;
                })
                .when(connector)
                .consumeBySql(
                        ArgumentMatchers.eq(source),
                        ArgumentMatchers.anyString(),
                        ArgumentMatchers.eq(2),
                        ArgumentMatchers.any());
        DatacenterConnectorRegistry connectors =
                Mockito.mock(
                        DatacenterConnectorRegistry.class);
        Mockito.when(connectors.getConnector("MYSQL"))
                .thenReturn(connector);

        DatacenterDatasetQueryServiceImpl service =
                new DatacenterDatasetQueryServiceImpl();
        setField(service, "registryService", registry);
        setField(service, "connectorRegistry", connectors);
        DatacenterSqlQueryRequest request =
                new DatacenterSqlQueryRequest();
        DatasetRef datasetRef = new DatasetRef();
        datasetRef.setSourceId(sourceId);
        request.setDatasetRef(datasetRef);
        request.setSql("SELECT * FROM orders ORDER BY id");

        List<Integer> ids = new ArrayList<>();
        service.consumeBySql(
                request,
                2,
                current -> ids.add(
                        current.getInt("id")));

        Assert.assertEquals(
                List.of(1, 2, 3), ids);
        Mockito.verify(connector, Mockito.times(1))
                .consumeBySql(
                        ArgumentMatchers.eq(source),
                        ArgumentMatchers.anyString(),
                        ArgumentMatchers.eq(2),
                        ArgumentMatchers.any());
    }

    /**
     * 验证内部 Excel 查询移除逻辑目录并转换 MySQL 标识符引号。
     *
     * @throws Exception 测试依赖注入失败时抛出
     */
    @Test
    public void queryBySqlShouldNormalizeInternalMysqlSql()
            throws Exception {
        BigInteger sourceId = BigInteger.valueOf(3001L);
        BigInteger catalogId = BigInteger.valueOf(3002L);
        DatacenterSource source = new DatacenterSource();
        source.setId(sourceId);
        source.setSourceType("EXCEL");
        DatacenterTable table = new DatacenterTable();
        table.setId(BigInteger.valueOf(3003L));
        table.setSourceId(sourceId);
        table.setCatalogId(catalogId);
        table.setTableName("Sheet1");
        table.setMaterializedTable("tb_excel_budget");

        DatacenterCatalog catalog = new DatacenterCatalog();
        catalog.setId(catalogId);
        catalog.setCatalogName("ama 实验基线模型预算");
        DatacenterDatasetRegistryService registry =
                Mockito.mock(DatacenterDatasetRegistryService.class);
        Mockito.when(registry.getSourceRequired(sourceId))
                .thenReturn(source);
        Mockito.when(registry.listManagedTables(sourceId, null))
                .thenReturn(List.of(table));
        DatacenterCatalogMapper catalogMapper =
                Mockito.mock(DatacenterCatalogMapper.class);
        Mockito.when(catalogMapper.selectListByQuery(
                        ArgumentMatchers.any()))
                .thenReturn(List.of(catalog));
        AbstractInternalTableConnector connector =
                Mockito.mock(AbstractInternalTableConnector.class);
        DatacenterConnectorRegistry connectors =
                Mockito.mock(DatacenterConnectorRegistry.class);
        Mockito.when(connectors.getConnector("EXCEL"))
                .thenReturn(connector);

        DatacenterDatasetQueryServiceImpl service =
                new DatacenterDatasetQueryServiceImpl();
        setField(service, "registryService", registry);
        setField(service, "connectorRegistry", connectors);
        setField(service, "catalogMapper", catalogMapper);
        DatacenterSqlQueryRequest request =
                new DatacenterSqlQueryRequest();
        DatasetRef datasetRef = new DatasetRef();
        datasetRef.setSourceId(sourceId);
        request.setDatasetRef(datasetRef);
        request.setSql("""
                SELECT "token", "token_1"
                FROM "ama 实验基线模型预算"."Sheet1"
                WHERE "col_id" = 'deepseek-v4-pro'
                """);

        service.queryBySql(request);

        ArgumentCaptor<String> sqlCaptor =
                ArgumentCaptor.forClass(String.class);
        Mockito.verify(connector).queryBySql(
                ArgumentMatchers.eq(source),
                sqlCaptor.capture());
        Assert.assertEquals(
                "SELECT `token`, `token_1` FROM tb_excel_budget "
                        + "WHERE `col_id` = 'deepseek-v4-pro'",
                sqlCaptor.getValue());
    }

    /**
     * 创建测试数据行。
     *
     * @param id 行 ID
     * @return 数据行
     */
    private Row row(int id) {
        Row row = new Row();
        row.put("id", id);
        return row;
    }

    /**
     * 注入被测服务依赖。
     *
     * @param target 被测对象
     * @param name 字段名
     * @param value 字段值
     * @throws Exception 反射失败时抛出
     */
    private void setField(
            Object target,
            String name,
            Object value) throws Exception {
        Field field = target.getClass()
                .getDeclaredField(name);
        field.setAccessible(true);
        field.set(target, value);
    }
}
