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

import com.alibaba.fastjson2.JSONObject;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import tech.easyflow.common.cache.RedisIdempotencyExecutor;
import tech.easyflow.datacenter.connector.DatacenterConnector;
import tech.easyflow.datacenter.connector.DatacenterConnectorRegistry;
import tech.easyflow.datacenter.entity.DatacenterTable;
import tech.easyflow.datacenter.execution.model.DatasetRef;
import tech.easyflow.datacenter.meta.entity.DatacenterSource;
import tech.easyflow.datacenter.meta.enums.DatacenterSourceType;
import tech.easyflow.datacenter.meta.service.DatacenterDatasetRegistryService;

import java.lang.reflect.Field;
import java.math.BigInteger;
import java.util.List;

import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

/**
 * 数据集写入服务的幂等批写测试。
 */
public class DatacenterDatasetWriteServiceImplTest {

    /**
     * 验证服务只调用一次连接器，并保留配置的批大小供连接器复用连接处理。
     *
     * @throws Exception 测试依赖注入失败时抛出
     */
    @Test
    public void shouldDelegateIdempotentRowsInSingleConnectorCall() throws Exception {
        DatacenterDatasetRegistryService registryService =
                mock(DatacenterDatasetRegistryService.class);
        DatacenterConnectorRegistry connectorRegistry =
                mock(DatacenterConnectorRegistry.class);
        DatacenterConnector connector = mock(DatacenterConnector.class);
        RedisIdempotencyExecutor idempotencyExecutor =
                mock(RedisIdempotencyExecutor.class);

        BigInteger tableId = BigInteger.ONE;
        BigInteger sourceId = BigInteger.TWO;
        DatasetRef datasetRef = new DatasetRef();
        datasetRef.setTableId(tableId);
        DatacenterTable table = new DatacenterTable();
        table.setSourceId(sourceId);
        DatacenterSource source = new DatacenterSource();
        source.setSourceType(DatacenterSourceType.PROJECT_MYSQL.name());
        when(registryService.getTableWithFields(tableId)).thenReturn(table);
        when(registryService.getSourceRequired(sourceId)).thenReturn(source);
        when(connectorRegistry.getConnector(
                DatacenterSourceType.PROJECT_MYSQL.name())).thenReturn(connector);
        when(idempotencyExecutor.executeOnce(
                anyString(), anyString(), any(Runnable.class)))
                .thenAnswer(invocation -> {
                    invocation.<Runnable>getArgument(2).run();
                    return true;
                });
        when(connector.saveRowsIdempotently(
                any(), any(), anyList(), any(), anyInt(), anyString(), anyString()))
                .thenReturn(true);

        DatacenterDatasetWriteServiceImpl service =
                new DatacenterDatasetWriteServiceImpl();
        inject(service, "registryService", registryService);
        inject(service, "connectorRegistry", connectorRegistry);
        inject(service, "idempotencyExecutor", idempotencyExecutor);
        List<JSONObject> rows = List.of(
                row("row-0"), row("row-1"), row("row-2"));

        Assert.assertTrue(service.saveRowsIdempotently(
                datasetRef, rows, null, 64, "stable-execution-key"));

        @SuppressWarnings("unchecked")
        ArgumentCaptor<List<JSONObject>> rowsCaptor =
                ArgumentCaptor.forClass(List.class);
        verify(connector, times(1)).saveRowsIdempotently(
                eq(source),
                eq(table),
                rowsCaptor.capture(),
                any(),
                eq(64),
                anyString(),
                anyString());
        Assert.assertEquals(rows, rowsCaptor.getValue());
    }

    /**
     * 创建测试行。
     *
     * @param name 行名称
     * @return JSON 行
     */
    private JSONObject row(String name) {
        JSONObject row = new JSONObject();
        row.put("name", name);
        return row;
    }

    /**
     * 注入服务测试依赖。
     *
     * @param target 目标服务
     * @param fieldName 字段名
     * @param value 字段值
     * @throws Exception 反射访问失败时抛出
     */
    private void inject(Object target, String fieldName, Object value) throws Exception {
        Field field = target.getClass().getDeclaredField(fieldName);
        field.setAccessible(true);
        field.set(target, value);
    }
}
