package changefeed

import (
	"context"
	"fmt"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	config_manager_mock "github.com/tidbcloud/dataflow-service/app/models/config_manager/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	changefeedopenbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/openapi/v1beta1"
	"github.com/tidbcloud/pkgv2/bizerr"
)

func requireBizErrCode(t *testing.T, err error, want bizerr.ErrCode) {
	t.Helper()
	require.Error(t, err)
	type codedErr interface {
		ErrCode() bizerr.ErrCode
	}
	codeErr, ok := err.(codedErr)
	require.True(t, ok)
	require.Equal(t, want, codeErr.ErrCode())
}

func TestParseChangefeedID(t *testing.T) {
	tests := map[string]struct {
		raw     string
		field   string
		want    uint64
		wantErr bool
	}{
		"valid decimal id": {
			raw:   "18446744073709551615",
			field: "changefeed_id",
			want:  18446744073709551615,
		},
		"valid decimal id 2": {
			raw:   "2097143114437660672",
			field: "changefeed_id",
			want:  2097143114437660672,
		},
		"invalid input returns params error": {
			raw:     "abc",
			field:   "changefeed_id",
			wantErr: true,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			got, err := parseUintID(tc.raw, tc.field)
			if tc.wantErr {
				requireBizErrCode(t, err, bizerr.ErrCodeParams)
				require.Contains(t, err.Error(), "changefeed_id is invalid")
				return
			}

			require.NoError(t, err)
			require.Equal(t, tc.want, got)
		})
	}
}

func TestToDataflowKafkaInfo_NetworkInfoNil(t *testing.T) {
	cf := &changefeedopenbp.Changefeed{
		DownstreamConfig: &changefeedopenbp.Changefeed_Kafka{
			Kafka: &changefeedopenbp.KafkaDownstreamConfig{},
		},
	}

	require.NotPanics(t, func() {
		info := toDataflowKafkaInfo(cf, changefeedScope{})
		require.NotNil(t, info)
		require.Zero(t, info.GetBizEndpointId())
	})
}

func TestToDataflowMySQLInfo_NetworkInfoNil(t *testing.T) {
	cf := &changefeedopenbp.Changefeed{
		DownstreamConfig: &changefeedopenbp.Changefeed_Mysql{
			Mysql: &changefeedopenbp.MySQLDownstreamConfig{},
		},
	}

	require.NotPanics(t, func() {
		info := toDataflowMySQLInfo(cf, changefeedScope{})
		require.NotNil(t, info)
		require.Nil(t, info.GetPrivatelinkInfo())
	})
}

func TestMergeChangefeedEdit_ServerlessMissingInfo(t *testing.T) {
	// Use concrete adapter type to access internal merge helper
	a := &connectorChangefeedAdapter{}
	connector := &dataflow.Connector{
		ConnectorId:   123,
		ConnectorType: dataflow.ConnectorType_SERVERLESS_TIDB,
		Request: &dataflow.ConnectorCreateReq{
			ClusterId: 1,
			// ServerlessTidbInfo is intentionally nil to simulate malformed existing config
			MysqlInfo: &dataflow.MySQLInfo{},
		},
	}

	_, err := a.mergeChangefeedEdit(connector, &changefeedopenbp.EditChangefeedDownstreamConfigRequest{
		DownstreamType: changefeedopenbp.DownstreamType_MYSQL,
	})
	requireBizErrCode(t, err, bizerr.ErrCodeParams)
	require.Contains(t, err.Error(), "serverless_tidb_info")
}

func TestAdapterGetChangefeedKernelInfo(t *testing.T) {
	const (
		ticdcClusterID = uint64(67890)
		changefeedID   = uint64(2068955910051614720)
	)
	req := &changefeedopenbp.GetChangefeedKernelInfoRequest{
		ChangefeedId: fmt.Sprint(changefeedID),
	}

	tests := map[string]struct {
		req             *changefeedopenbp.GetChangefeedKernelInfoRequest
		changefeed      *entity.BizChangefeed
		changefeedErr   bizerr.BizError
		ticdcCluster    *entity.TiCDCCluster
		ticdcClusterErr bizerr.BizError
		versionTag      *domain.GetTiCDCClusterVersionAndImageTagResp
		versionTagErr   bizerr.BizError
		wantVersion     string
		wantImageTag    string
		wantKernelType  changefeedopenbp.ChangefeedKernelInfo_KernelType
		wantErrCode     bizerr.ErrCode
		wantErrContains string
	}{
		"classic kernel": {
			changefeed:     &entity.BizChangefeed{ID: entity.ID(changefeedID), TiCDCClusterID: entity.ID(ticdcClusterID)},
			ticdcCluster:   &entity.TiCDCCluster{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(ticdcClusterID)}}, IsNewKernel: false},
			versionTag:     &domain.GetTiCDCClusterVersionAndImageTagResp{Version: "v7.5.6", ImageTag: "v8.5.4-release.1"},
			wantVersion:    "v7.5.6",
			wantImageTag:   "v8.5.4-release.1",
			wantKernelType: changefeedopenbp.ChangefeedKernelInfo_KERNEL_TYPE_CLASSIC,
		},
		"new kernel": {
			changefeed:     &entity.BizChangefeed{ID: entity.ID(changefeedID), TiCDCClusterID: entity.ID(ticdcClusterID)},
			ticdcCluster:   &entity.TiCDCCluster{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(ticdcClusterID)}}, IsNewKernel: true},
			versionTag:     &domain.GetTiCDCClusterVersionAndImageTagResp{Version: "v8.5.4", ImageTag: "v8.5.4-release.1"},
			wantVersion:    "v8.5.4",
			wantImageTag:   "v8.5.4-release.1",
			wantKernelType: changefeedopenbp.ChangefeedKernelInfo_KERNEL_TYPE_NEW,
		},
		"changefeed not found": {
			changefeedErr:   bizerr.New(bizerr.ErrCodeRecordNotFound, "changefeed not found"),
			wantErrCode:     bizerr.ErrCodeRecordNotFound,
			wantErrContains: "changefeed not found",
		},
		"changefeed without bound ticdc cluster": {
			changefeed:      &entity.BizChangefeed{ID: entity.ID(changefeedID)},
			wantErrCode:     bizerr.ErrCodeRecordNotFound,
			wantErrContains: "no bound ticdc cluster",
		},
		"ticdc cluster load error": {
			changefeed:      &entity.BizChangefeed{ID: entity.ID(changefeedID), TiCDCClusterID: entity.ID(ticdcClusterID)},
			ticdcClusterErr: bizerr.New(bizerr.ErrCodeRecordNotFound, "ticdc cluster not found"),
			wantErrCode:     bizerr.ErrCodeRecordNotFound,
			wantErrContains: "ticdc cluster not found",
		},
		"invalid changefeed id": {
			req: &changefeedopenbp.GetChangefeedKernelInfoRequest{
				ChangefeedId: "not-a-number",
			},
			wantErrCode:     bizerr.ErrCodeParams,
			wantErrContains: "changefeed_id is invalid",
		},
		"config center version lookup error": {
			changefeed:      &entity.BizChangefeed{ID: entity.ID(changefeedID), TiCDCClusterID: entity.ID(ticdcClusterID)},
			ticdcCluster:    &entity.TiCDCCluster{Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(ticdcClusterID)}}, IsNewKernel: true},
			versionTagErr:   bizerr.New(bizerr.ErrCodeGRPC, "config center down"),
			wantErrContains: "config center down",
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			bizRepoMock := repo_mock.NewBizChangefeedRepo(t)
			ticdcClusterRepoMock := repo_mock.NewTiCDCClusterRepo(t)
			configMgrMock := config_manager_mock.NewTiCDCClusterConfigManager(t)

			testReq := tc.req
			if testReq == nil {
				testReq = req
			}

			// The changefeed_id parse failure path never reaches the model calls.
			if testReq.GetChangefeedId() != "not-a-number" {
				bizRepoMock.On("Get", mock.Anything, mock.Anything).
					Return(tc.changefeed, tc.changefeedErr).Once()
			}

			if tc.changefeed != nil && tc.changefeed.TiCDCClusterID != 0 && testReq.GetChangefeedId() != "not-a-number" {
				ticdcClusterRepoMock.On("Load", mock.Anything, ticdcClusterID).
					Return(tc.ticdcCluster, tc.ticdcClusterErr).Once()
			}

			if tc.ticdcCluster != nil && testReq.GetChangefeedId() != "not-a-number" {
				configMgrMock.On("GetVersionAndImageTag", mock.Anything,
					&domain.GetTiCDCClusterVersionAndImageTagReq{TiCDCClusterID: ticdcClusterID}).
					Return(tc.versionTag, tc.versionTagErr).Once()
			}

			adapter := newConnectorChangefeedAdapter(
				&connectorServiceStub{}, &cdcInstanceTypeModelStub{}, newNoopBizChangefeedModel(),
				bizRepoMock, ticdcClusterRepoMock, configMgrMock)

			resp, err := adapter.GetChangefeedKernelInfo(context.Background(), testReq)
			if tc.wantErrCode != 0 || tc.wantErrContains != "" {
				require.Error(t, err)
				if tc.wantErrContains != "" {
					require.Contains(t, err.Error(), tc.wantErrContains)
				}
				if tc.wantErrCode != 0 {
					requireBizErrCode(t, err, tc.wantErrCode)
				}
				require.Nil(t, resp)
				return
			}
			require.NoError(t, err)
			require.Equal(t, fmt.Sprint(changefeedID), resp.GetChangefeedId())
			require.Equal(t, tc.wantVersion, resp.GetTicdcVersion())
			require.Equal(t, tc.wantImageTag, resp.GetImageTag())
			require.Equal(t, tc.wantKernelType, resp.GetKernelType())
		})
	}
}