package changefeed

import (
	"testing"

	"github.com/stretchr/testify/require"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/client/connector_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/model"
)

func TestBuildServerlessInfoForEdit_UsesExistingServerlessInfo(t *testing.T) {
	existing := &model.DataflowServerlessTidbInfo{
		ClusterID: "serverless-1",
		MysqlInfo: &model.DataflowMySQLInfo{
			Connection: &model.DataflowConnection{
				Endpoint: "a",
				Username: "u",
				Password: "p",
			},
		},
	}

	info, err := buildServerlessInfoForEdit(&connector_service.GetConnectorOK{
		Payload: &model.DataflowConnector{
			Request: &model.DataflowConnectorCreateReq{ServerlessTidbInfo: existing},
		},
	}, &TiDBServerlessEntity{ClusterID: "serverless-1"})

	require.NoError(t, err)
	require.Same(t, existing, info)
}

func TestBuildServerlessInfoForEdit_FallsBackToMysqlInfo(t *testing.T) {
	mysqlInfo := &model.DataflowMySQLInfo{
		Connection: &model.DataflowConnection{
			Endpoint: "b",
			Username: "u",
		},
	}

	info, err := buildServerlessInfoForEdit(&connector_service.GetConnectorOK{
		Payload: &model.DataflowConnector{
			Request: &model.DataflowConnectorCreateReq{MysqlInfo: mysqlInfo},
		},
	}, &TiDBServerlessEntity{ClusterID: "serverless-2"})

	require.NoError(t, err)
	require.NotNil(t, info)
	require.Equal(t, "serverless-2", info.ClusterID)
	require.Same(t, mysqlInfo, info.MysqlInfo)
}

func TestBuildServerlessInfoForEdit_ReturnsErrorWhenMysqlInfoMissing(t *testing.T) {
	_, err := buildServerlessInfoForEdit(&connector_service.GetConnectorOK{
		Payload: &model.DataflowConnector{
			Request: &model.DataflowConnectorCreateReq{},
		},
	}, &TiDBServerlessEntity{ClusterID: "serverless-3"})

	require.Error(t, err)
	require.Contains(t, err.Error(), "mysql info is empty")
}

func TestBuildServerlessEditChangefeedReq_UsesMysqlForMysqlConnector(t *testing.T) {
	connectorType := model.DataflowConnectorTypeMYSQL
	connectorID := "cf-1"
	mysqlInfo := &model.DataflowMySQLInfo{
		Connection: &model.DataflowConnection{Endpoint: "mysql"},
	}
	serverlessInfo := &model.DataflowServerlessTidbInfo{
		ClusterID: "serverless-1",
		MysqlInfo: mysqlInfo,
	}

	editReq, err := buildServerlessEditChangefeedReq(&connector_service.GetConnectorOK{
		Payload: &model.DataflowConnector{
			ConnectorID: &connectorID,
			Name:        "cf",
			Request: &model.DataflowConnectorCreateReq{
				ConnectorType: &connectorType,
			},
		},
	}, serverlessInfo)

	require.NoError(t, err)
	require.Same(t, mysqlInfo, editReq.MysqlInfo)
	require.Nil(t, editReq.ServerlessTidbInfo)
}

func TestBuildServerlessEditChangefeedReq_UsesServerlessForServerlessConnector(t *testing.T) {
	connectorType := model.DataflowConnectorTypeSERVERLESSTIDB
	connectorID := "cf-2"
	serverlessInfo := &model.DataflowServerlessTidbInfo{
		ClusterID: "serverless-2",
		MysqlInfo: &model.DataflowMySQLInfo{},
	}

	editReq, err := buildServerlessEditChangefeedReq(&connector_service.GetConnectorOK{
		Payload: &model.DataflowConnector{
			ConnectorID: &connectorID,
			Name:        "cf",
			Request: &model.DataflowConnectorCreateReq{
				ConnectorType: &connectorType,
			},
		},
	}, serverlessInfo)

	require.NoError(t, err)
	require.Same(t, serverlessInfo, editReq.ServerlessTidbInfo)
	require.Nil(t, editReq.MysqlInfo)
}

func TestBuildServerlessEditChangefeedReq_ReturnsErrorForMysqlConnectorWithoutMysqlInfo(t *testing.T) {
	connectorType := model.DataflowConnectorTypeMYSQL
	connectorID := "cf-3"

	_, err := buildServerlessEditChangefeedReq(&connector_service.GetConnectorOK{
		Payload: &model.DataflowConnector{
			ConnectorID: &connectorID,
			Name:        "cf",
			Request: &model.DataflowConnectorCreateReq{
				ConnectorType: &connectorType,
			},
		},
	}, &model.DataflowServerlessTidbInfo{})

	require.Error(t, err)
	require.Contains(t, err.Error(), "mysql info is empty")
}