package ops_svc

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/service/mock_service"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"go.uber.org/mock/gomock"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/structpb"
)

func TestToChangefeedInfo(t *testing.T) {
	createTime := time.Date(2026, 7, 1, 8, 0, 0, 0, time.UTC)
	checkpointTime := time.Date(2026, 7, 9, 3, 30, 0, 0, time.UTC)
	cf := &vo.Changefeed{
		ClusterID:     "1001",
		ChangefeedID:  "cf-abc",
		State:         constant.ChangfeedStateRunning,
		Arch:          constant.ChangefeedArchTiCDCNewArch,
		RCU:           3,
		Name:          "my-feed",
		Sink:          &vo.SinkInfo{Type: constant.ChangefeedTypeKafka},
		Checkpoint:    &vo.CDCCheckpoint{TSO: 456, Time: &checkpointTime},
		Error:         "boom",
		CreateTime:    &createTime,
		CloudProvider: "aws",
		Region:        "us-east-1",
		Kubernetes:    "k8s-1",
		StartPosition: &vo.StartPosition{Mode: constant.StartModeFromTSO, TSO: 123},
	}

	info := toChangefeedInfo(cf)
	require.Equal(t, "aws", info.GetProvider())
	require.Equal(t, "us-east-1", info.GetRegion())
	require.Equal(t, "k8s-1", info.GetKubernetes())
	require.Equal(t, "1001", info.GetClusterId())
	require.Equal(t, "cf-abc", info.GetChangefeedId())
	require.Equal(t, string(constant.ChangfeedStateRunning), info.GetState())
	require.Equal(t, string(constant.ChangefeedArchTiCDCNewArch), info.GetArch())
	require.Equal(t, uint32(3), info.GetRcu())
	require.Equal(t, "my-feed", info.GetDisplayName())
	require.Equal(t, "kafka", info.GetSinkType())
	require.Equal(t, uint64(456), info.GetCheckpointTso())
	require.Equal(t, uint64(123), info.GetStartTso())
	require.Equal(t, "2026-07-09T03:30:00Z", info.GetCheckpointTime())
	require.Equal(t, "boom", info.GetError())
	require.Equal(t, "2026-07-01T08:00:00Z", info.GetCreatedAt())
	require.Empty(t, info.GetUpdatedAt())
}

func TestToMigrateChangefeedResponse(t *testing.T) {
	resp := toMigrateChangefeedResponse(&vo.ChangefeedMigrationResult{
		SourceChangefeedID:  "cf-old",
		SourceCheckpointTSO: 123,
		Target: &vo.Changefeed{
			ClusterID: "1001", ChangefeedID: "cf-new",
			Arch: constant.ChangefeedArchTiCDCNewArch, State: constant.ChangefeedStateCreating,
		},
	})
	require.Equal(t, "cf-old", resp.GetSourceChangefeedId())
	require.Equal(t, "cf-new", resp.GetTargetChangefeedId())
	require.Equal(t, uint64(123), resp.GetSourceCheckpointTso())
	require.Equal(t, uint64(123), resp.GetTargetStartTso())
}

func TestToChangefeedInfo_NilFields(t *testing.T) {
	info := toChangefeedInfo(nil)
	require.NotNil(t, info)

	info = toChangefeedInfo(&vo.Changefeed{ClusterID: "1001"})
	require.Equal(t, "1001", info.GetClusterId())
	require.Empty(t, info.GetSinkType())
	require.Zero(t, info.GetCheckpointTso())
	require.Empty(t, info.GetCheckpointTime())
}

func TestListClusterChangefeeds_DefaultLimit(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		ListChangefeeds(gomock.Any(), &vo.ListChangefeedsOptions{
			ClusterID: "1001",
			Limit:     defaultListChangefeedsLimit,
		}).
		Return([]*vo.Changefeed{
			{ClusterID: "1001", ChangefeedID: "cf-1"},
			{ClusterID: "1001", ChangefeedID: "cf-2"},
		}, uint32(2), nil)

	resp, err := svc.ListClusterChangefeeds(context.Background(), &rpc.ListClusterChangefeedsRequest{
		ClusterId: "1001",
	})
	require.NoError(t, err)
	require.Len(t, resp.GetItems(), 2)
	require.Equal(t, uint32(2), resp.GetTotal())
	require.Equal(t, "cf-1", resp.GetItems()[0].GetChangefeedId())
}

func TestListClusterChangefeeds_ExplicitPaging(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		ListChangefeeds(gomock.Any(), &vo.ListChangefeedsOptions{
			ClusterID: "1001",
			SinkType:  "kafka",
			Limit:     10,
			Offset:    20,
		}).
		Return([]*vo.Changefeed{}, uint32(0), nil)

	resp, err := svc.ListClusterChangefeeds(context.Background(), &rpc.ListClusterChangefeedsRequest{
		ClusterId: "1001",
		SinkType:  "kafka",
		Limit:     10,
		Offset:    20,
	})
	require.NoError(t, err)
	require.Empty(t, resp.GetItems())
}

func TestApplyConfigChanges(t *testing.T) {
	tests := []struct {
		name                  string
		configurationFile     rpc.ConfigFileType
		configurationFilename string
		templateID            *int64
	}{
		{
			name:                  "changefeed file",
			configurationFile:     rpc.ConfigFileType_CHANGEFEED_FILE,
			configurationFilename: "changefeed",
			templateID:            pointerToInt64(42),
		},
		{
			name:                  "sink uri",
			configurationFile:     rpc.ConfigFileType_SINK_URI,
			configurationFilename: "changefeed-uri",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			cdcSvc := mock_service.NewMockCDCService(ctrl)
			svc := NewCdcRpcService(cdcSvc)

			value := structpb.NewNumberValue(256)
			cdcSvc.EXPECT().ApplyChangefeedConfiguration(
				gomock.Any(),
				"1001",
				"cf-1",
				tt.configurationFilename,
				map[string]*config_center.ParamItem{
					"memory_quota": {Value: value, Comment: "ops override"},
				},
				tt.templateID,
			).Return(nil)

			resp, err := svc.ApplyConfigChanges(context.Background(), &rpc.ApplyConfigChangesRequest{
				ClusterType:       rpc.ConfigClusterType_CHANGEFEED_CLUSTER,
				ClusterId:         "1001",
				ChangefeedId:      "cf-1",
				ConfigurationFile: tt.configurationFile,
				CustomParams: map[string]*rpc.ParamItem{
					"memory_quota": {Value: value, Comment: "ops override"},
				},
				TemplateId: tt.templateID,
			})
			require.NoError(t, err)
			require.NotNil(t, resp)
		})
	}
}

func pointerToInt64(value int64) *int64 {
	return &value
}

func TestApplyConfigChangesRejectsUnsupportedSelectors(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	_, err := svc.ApplyConfigChanges(context.Background(), &rpc.ApplyConfigChangesRequest{
		ClusterType:       rpc.ConfigClusterType_TICDC_CLUSTER,
		ConfigurationFile: rpc.ConfigFileType_CHANGEFEED_FILE,
	})
	require.Equal(t, codes.InvalidArgument, status.Code(err))

	_, err = svc.ApplyConfigChanges(context.Background(), &rpc.ApplyConfigChangesRequest{
		ClusterType:       rpc.ConfigClusterType_CHANGEFEED_CLUSTER,
		ConfigurationFile: rpc.ConfigFileType_TICDC_FILE,
	})
	require.Equal(t, codes.InvalidArgument, status.Code(err))
}

func TestPauseChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	gomock.InOrder(
		cdcSvc.EXPECT().
			GetChangefeed(gomock.Any(), "1001", "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1", Arch: constant.ChangefeedArchReplicationWorker, State: constant.ChangfeedStateRunning}, nil),
		cdcSvc.EXPECT().
			StopChangefeed(gomock.Any(), "1001", "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1", State: constant.ChangefeedStatePaused}, nil),
	)

	info, err := svc.PauseChangefeed(context.Background(), &rpc.PauseChangefeedRequest{
		ClusterId:    "1001",
		ChangefeedId: "cf-1",
		ExpectedArch: string(constant.ChangefeedArchReplicationWorker),
	})
	require.NoError(t, err)
	require.Equal(t, string(constant.ChangefeedStatePaused), info.GetState())
}

func TestResumeChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	gomock.InOrder(
		cdcSvc.EXPECT().
			GetChangefeed(gomock.Any(), "1001", "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1", Arch: constant.ChangefeedArchReplicationWorker, State: constant.ChangefeedStatePaused}, nil),
		cdcSvc.EXPECT().
			StartChangefeed(gomock.Any(), "1001", "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1", State: constant.ChangfeedStateRunning}, nil),
	)

	info, err := svc.ResumeChangefeed(context.Background(), &rpc.ResumeChangefeedRequest{
		ClusterId:    "1001",
		ChangefeedId: "cf-1",
		ExpectedArch: string(constant.ChangefeedArchReplicationWorker),
	})
	require.NoError(t, err)
	require.Equal(t, string(constant.ChangfeedStateRunning), info.GetState())
}

func TestPauseChangefeed_Idempotent(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		GetChangefeed(gomock.Any(), "1001", "cf-1").
		Return(&vo.Changefeed{
			ClusterID: "1001", ChangefeedID: "cf-1",
			Arch: constant.ChangefeedArchReplicationWorker, State: constant.ChangefeedStatePaused,
		}, nil)

	info, err := svc.PauseChangefeed(context.Background(), &rpc.PauseChangefeedRequest{
		ClusterId: "1001", ChangefeedId: "cf-1", ExpectedArch: "replication_worker",
	})
	require.NoError(t, err)
	require.Equal(t, string(constant.ChangefeedStatePaused), info.GetState())
}

func TestDeleteChangefeed_ExpectedArch(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		GetChangefeed(gomock.Any(), "1001", "cf-new").
		Return(&vo.Changefeed{
			ClusterID: "1001", ChangefeedID: "cf-new", Arch: constant.ChangefeedArchTiCDCNewArch,
		}, nil)
	cdcSvc.EXPECT().
		DeleteChangefeed(gomock.Any(), "1001", "cf-new").
		Return(&vo.Changefeed{
			ClusterID: "1001", ChangefeedID: "cf-new",
			Arch: constant.ChangefeedArchTiCDCNewArch, State: constant.ChangefeedStateDeleting,
		}, nil)

	info, err := svc.DeleteChangefeed(context.Background(), &rpc.DeleteChangefeedRequest{
		ClusterId: "1001", ChangefeedId: "cf-new", ExpectedArch: "ticdc_newarch",
	})
	require.NoError(t, err)
	require.Equal(t, string(constant.ChangefeedStateDeleting), info.GetState())
}

func TestChangefeedMutations_MapRecordNotFound(t *testing.T) {
	tests := []struct {
		name string
		call func(rpc.ChangefeedOpsServiceServer) error
	}{
		{
			name: "pause",
			call: func(svc rpc.ChangefeedOpsServiceServer) error {
				_, err := svc.PauseChangefeed(context.Background(), &rpc.PauseChangefeedRequest{
					ClusterId: "1001", ChangefeedId: "missing", ExpectedArch: "replication_worker",
				})
				return err
			},
		},
		{
			name: "resume",
			call: func(svc rpc.ChangefeedOpsServiceServer) error {
				_, err := svc.ResumeChangefeed(context.Background(), &rpc.ResumeChangefeedRequest{
					ClusterId: "1001", ChangefeedId: "missing", ExpectedArch: "replication_worker",
				})
				return err
			},
		},
		{
			name: "delete",
			call: func(svc rpc.ChangefeedOpsServiceServer) error {
				_, err := svc.DeleteChangefeed(context.Background(), &rpc.DeleteChangefeedRequest{
					ClusterId: "1001", ChangefeedId: "missing", ExpectedArch: "replication_worker",
				})
				return err
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			cdcSvc := mock_service.NewMockCDCService(ctrl)
			svc := NewCdcRpcService(cdcSvc)
			cdcSvc.EXPECT().
				GetChangefeed(gomock.Any(), "1001", "missing").
				Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "changefeed not found"))

			err := tt.call(svc)
			require.Equal(t, codes.NotFound, status.Code(err))
		})
	}
}

func TestPrecheckMigrateChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		PrecheckMigrateChangefeed(gomock.Any(), "1001", "cf-old", uint32(3), "target").
		Return(&vo.ChangefeedMigrationPrecheck{
			Passed:     true,
			Checks:     []vo.MigrationCheck{{Name: "source kernel state is normal", Passed: true}},
			SourceArch: "replication_worker", SourceState: "running", CurrentCheckpointTSO: 123,
			TargetRCU: 3, TargetProvider: "aws", TargetRegion: "us-east-1", TargetKubernetes: "k8s-1",
		}, nil)

	resp, err := svc.PrecheckMigrateChangefeed(context.Background(), &rpc.PrecheckMigrateChangefeedRequest{
		ClusterId: "1001", ChangefeedId: "cf-old", Rcu: 3, DisplayName: "target",
	})
	require.NoError(t, err)
	require.True(t, resp.GetPassed())
	require.Equal(t, uint64(123), resp.GetCurrentCheckpointTso())
	require.Len(t, resp.GetChecks(), 1)
}

func TestCompareChangefeeds(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		CompareChangefeeds(gomock.Any(), "1001", "cf-old", "cf-new").
		Return(&vo.ChangefeedComparison{
			Passed:              true,
			Checks:              []vo.MigrationCheck{{Name: "sink URI matches", Passed: true}},
			SourceCheckpointTSO: 123, TargetStartTSO: 123, TargetCheckpointTSO: 456,
			SourceState: "stopped", TargetState: "normal",
		}, nil)

	resp, err := svc.CompareChangefeeds(context.Background(), &rpc.CompareChangefeedsRequest{
		ClusterId: "1001", SourceChangefeedId: "cf-old", TargetChangefeedId: "cf-new",
	})
	require.NoError(t, err)
	require.True(t, resp.GetPassed())
	require.Equal(t, uint64(123), resp.GetTargetStartTso())
}

func TestListClusterChangefeeds_NewArchGetsTiCDCClusterName(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		ListChangefeeds(gomock.Any(), gomock.Any()).
		Return([]*vo.Changefeed{
			{ClusterID: "1001", ChangefeedID: "cf-new", Arch: constant.ChangefeedArchTiCDCNewArch},
			{ClusterID: "1001", ChangefeedID: "cf-old", Arch: constant.ChangefeedArchReplicationWorker},
		}, uint32(2), nil)
	// One batched resolution covering only the newarch changefeed.
	cdcSvc.EXPECT().
		GetChangefeedTiCDCClusterNames(gomock.Any(), "1001", []string{"cf-new"}).
		Return(map[string]string{"cf-new": "ess-ticdc-us-east-1-abcd1234"}, nil)

	resp, err := svc.ListClusterChangefeeds(context.Background(), &rpc.ListClusterChangefeedsRequest{
		ClusterId: "1001",
	})
	require.NoError(t, err)
	require.Len(t, resp.GetItems(), 2)
	require.Equal(t, "ess-ticdc-us-east-1-abcd1234", resp.GetItems()[0].GetTicdcClusterName())
	require.Empty(t, resp.GetItems()[1].GetTicdcClusterName())
}

func TestRestartChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	gomock.InOrder(
		cdcSvc.EXPECT().
			StopChangefeed(gomock.Any(), "1001", "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1", State: constant.ChangefeedStatePaused}, nil),
		cdcSvc.EXPECT().
			StartChangefeed(gomock.Any(), "1001", "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1", State: constant.ChangfeedStateRunning}, nil),
	)

	info, err := svc.RestartChangefeed(context.Background(), &rpc.RestartChangefeedRequest{
		ClusterId:    "1001",
		ChangefeedId: "cf-1",
	})
	require.NoError(t, err)
	require.Equal(t, string(constant.ChangfeedStateRunning), info.GetState())
}

func TestScaleChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		ScaleChangefeed(gomock.Any(), "1001", "cf-1", uint32(8)).
		Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1", RCU: 8}, nil)

	info, err := svc.ScaleChangefeed(context.Background(), &rpc.ScaleChangefeedRequest{
		ClusterId:    "1001",
		ChangefeedId: "cf-1",
		Rcu:          8,
	})
	require.NoError(t, err)
	require.Equal(t, uint32(8), info.GetRcu())
}

func TestGetChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		GetChangefeed(gomock.Any(), "1001", "cf-1").
		Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1"}, nil)
	cdcSvc.EXPECT().
		EnsureChangefeedConfiguration(gomock.Any(), "1001", "cf-1").
		Return(nil)

	info, err := svc.GetChangefeed(context.Background(), &rpc.GetChangefeedRequest{
		ClusterId:    "1001",
		ChangefeedId: "cf-1",
	})
	require.NoError(t, err)
	require.Equal(t, "cf-1", info.GetChangefeedId())
}

func TestGetChangefeedDoesNotFailWhenConfigurationInitializationFails(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		GetChangefeed(gomock.Any(), "1001", "cf-1").
		Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1"}, nil)
	cdcSvc.EXPECT().
		EnsureChangefeedConfiguration(gomock.Any(), "1001", "cf-1").
		Return(bizerr.New(bizerr.ErrCodeUnknown, "config center unavailable"))

	info, err := svc.GetChangefeed(context.Background(), &rpc.GetChangefeedRequest{
		ClusterId:    "1001",
		ChangefeedId: "cf-1",
	})
	require.NoError(t, err)
	require.Equal(t, "cf-1", info.GetChangefeedId())
}

func TestGetChangefeedResolvesClusterFromChangefeedID(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	gomock.InOrder(
		cdcSvc.EXPECT().
			GetChangefeedByID(gomock.Any(), "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1"}, nil),
		cdcSvc.EXPECT().
			GetChangefeed(gomock.Any(), "1001", "cf-1").
			Return(&vo.Changefeed{ClusterID: "1001", ChangefeedID: "cf-1"}, nil),
		cdcSvc.EXPECT().
			EnsureChangefeedConfiguration(gomock.Any(), "1001", "cf-1").
			Return(nil),
	)

	info, err := svc.GetChangefeed(context.Background(), &rpc.GetChangefeedRequest{
		ClusterId:    "cf-1",
		ChangefeedId: "cf-1",
	})
	require.NoError(t, err)
	require.Equal(t, "1001", info.GetClusterId())
	require.Equal(t, "cf-1", info.GetChangefeedId())
}

func TestGetChangefeed_NotFoundUsesGrpcNotFound(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcSvc := mock_service.NewMockCDCService(ctrl)
	svc := NewCdcRpcService(cdcSvc)

	cdcSvc.EXPECT().
		GetChangefeed(gomock.Any(), "1001", "missing").
		Return(nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "record not found"))

	_, err := svc.GetChangefeed(context.Background(), &rpc.GetChangefeedRequest{
		ClusterId: "1001", ChangefeedId: "missing",
	})
	require.Error(t, err)
	require.Equal(t, codes.NotFound, status.Code(err))
}