package cloudconfig

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/devops/pkg/infra/clients/dataflowcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/serverlesscli"
	ccsvcapi "github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	serverlessrpc "github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"go.uber.org/mock/gomock"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type connectorListStub struct {
	dataflow.ConnectorServiceClient
	resp           *dataflow.Connectors
	req            *dataflow.ConnectorListReq
	err            error
	waitForContext bool

	getResp *dataflow.Connector
	getReq  *dataflow.ConnectorID
	getErr  error
}

func (s *connectorListStub) ListConnectors(
	ctx context.Context,
	req *dataflow.ConnectorListReq,
	_ ...grpc.CallOption,
) (*dataflow.Connectors, error) {
	s.req = req
	if s.waitForContext {
		<-ctx.Done()
		return nil, ctx.Err()
	}
	return s.resp, s.err
}

func (s *connectorListStub) GetConnector(
	_ context.Context,
	req *dataflow.ConnectorID,
	_ ...grpc.CallOption,
) (*dataflow.Connector, error) {
	s.getReq = req
	return s.getResp, s.getErr
}

type internalChangefeedListStub struct {
	dataflow.OpsInternalServiceClient
	resp *dataflow.ListInternalChangefeedsResponse
	req  *dataflow.ListInternalChangefeedsRequest
	err  error

	getResp *dataflow.InternalChangefeed
	getReq  *dataflow.GetInternalChangefeedRequest
	getErr  error
}

func (s *internalChangefeedListStub) ListInternalChangefeeds(
	_ context.Context,
	req *dataflow.ListInternalChangefeedsRequest,
	_ ...grpc.CallOption,
) (*dataflow.ListInternalChangefeedsResponse, error) {
	s.req = req
	return s.resp, s.err
}

func (s *internalChangefeedListStub) GetInternalChangefeed(
	_ context.Context,
	req *dataflow.GetInternalChangefeedRequest,
	_ ...grpc.CallOption,
) (*dataflow.InternalChangefeed, error) {
	s.getReq = req
	return s.getResp, s.getErr
}

type ngChangefeedListStub struct {
	dataflowng.ChangefeedServiceClient
	resp           *dataflowng.Changefeeds
	req            *dataflowng.ChangefeedListReq
	err            error
	waitForContext bool

	getResp *dataflowng.Changefeed
	getReq  *dataflowng.ChangefeedID
	getErr  error
}

func (s *ngChangefeedListStub) ListChangefeeds(
	ctx context.Context,
	req *dataflowng.ChangefeedListReq,
	_ ...grpc.CallOption,
) (*dataflowng.Changefeeds, error) {
	s.req = req
	if s.waitForContext {
		<-ctx.Done()
		return nil, ctx.Err()
	}
	return s.resp, s.err
}

func (s *ngChangefeedListStub) GetChangefeed(
	_ context.Context,
	req *dataflowng.ChangefeedID,
	_ ...grpc.CallOption,
) (*dataflowng.Changefeed, error) {
	s.getReq = req
	return s.getResp, s.getErr
}

type ngInternalChangefeedListStub struct {
	dataflowng.OpsInternalServiceClient
	resp *dataflowng.ListInternalChangefeedsResponse
	req  *dataflowng.ListInternalChangefeedsRequest
	err  error

	getResp *dataflowng.InternalChangefeed
	getReq  *dataflowng.GetInternalChangefeedRequest
	getErr  error
}

func (s *ngInternalChangefeedListStub) ListInternalChangefeeds(
	_ context.Context,
	req *dataflowng.ListInternalChangefeedsRequest,
	_ ...grpc.CallOption,
) (*dataflowng.ListInternalChangefeedsResponse, error) {
	s.req = req
	return s.resp, s.err
}

func (s *ngInternalChangefeedListStub) GetInternalChangefeed(
	_ context.Context,
	req *dataflowng.GetInternalChangefeedRequest,
	_ ...grpc.CallOption,
) (*dataflowng.InternalChangefeed, error) {
	s.getReq = req
	return s.getResp, s.getErr
}

func TestListInternalChangefeeds(t *testing.T) {
	now := time.Now()
	t.Run("dedicated cluster only calls dedicated Dataflow", func(t *testing.T) {
		dedicatedBusiness := &connectorListStub{resp: &dataflow.Connectors{
			Items: []*dataflow.Connector{{
				ConnectorId:   201,
				Name:          "dedicated",
				ConnectorType: dataflow.ConnectorType_KAFKA,
				Status:        dataflow.Connector_RUNNING,
				CreatedAt:     timestamppb.New(now),
				CheckpointTso: 12345,
				Request: &dataflow.ConnectorCreateReq{
					ClusterId: 100,
					KafkaInfo: &dataflow.KafkaInfo{
						DataFormat: &dataflow.DataFormat{Protocol: dataflow.Protocol_CANAL_JSON},
					},
				},
			}},
		}}
		dedicatedInternal := &internalChangefeedListStub{
			resp: &dataflow.ListInternalChangefeedsResponse{
				Changefeeds: []*dataflow.InternalChangefeed{{
					Id:             301,
					DisplayId:      201,
					TicdcCluster:   &dataflow.InternalTiCDCCluster{Id: 401, Name: "ticdc"},
					TidbClusterId:  100,
					DisplayName:    "stale-name",
					DownstreamType: "stale-type",
					Status:         "stale-status",
				}},
			},
		}
		ngBusiness := &ngChangefeedListStub{}
		ngInternal := &ngInternalChangefeedListStub{}
		service := &svc{dataflow: &dataflowcli.DataflowCli{
			Connector:    dedicatedBusiness,
			Ops:          dedicatedInternal,
			NgChangefeed: ngBusiness,
			NgOps:        ngInternal,
		}}

		resp, err := service.ListInternalChangefeeds(context.Background(), ccsvcapi.ListInternalChangefeedsParam{
			TiDBClusterID: 100,
			ClusterType:   ccproto.ClusterType_name[int32(ccproto.ClusterType_DEDICATED_TIDB)],
		})
		require.NoError(t, err)
		require.Len(t, resp.Changefeeds, 1)
		require.Equal(t, "301", resp.Changefeeds[0].Id)
		require.Equal(t, "201", resp.Changefeeds[0].DisplayId)
		require.Equal(t, "dedicated", resp.Changefeeds[0].DisplayName)
		require.Equal(t, "normal", resp.Changefeeds[0].Status)
		require.Equal(t, "CHANGEFEED", resp.Changefeeds[0].ClusterType)
		require.Equal(t, "401", resp.Changefeeds[0].TicdcCluster.Id)
		require.Equal(t, "12345", resp.Changefeeds[0].CheckpointTSO)
		require.Equal(t, "CANAL_JSON", resp.Changefeeds[0].DownstreamConfig["data_format"].(map[string]interface{})["protocol"])
		require.Equal(t, uint64(100), dedicatedBusiness.req.ClusterId)
		require.Equal(t, dataflow.ConnectorType_TOTAL, dedicatedBusiness.req.ConnectorType)
		require.Equal(t, uint64(100), dedicatedInternal.req.TidbClusterId)
		require.Nil(t, ngBusiness.req)
		require.Nil(t, ngInternal.req)
	})

	t.Run("premium cluster only calls next-gen Dataflow", func(t *testing.T) {
		ngBusiness := &ngChangefeedListStub{resp: &dataflowng.Changefeeds{
			Items: []*dataflowng.Changefeed{{
				ChangefeedId:   202,
				Name:           "creating",
				ChangefeedType: dataflowng.ChangefeedType_MYSQL,
				Status:         dataflowng.Changefeed_CREATING,
				CreatedAt:      timestamppb.New(now),
				CheckpointTso:  67890,
				Request: &dataflowng.ChangefeedCreateReq{
					ClusterId: 100,
					MysqlInfo: &dataflowng.MySQLInfo{
						Connection: &dataflowng.Connection{Endpoint: "mysql.example.com:3306"},
					},
				},
			}},
		}}
		ngInternal := &ngInternalChangefeedListStub{
			resp: &dataflowng.ListInternalChangefeedsResponse{},
		}
		dedicatedBusiness := &connectorListStub{}
		dedicatedInternal := &internalChangefeedListStub{}
		service := &svc{dataflow: &dataflowcli.DataflowCli{
			Connector:    dedicatedBusiness,
			Ops:          dedicatedInternal,
			NgChangefeed: ngBusiness,
			NgOps:        ngInternal,
		}}

		resp, err := service.ListInternalChangefeeds(context.Background(), ccsvcapi.ListInternalChangefeedsParam{
			TiDBClusterID: 100,
			ClusterType:   ccproto.ClusterType_name[int32(ccproto.ClusterType_LOGICAL_CLUSTER)],
		})
		require.NoError(t, err)
		require.Len(t, resp.Changefeeds, 1)
		require.Equal(t, "", resp.Changefeeds[0].Id)
		require.Equal(t, "202", resp.Changefeeds[0].DisplayId)
		require.Equal(t, "creating", resp.Changefeeds[0].Status)
		require.Equal(t, "CHANGEFEED_NG", resp.Changefeeds[0].ClusterType)
		require.Equal(t, "67890", resp.Changefeeds[0].CheckpointTSO)
		require.Equal(t, "mysql.example.com:3306", resp.Changefeeds[0].DownstreamConfig["connection"].(map[string]interface{})["endpoint"])
		require.Equal(t, uint64(100), ngBusiness.req.ClusterId)
		require.Equal(t, dataflowng.ChangefeedType_TOTAL, ngBusiness.req.ChangefeedType)
		require.Equal(t, uint64(100), ngInternal.req.TidbClusterId)
		require.Nil(t, dedicatedBusiness.req)
		require.Nil(t, dedicatedInternal.req)
	})
}

func TestListInternalChangefeedsTargetFailures(t *testing.T) {
	t.Run("dedicated failure is not masked by empty next-gen result", func(t *testing.T) {
		dedicatedBusiness := &connectorListStub{err: errors.New("dedicated unavailable")}
		ngBusiness := &ngChangefeedListStub{resp: &dataflowng.Changefeeds{}}
		service := &svc{dataflow: &dataflowcli.DataflowCli{
			Connector:    dedicatedBusiness,
			Ops:          &internalChangefeedListStub{},
			NgChangefeed: ngBusiness,
			NgOps:        &ngInternalChangefeedListStub{},
		}}

		resp, err := service.ListInternalChangefeeds(context.Background(), ccsvcapi.ListInternalChangefeedsParam{
			TiDBClusterID: 100,
			ClusterType:   ccproto.ClusterType_name[int32(ccproto.ClusterType_DEDICATED_TIDB)],
		})
		require.Nil(t, resp)
		require.ErrorContains(t, err, "list dedicated business changefeeds")
		require.ErrorContains(t, err, "dedicated unavailable")
		require.Nil(t, ngBusiness.req)
	})

	t.Run("next-gen failure is not masked by empty dedicated result", func(t *testing.T) {
		dedicatedBusiness := &connectorListStub{resp: &dataflow.Connectors{}}
		ngBusiness := &ngChangefeedListStub{err: errors.New("next-gen unavailable")}
		service := &svc{dataflow: &dataflowcli.DataflowCli{
			Connector:    dedicatedBusiness,
			Ops:          &internalChangefeedListStub{},
			NgChangefeed: ngBusiness,
			NgOps:        &ngInternalChangefeedListStub{},
		}}

		resp, err := service.ListInternalChangefeeds(context.Background(), ccsvcapi.ListInternalChangefeedsParam{
			TiDBClusterID: 100,
			ClusterType:   ccproto.ClusterType_name[int32(ccproto.ClusterType_SHARED_POOL)],
		})
		require.Nil(t, resp)
		require.ErrorContains(t, err, "list next-gen business changefeeds")
		require.ErrorContains(t, err, "next-gen unavailable")
		require.Nil(t, dedicatedBusiness.req)
	})
}

func TestListInternalChangefeedsKeepsBusinessResultsWhenEnrichmentFails(t *testing.T) {
	service := &svc{dataflow: &dataflowcli.DataflowCli{
		Connector: &connectorListStub{resp: &dataflow.Connectors{
			Items: []*dataflow.Connector{{
				ConnectorId: 201,
				Name:        "dedicated",
				Request:     &dataflow.ConnectorCreateReq{ClusterId: 100},
			}},
		}},
		Ops: &internalChangefeedListStub{err: errors.New("internal unavailable")},
	}}

	resp, err := service.ListInternalChangefeeds(context.Background(), ccsvcapi.ListInternalChangefeedsParam{
		TiDBClusterID: 100,
		ClusterType:   ccproto.ClusterType_name[int32(ccproto.ClusterType_DEDICATED_TIDB)],
	})
	require.NoError(t, err)
	require.Len(t, resp.Changefeeds, 1)
	require.Equal(t, "", resp.Changefeeds[0].Id)
	require.Equal(t, "201", resp.Changefeeds[0].DisplayId)
}

func TestListInternalChangefeedsUsesBoundedContext(t *testing.T) {
	service := &svc{dataflow: &dataflowcli.DataflowCli{
		Connector: &connectorListStub{waitForContext: true},
		Ops:       &internalChangefeedListStub{},
	}}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()

	start := time.Now()
	resp, err := service.ListInternalChangefeeds(ctx, ccsvcapi.ListInternalChangefeedsParam{
		TiDBClusterID: 100,
		ClusterType:   ccproto.ClusterType_name[int32(ccproto.ClusterType_DEDICATED_TIDB)],
	})
	require.Nil(t, resp)
	require.ErrorIs(t, err, context.DeadlineExceeded)
	require.Less(t, time.Since(start), time.Second)
}

func TestGetInternalChangefeed(t *testing.T) {
	t.Run("dedicated", func(t *testing.T) {
		connector := &connectorListStub{getResp: &dataflow.Connector{
			ConnectorId:   201,
			Name:          "dedicated",
			ConnectorType: dataflow.ConnectorType_KAFKA,
			CheckpointTso: 12345,
			Request: &dataflow.ConnectorCreateReq{
				ClusterId: 100,
				KafkaInfo: &dataflow.KafkaInfo{
					DataFormat: &dataflow.DataFormat{Protocol: dataflow.Protocol_CANAL_JSON},
				},
			},
		}}
		internal := &internalChangefeedListStub{getResp: &dataflow.InternalChangefeed{
			Id:           301,
			DisplayId:    201,
			TicdcCluster: &dataflow.InternalTiCDCCluster{Id: 401},
		}}
		service := &svc{dataflow: &dataflowcli.DataflowCli{
			Connector: connector,
			Ops:       internal,
		}}

		result, err := service.GetInternalChangefeed(context.Background(), ccsvcapi.GetInternalChangefeedParam{
			ChangefeedID: 301,
			ClusterType:  ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED)],
		})
		require.NoError(t, err)
		require.Equal(t, uint64(100), result.TiDBClusterID)
		require.Equal(t, "301", result.Changefeed.Id)
		require.Equal(t, "201", result.Changefeed.DisplayId)
		require.Equal(t, "401", result.Changefeed.TicdcCluster.Id)
		require.Equal(t, "12345", result.Changefeed.CheckpointTSO)
		require.Equal(t, "CANAL_JSON", result.Changefeed.DownstreamConfig["data_format"].(map[string]interface{})["protocol"])
		require.Equal(t, uint64(301), internal.getReq.ChangefeedId)
		require.Equal(t, uint64(201), connector.getReq.ConnectorId)
	})

	t.Run("next-gen", func(t *testing.T) {
		changefeed := &ngChangefeedListStub{getResp: &dataflowng.Changefeed{
			ChangefeedId:   202,
			Name:           "next-gen",
			ChangefeedType: dataflowng.ChangefeedType_MYSQL,
			CheckpointTso:  67890,
			Request: &dataflowng.ChangefeedCreateReq{
				ClusterId: 100,
				MysqlInfo: &dataflowng.MySQLInfo{
					Connection: &dataflowng.Connection{Endpoint: "mysql.example.com:3306"},
				},
			},
		}}
		internal := &ngInternalChangefeedListStub{getResp: &dataflowng.InternalChangefeed{
			Id:           302,
			DisplayId:    202,
			TicdcCluster: &dataflowng.InternalTiCDCCluster{Id: 402},
		}}
		service := &svc{dataflow: &dataflowcli.DataflowCli{
			NgChangefeed: changefeed,
			NgOps:        internal,
		}}

		result, err := service.GetInternalChangefeed(context.Background(), ccsvcapi.GetInternalChangefeedParam{
			ChangefeedID: 302,
			ClusterType:  ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED_NG)],
		})
		require.NoError(t, err)
		require.Equal(t, uint64(100), result.TiDBClusterID)
		require.Equal(t, "302", result.Changefeed.Id)
		require.Equal(t, "202", result.Changefeed.DisplayId)
		require.Equal(t, "402", result.Changefeed.TicdcCluster.Id)
		require.Equal(t, "67890", result.Changefeed.CheckpointTSO)
		require.Equal(t, uint64(302), internal.getReq.ChangefeedId)
		require.Equal(t, uint64(202), changefeed.getReq.ChangefeedId)
	})
}

func TestEssentialInternalChangefeeds(t *testing.T) {
	ctrl := gomock.NewController(t)
	changefeedOps := serverlesscli.NewMockChangefeedOpsServiceClient(ctrl)
	service := &svc{serverlessCli: serverlesscli.NewTestClient(nil, nil, changefeedOps)}

	changefeedOps.EXPECT().ListClusterChangefeeds(gomock.Any(), gomock.Any()).
		DoAndReturn(func(_ context.Context, req *serverlessrpc.ListClusterChangefeedsRequest, _ ...grpc.CallOption) (*serverlessrpc.ListClusterChangefeedsResponse, error) {
			require.Equal(t, "100", req.GetClusterId())
			return &serverlessrpc.ListClusterChangefeedsResponse{Items: []*serverlessrpc.ChangefeedInfo{{
				ClusterId:        "100",
				ChangefeedId:     "cfd-essential",
				DisplayName:      "essential",
				State:            "running",
				Arch:             "ticdc_newarch",
				SinkType:         "mysql",
				TicdcClusterName: "ess-ticdc-us-east-1-test",
				CreatedAt:        "2026-08-17T08:00:00Z",
				CheckpointTso:    12345,
				CheckpointTime:   "2026-08-17T08:01:00Z",
				Rcu:              4,
			}}}, nil
		})

	resp, err := service.ListInternalChangefeeds(context.Background(), ccsvcapi.ListInternalChangefeedsParam{
		TiDBClusterID: 100,
		ClusterType:   ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED_ESSENTIAL)],
	})
	require.NoError(t, err)
	require.Len(t, resp.Changefeeds, 1)
	require.Equal(t, "cfd-essential", resp.Changefeeds[0].Id)
	require.Equal(t, "ticdc_newarch", resp.Changefeeds[0].Arch)
	require.Equal(t, "ess-ticdc-us-east-1-test", resp.Changefeeds[0].TiCDCClusterName)
	require.Equal(t, "12345", resp.Changefeeds[0].CheckpointTSO)
	require.Equal(t, "4", resp.Changefeeds[0].RCU)

	changefeedOps.EXPECT().GetChangefeed(gomock.Any(), gomock.Any()).
		DoAndReturn(func(_ context.Context, req *serverlessrpc.GetChangefeedRequest, _ ...grpc.CallOption) (*serverlessrpc.ChangefeedInfo, error) {
			require.Equal(t, "cfd-essential", req.GetChangefeedId())
			return &serverlessrpc.ChangefeedInfo{
				ClusterId:    "100",
				ChangefeedId: "cfd-essential",
				Arch:         "replication_worker",
			}, nil
		})

	result, err := service.GetInternalChangefeed(context.Background(), ccsvcapi.GetInternalChangefeedParam{
		ChangefeedIDString: "cfd-essential",
		ClusterType:        ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED_ESSENTIAL)],
	})
	require.NoError(t, err)
	require.Equal(t, uint64(100), result.TiDBClusterID)
	require.Equal(t, "cfd-essential", result.Changefeed.Id)
	require.Equal(t, "replication_worker", result.Changefeed.Arch)
}