package changefeed

import (
	"context"
	"errors"
	"fmt"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/app/controllers/open"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/ticdc_spec"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	changefeedopenbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/openapi/v1beta1"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/sdk/auth"
	"google.golang.org/grpc/metadata"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/structpb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type mgmtSvcMock struct {
	grpcli.MgmtSvcGrpcClient
	getAnyDedicatedClusterBasicByID func(context.Context, uint64) (*mgmt.Cluster, bizerr.BizError)
}

func (m *mgmtSvcMock) GetAnyDedicatedClusterBasicByID(ctx context.Context, clusterID uint64) (*mgmt.Cluster, bizerr.BizError) {
	if m.getAnyDedicatedClusterBasicByID == nil {
		panic("unexpected GetAnyDedicatedClusterBasicByID call")
	}
	return m.getAnyDedicatedClusterBasicByID(ctx, clusterID)
}

type changefeedAdapterMock struct {
	resolveChangefeedScope  func(context.Context, string) (changefeedScope, error)
	listChangefeeds         func(context.Context, *changefeedopenbp.ListChangefeedsRequest, changefeedScope) (*changefeedopenbp.ListChangefeedsResponse, error)
	createChangefeed        func(context.Context, *changefeedopenbp.CreateChangefeedRequest, changefeedScope) (*changefeedopenbp.Changefeed, error)
	deleteChangefeed        func(context.Context, *changefeedopenbp.DeleteChangefeedRequest, changefeedScope) (*emptypb.Empty, error)
	editChangefeed          func(context.Context, *changefeedopenbp.EditChangefeedDownstreamConfigRequest, changefeedScope) (*changefeedopenbp.Changefeed, error)
	getChangefeed           func(context.Context, *changefeedopenbp.GetChangefeedRequest, changefeedScope) (*changefeedopenbp.Changefeed, error)
	pauseChangefeed         func(context.Context, *changefeedopenbp.PauseChangefeedRequest, changefeedScope) (*emptypb.Empty, error)
	listChangefeedRCUs      func(context.Context, *changefeedopenbp.ListChangefeedRCUsRequest) (*changefeedopenbp.ListChangefeedRCUsResponse, error)
	scaleChangefeed         func(context.Context, *changefeedopenbp.ScaleChangefeedRequest, changefeedScope) (*changefeedopenbp.Changefeed, error)
	resumeChangefeed        func(context.Context, *changefeedopenbp.ResumeChangefeedRequest, changefeedScope) (*emptypb.Empty, error)
	getEventFilters         func(context.Context, *changefeedopenbp.GetEventFiltersRequest, changefeedScope) (*changefeedopenbp.EventFilters, error)
	setEventFilters         func(context.Context, *changefeedopenbp.SetEventFiltersRequest, changefeedScope) (*changefeedopenbp.SetEventFiltersResponse, error)
	getChangefeedKernelInfo func(context.Context, *changefeedopenbp.GetChangefeedKernelInfoRequest) (*changefeedopenbp.ChangefeedKernelInfo, error)
}

func (m *changefeedAdapterMock) ResolveChangefeedScope(ctx context.Context, changefeedID string) (changefeedScope, error) {
	if m.resolveChangefeedScope == nil {
		panic("unexpected ResolveChangefeedScope call")
	}
	return m.resolveChangefeedScope(ctx, changefeedID)
}

func (m *changefeedAdapterMock) ListChangefeeds(ctx context.Context, request *changefeedopenbp.ListChangefeedsRequest, scope changefeedScope) (*changefeedopenbp.ListChangefeedsResponse, error) {
	if m.listChangefeeds == nil {
		panic("unexpected ListChangefeeds call")
	}
	return m.listChangefeeds(ctx, request, scope)
}

func (m *changefeedAdapterMock) CreateChangefeed(ctx context.Context, request *changefeedopenbp.CreateChangefeedRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error) {
	if m.createChangefeed == nil {
		panic("unexpected CreateChangefeed call")
	}
	return m.createChangefeed(ctx, request, scope)
}

func (m *changefeedAdapterMock) DeleteChangefeed(ctx context.Context, request *changefeedopenbp.DeleteChangefeedRequest, scope changefeedScope) (*emptypb.Empty, error) {
	if m.deleteChangefeed == nil {
		panic("unexpected DeleteChangefeed call")
	}
	return m.deleteChangefeed(ctx, request, scope)
}

func (m *changefeedAdapterMock) EditChangefeedDownstreamConfig(ctx context.Context, request *changefeedopenbp.EditChangefeedDownstreamConfigRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error) {
	if m.editChangefeed == nil {
		panic("unexpected EditChangefeedDownstreamConfig call")
	}
	return m.editChangefeed(ctx, request, scope)
}

func (m *changefeedAdapterMock) GetChangefeed(ctx context.Context, request *changefeedopenbp.GetChangefeedRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error) {
	if m.getChangefeed == nil {
		panic("unexpected GetChangefeed call")
	}
	return m.getChangefeed(ctx, request, scope)
}

func (m *changefeedAdapterMock) PauseChangefeed(ctx context.Context, request *changefeedopenbp.PauseChangefeedRequest, scope changefeedScope) (*emptypb.Empty, error) {
	if m.pauseChangefeed == nil {
		panic("unexpected PauseChangefeed call")
	}
	return m.pauseChangefeed(ctx, request, scope)
}

func (m *changefeedAdapterMock) ListChangefeedRCUs(ctx context.Context, request *changefeedopenbp.ListChangefeedRCUsRequest) (*changefeedopenbp.ListChangefeedRCUsResponse, error) {
	if m.listChangefeedRCUs == nil {
		panic("unexpected ListChangefeedRCUs call")
	}
	return m.listChangefeedRCUs(ctx, request)
}

func (m *changefeedAdapterMock) ScaleChangefeed(ctx context.Context, request *changefeedopenbp.ScaleChangefeedRequest, scope changefeedScope) (*changefeedopenbp.Changefeed, error) {
	if m.scaleChangefeed == nil {
		panic("unexpected ScaleChangefeed call")
	}
	return m.scaleChangefeed(ctx, request, scope)
}

func (m *changefeedAdapterMock) ResumeChangefeed(ctx context.Context, request *changefeedopenbp.ResumeChangefeedRequest, scope changefeedScope) (*emptypb.Empty, error) {
	if m.resumeChangefeed == nil {
		panic("unexpected ResumeChangefeed call")
	}
	return m.resumeChangefeed(ctx, request, scope)
}

func (m *changefeedAdapterMock) GetEventFilters(ctx context.Context, request *changefeedopenbp.GetEventFiltersRequest, scope changefeedScope) (*changefeedopenbp.EventFilters, error) {
	if m.getEventFilters == nil {
		panic("unexpected GetEventFilters call")
	}
	return m.getEventFilters(ctx, request, scope)
}

func (m *changefeedAdapterMock) SetEventFilters(ctx context.Context, request *changefeedopenbp.SetEventFiltersRequest, scope changefeedScope) (*changefeedopenbp.SetEventFiltersResponse, error) {
	if m.setEventFilters == nil {
		panic("unexpected SetEventFilters call")
	}
	return m.setEventFilters(ctx, request, scope)
}

func (m *changefeedAdapterMock) GetChangefeedKernelInfo(ctx context.Context, request *changefeedopenbp.GetChangefeedKernelInfoRequest) (*changefeedopenbp.ChangefeedKernelInfo, error) {
	if m.getChangefeedKernelInfo == nil {
		panic("unexpected GetChangefeedKernelInfo call")
	}
	return m.getChangefeedKernelInfo(ctx, request)
}

func newTestService(adapter ChangefeedAdapter, rbacClient RBACClient, mgmtCli grpcli.MgmtSvcGrpcClient) *Service {
	return &Service{
		adapter:     adapter,
		rbacChecker: nwRBACChecker(rbacClient),
		mgmtCli:     mgmtCli,
	}
}

func newNoopBizChangefeedModel() *biz_changefeed_mock.BizChangefeedModel {
	return &biz_changefeed_mock.BizChangefeedModel{}
}

func permissionContext() context.Context {
	return metadata.NewIncomingContext(context.Background(), metadata.New(map[string]string{
		open.OpenAPIPublicKey: "apk-key",
	}))
}

func testChangefeedScope(tenantID, projectID, clusterID uint64) changefeedScope {
	return testChangefeedScopeWithProvider(tenantID, projectID, clusterID, string(domain.CloudProviderTypeAWS))
}

func testChangefeedScopeWithProvider(tenantID, projectID, clusterID uint64, provider string) changefeedScope {
	return newChangefeedScope(&mgmt.Cluster{
		OrganizationId: tenantID,
		ProjectId:      projectID,
		ClusterId:      clusterID,
		CloudProvider:  provider,
	})
}

func assertPermissionRequest(t *testing.T, params *auth.CheckPermissionParam, expectedURI string, scope changefeedScope) {
	t.Helper()

	require.Equal(t, expectedURI, params.ResourceUri)
	require.Equal(t, "apk-key", params.IdentityVal)
	require.Equal(t, fmt.Sprint(scope.tenantID()), params.ResourceParam["org_id"])
	require.Equal(t, fmt.Sprint(scope.projectID()), params.ResourceParam["project_id"])
	require.Equal(t, fmt.Sprint(scope.clusterID()), params.ResourceParam["cluster_id"])
}

func permissionRBACClient(t *testing.T, expectedURI string, scope changefeedScope, allow bool, authorized *bool) RBACClient {
	t.Helper()

	return RBACClientFunc(func(ctx context.Context, params *auth.CheckPermissionParam) (*auth.CheckPermissionRet, error) {
		assertPermissionRequest(t, params, expectedURI, scope)
		if authorized != nil {
			*authorized = true
		}
		if allow {
			return &auth.CheckPermissionRet{HasPermission: true}, nil
		}
		return &auth.CheckPermissionRet{HasPermission: false, Message: "forbidden"}, nil
	})
}

func TestListChangefeeds(t *testing.T) {
	const (
		clusterID uint64 = 1
		orgID     uint64 = 10
		projectID uint64 = 20
	)

	scope := testChangefeedScope(orgID, projectID, clusterID)
	mockCluster := &mgmt.Cluster{
		ClusterId:      clusterID,
		OrganizationId: orgID,
		ProjectId:      projectID,
		CloudProvider:  string(domain.CloudProviderTypeAWS),
	}

	t.Run("success", func(t *testing.T) {
		mockMgmt := &mgmtSvcMock{
			getAnyDedicatedClusterBasicByID: func(_ context.Context, gotID uint64) (*mgmt.Cluster, bizerr.BizError) {
				require.Equal(t, clusterID, gotID)
				return mockCluster, nil
			},
		}

		authorized := false
		service := newTestService(&changefeedAdapterMock{
			listChangefeeds: func(ctx context.Context, req *changefeedopenbp.ListChangefeedsRequest, gotScope changefeedScope) (*changefeedopenbp.ListChangefeedsResponse, error) {
				require.True(t, authorized)
				require.Equal(t, scope, gotScope)
				require.Equal(t, fmt.Sprint(clusterID), req.GetClusterId())
				return &changefeedopenbp.ListChangefeedsResponse{
					Changefeeds: []*changefeedopenbp.Changefeed{{Id: "101", Name: "cf-1"}},
				}, nil
			},
		}, permissionRBACClient(t, ListChangefeedsReq, scope, true, &authorized), mockMgmt)

		resp, err := service.ListChangefeeds(permissionContext(), &changefeedopenbp.ListChangefeedsRequest{
			ClusterId: fmt.Sprint(clusterID),
		})
		require.NoError(t, err)
		require.Len(t, resp.GetChangefeeds(), 1)
		require.Equal(t, "cf-1", resp.GetChangefeeds()[0].GetName())
	})

	t.Run("adapter error", func(t *testing.T) {
		mockMgmt := &mgmtSvcMock{
			getAnyDedicatedClusterBasicByID: func(_ context.Context, gotID uint64) (*mgmt.Cluster, bizerr.BizError) {
				require.Equal(t, clusterID, gotID)
				return mockCluster, nil
			},
		}

		authorized := false
		service := newTestService(&changefeedAdapterMock{
			listChangefeeds: func(ctx context.Context, req *changefeedopenbp.ListChangefeedsRequest, gotScope changefeedScope) (*changefeedopenbp.ListChangefeedsResponse, error) {
				require.True(t, authorized)
				require.Equal(t, scope, gotScope)
				return nil, errors.New("adapter list failed")
			},
		}, permissionRBACClient(t, ListChangefeedsReq, scope, true, &authorized), mockMgmt)

		resp, err := service.ListChangefeeds(permissionContext(), &changefeedopenbp.ListChangefeedsRequest{
			ClusterId: fmt.Sprint(clusterID),
		})
		require.Error(t, err)
		require.Contains(t, err.Error(), "adapter list failed")
		require.Nil(t, resp)
	})
}

func TestGetChangefeed(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(1, 10, 100)
	ctx := permissionContext()

	t.Run("success", func(t *testing.T) {
		resolved := false
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				resolved = true
				return scope, nil
			},
			getChangefeed: func(ctx context.Context, req *changefeedopenbp.GetChangefeedRequest, gotScope changefeedScope) (*changefeedopenbp.Changefeed, error) {
				require.True(t, resolved)
				require.True(t, authorized)
				require.Equal(t, changefeedID, req.GetChangefeedId())
				require.Equal(t, scope, gotScope)
				return &changefeedopenbp.Changefeed{Id: changefeedID, Name: "cf-1", State: changefeedopenbp.Changefeed_RUNNING}, nil
			},
		}, permissionRBACClient(t, GetChangefeedReq, scope, true, &authorized), nil)

		resp, err := service.GetChangefeed(ctx, &changefeedopenbp.GetChangefeedRequest{ChangefeedId: changefeedID})
		require.NoError(t, err)
		require.Equal(t, changefeedID, resp.GetId())
		require.Equal(t, "cf-1", resp.GetName())
		require.Equal(t, changefeedopenbp.Changefeed_RUNNING, resp.GetState())
	})

	t.Run("rbac denied", func(t *testing.T) {
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				return scope, nil
			},
			getChangefeed: func(ctx context.Context, req *changefeedopenbp.GetChangefeedRequest, gotScope changefeedScope) (*changefeedopenbp.Changefeed, error) {
				t.Fatal("GetChangefeed should not be called after RBAC denial")
				return nil, nil
			},
		}, RBACClientFunc(func(ctx context.Context, params *auth.CheckPermissionParam) (*auth.CheckPermissionRet, error) {
			assertPermissionRequest(t, params, GetChangefeedReq, scope)
			return &auth.CheckPermissionRet{HasPermission: false, Message: "forbidden"}, nil
		}), nil)

		resp, err := service.GetChangefeed(ctx, &changefeedopenbp.GetChangefeedRequest{ChangefeedId: changefeedID})
		require.Error(t, err)
		require.Contains(t, err.Error(), "no permission forbidden")
		require.Nil(t, resp)
	})
}

func TestGetChangefeedKernelInfo(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(1, 10, 100)
	ctx := permissionContext()

	t.Run("success", func(t *testing.T) {
		resolved := false
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				resolved = true
				return scope, nil
			},
			getChangefeedKernelInfo: func(ctx context.Context, req *changefeedopenbp.GetChangefeedKernelInfoRequest) (*changefeedopenbp.ChangefeedKernelInfo, error) {
				require.True(t, resolved)
				require.True(t, authorized)
				require.Equal(t, changefeedID, req.GetChangefeedId())
				return &changefeedopenbp.ChangefeedKernelInfo{
					ChangefeedId: changefeedID,
					TicdcVersion: "v8.5.4",
					ImageTag:     "v8.5.4-release.1",
					KernelType:   changefeedopenbp.ChangefeedKernelInfo_KERNEL_TYPE_NEW,
				}, nil
			},
		}, permissionRBACClient(t, GetChangefeedReq, scope, true, &authorized), nil)

		resp, err := service.GetChangefeedKernelInfo(ctx, &changefeedopenbp.GetChangefeedKernelInfoRequest{ChangefeedId: changefeedID})
		require.NoError(t, err)
		require.Equal(t, changefeedID, resp.GetChangefeedId())
		require.Equal(t, "v8.5.4", resp.GetTicdcVersion())
		require.Equal(t, "v8.5.4-release.1", resp.GetImageTag())
		require.Equal(t, changefeedopenbp.ChangefeedKernelInfo_KERNEL_TYPE_NEW, resp.GetKernelType())
	})

	t.Run("rbac denied", func(t *testing.T) {
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				return scope, nil
			},
			getChangefeedKernelInfo: func(ctx context.Context, req *changefeedopenbp.GetChangefeedKernelInfoRequest) (*changefeedopenbp.ChangefeedKernelInfo, error) {
				t.Fatal("GetChangefeedKernelInfo should not be called after RBAC denial")
				return nil, nil
			},
		}, RBACClientFunc(func(ctx context.Context, params *auth.CheckPermissionParam) (*auth.CheckPermissionRet, error) {
			assertPermissionRequest(t, params, GetChangefeedReq, scope)
			return &auth.CheckPermissionRet{HasPermission: false, Message: "forbidden"}, nil
		}), nil)

		resp, err := service.GetChangefeedKernelInfo(ctx, &changefeedopenbp.GetChangefeedKernelInfoRequest{ChangefeedId: changefeedID})
		require.Error(t, err)
		require.Contains(t, err.Error(), "no permission forbidden")
		require.Nil(t, resp)
	})
}

func TestPauseChangefeed(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(1, 10, 100)
	ctx := permissionContext()

	t.Run("success", func(t *testing.T) {
		resolved := false
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				resolved = true
				return scope, nil
			},
			pauseChangefeed: func(ctx context.Context, req *changefeedopenbp.PauseChangefeedRequest, gotScope changefeedScope) (*emptypb.Empty, error) {
				require.True(t, resolved)
				require.True(t, authorized)
				require.Equal(t, changefeedID, req.GetChangefeedId())
				require.Equal(t, scope, gotScope)
				return &emptypb.Empty{}, nil
			},
		}, permissionRBACClient(t, PauseChangefeedReq, scope, true, &authorized), nil)

		resp, err := service.PauseChangefeed(ctx, &changefeedopenbp.PauseChangefeedRequest{ChangefeedId: changefeedID})
		require.NoError(t, err)
		require.NotNil(t, resp)
	})

	t.Run("adapter error", func(t *testing.T) {
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				return scope, nil
			},
			pauseChangefeed: func(ctx context.Context, req *changefeedopenbp.PauseChangefeedRequest, gotScope changefeedScope) (*emptypb.Empty, error) {
				require.True(t, authorized)
				return nil, errors.New("failed to stop")
			},
		}, permissionRBACClient(t, PauseChangefeedReq, scope, true, &authorized), nil)

		resp, err := service.PauseChangefeed(ctx, &changefeedopenbp.PauseChangefeedRequest{ChangefeedId: changefeedID})
		require.Error(t, err)
		require.Contains(t, err.Error(), "failed to stop")
		require.Nil(t, resp)
	})
}

func TestResumeChangefeed(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(1, 10, 100)
	ctx := permissionContext()

	t.Run("success", func(t *testing.T) {
		resolved := false
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				resolved = true
				return scope, nil
			},
			resumeChangefeed: func(ctx context.Context, req *changefeedopenbp.ResumeChangefeedRequest, gotScope changefeedScope) (*emptypb.Empty, error) {
				require.True(t, resolved)
				require.True(t, authorized)
				require.Equal(t, changefeedID, req.GetChangefeedId())
				require.Equal(t, scope, gotScope)
				return &emptypb.Empty{}, nil
			},
		}, permissionRBACClient(t, ResumeChangefeedReq, scope, true, &authorized), nil)

		resp, err := service.ResumeChangefeed(ctx, &changefeedopenbp.ResumeChangefeedRequest{ChangefeedId: changefeedID})
		require.NoError(t, err)
		require.NotNil(t, resp)
	})

	t.Run("adapter error", func(t *testing.T) {
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				return scope, nil
			},
			resumeChangefeed: func(ctx context.Context, req *changefeedopenbp.ResumeChangefeedRequest, gotScope changefeedScope) (*emptypb.Empty, error) {
				require.True(t, authorized)
				return nil, errors.New("failed to start")
			},
		}, permissionRBACClient(t, ResumeChangefeedReq, scope, true, &authorized), nil)

		resp, err := service.ResumeChangefeed(ctx, &changefeedopenbp.ResumeChangefeedRequest{ChangefeedId: changefeedID})
		require.Error(t, err)
		require.Contains(t, err.Error(), "failed to start")
		require.Nil(t, resp)
	})
}

func TestGetEventFilters(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(1, 10, 100)
	ctx := permissionContext()

	t.Run("success", func(t *testing.T) {
		resolved := false
		authorized := false
		expectedResp := &changefeedopenbp.EventFilters{
			EventFilters: []*changefeedopenbp.EventFilter{{TableMatchers: []string{"test.tbl1"}}},
		}

		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				resolved = true
				return scope, nil
			},
			getEventFilters: func(ctx context.Context, req *changefeedopenbp.GetEventFiltersRequest, gotScope changefeedScope) (*changefeedopenbp.EventFilters, error) {
				require.True(t, resolved)
				require.True(t, authorized)
				require.Equal(t, changefeedID, req.GetChangefeedId())
				require.Equal(t, scope, gotScope)
				return expectedResp, nil
			},
		}, permissionRBACClient(t, GetChangefeedFiltersReq, scope, true, &authorized), nil)

		resp, err := service.GetEventFilters(ctx, &changefeedopenbp.GetEventFiltersRequest{ChangefeedId: changefeedID})
		require.NoError(t, err)
		require.Equal(t, expectedResp, resp)
	})

	t.Run("adapter error", func(t *testing.T) {
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				return scope, nil
			},
			getEventFilters: func(ctx context.Context, req *changefeedopenbp.GetEventFiltersRequest, gotScope changefeedScope) (*changefeedopenbp.EventFilters, error) {
				require.True(t, authorized)
				return nil, errors.New("get failed")
			},
		}, permissionRBACClient(t, GetChangefeedFiltersReq, scope, true, &authorized), nil)

		resp, err := service.GetEventFilters(ctx, &changefeedopenbp.GetEventFiltersRequest{ChangefeedId: changefeedID})
		require.Error(t, err)
		require.Contains(t, err.Error(), "get failed")
		require.Nil(t, resp)
	})
}

func TestSetEventFilters(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(1, 10, 100)
	ctx := permissionContext()
	requestFilters := []*changefeedopenbp.EventFilter{{TableMatchers: []string{"db.new_table"}}}

	t.Run("success", func(t *testing.T) {
		resolved := false
		authorized := false
		expectedResp := &changefeedopenbp.SetEventFiltersResponse{
			EventFilters: []*changefeedopenbp.EventFilter{{TableMatchers: []string{"db.applied"}}},
		}

		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				resolved = true
				return scope, nil
			},
			setEventFilters: func(ctx context.Context, req *changefeedopenbp.SetEventFiltersRequest, gotScope changefeedScope) (*changefeedopenbp.SetEventFiltersResponse, error) {
				require.True(t, resolved)
				require.True(t, authorized)
				require.Equal(t, changefeedID, req.GetChangefeedId())
				require.Equal(t, requestFilters, req.GetEventFilters())
				require.Equal(t, scope, gotScope)
				return expectedResp, nil
			},
		}, permissionRBACClient(t, SetChangefeedFiltersReq, scope, true, &authorized), nil)

		resp, err := service.SetEventFilters(ctx, &changefeedopenbp.SetEventFiltersRequest{
			ChangefeedId: changefeedID,
			EventFilters: requestFilters,
		})
		require.NoError(t, err)
		require.Equal(t, expectedResp, resp)
	})

	t.Run("adapter error", func(t *testing.T) {
		authorized := false
		service := newTestService(&changefeedAdapterMock{
			resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
				require.Equal(t, changefeedID, gotChangefeedID)
				return scope, nil
			},
			setEventFilters: func(ctx context.Context, req *changefeedopenbp.SetEventFiltersRequest, gotScope changefeedScope) (*changefeedopenbp.SetEventFiltersResponse, error) {
				require.True(t, authorized)
				return nil, errors.New("apply failed")
			},
		}, permissionRBACClient(t, SetChangefeedFiltersReq, scope, true, &authorized), nil)

		resp, err := service.SetEventFilters(ctx, &changefeedopenbp.SetEventFiltersRequest{
			ChangefeedId: changefeedID,
			EventFilters: requestFilters,
		})
		require.Error(t, err)
		require.Contains(t, err.Error(), "apply failed")
		require.Nil(t, resp)
	})
}

type connectorServiceStub struct {
	createConnector     func(context.Context, *dataflow.ConnectorCreateReq) (*dataflow.ConnectorID, error)
	deleteConnector     func(context.Context, *dataflow.ConnectorID) (*emptypb.Empty, error)
	editConnector       func(context.Context, *dataflow.ConnectorEditReq) (*emptypb.Empty, error)
	getConnector        func(context.Context, *dataflow.ConnectorID) (*dataflow.Connector, error)
	getChangefeedDetail func(context.Context, *dataflow.ConnectorID) (*structpb.Struct, error)
	listConnectors      func(context.Context, *dataflow.ConnectorListReq) (*dataflow.Connectors, error)
	scaleChangefeed     func(context.Context, *dataflow.ChangefeedScaleReq) (*emptypb.Empty, error)
	startConnector      func(context.Context, *dataflow.ConnectorID) (*emptypb.Empty, error)
	stopConnector       func(context.Context, *dataflow.ConnectorID) (*emptypb.Empty, error)
	testConnector       func(context.Context, *dataflow.ConnectorCreateReq) (*emptypb.Empty, error)
}

type cdcInstanceTypeModelStub struct {
	listChangefeedInstanceTypes func(context.Context) (*domain.ListChangefeedInstanceTypesResp, bizerr.BizError)
}

var _ ticdc_spec.CDCInstanceTypeModel = (*cdcInstanceTypeModelStub)(nil)

func (s *cdcInstanceTypeModelStub) InitAllInstanceTypes(_ context.Context) bizerr.BizError {
	return nil
}

func (s *cdcInstanceTypeModelStub) CalCDCInstanceType(
	_ context.Context, _ *domain.CalCDCInstanceTypeReq,
) (*domain.CalCDCInstanceTypeResp, bizerr.BizError) {
	return nil, nil
}

func (s *cdcInstanceTypeModelStub) ListChangefeedInstanceTypes(
	ctx context.Context,
) (*domain.ListChangefeedInstanceTypesResp, bizerr.BizError) {
	if s.listChangefeedInstanceTypes == nil {
		return &domain.ListChangefeedInstanceTypesResp{
			InstanceTypes: entity.DefaultChangefeedInstanceTypes,
		}, nil
	}
	return s.listChangefeedInstanceTypes(ctx)
}

func (s *cdcInstanceTypeModelStub) GetChangefeedInstanceType(
	_ context.Context, _ *domain.GetChangefeedInstanceTypeReq,
) (*domain.GetChangefeedInstanceTypeResp, bizerr.BizError) {
	return nil, nil
}

func (s *cdcInstanceTypeModelStub) GetCDCInstanceType(
	_ context.Context, _ *domain.GetCDCInstanceTypeReq,
) (*domain.GetCDCInstanceTypeResp, bizerr.BizError) {
	return nil, nil
}

func (s *connectorServiceStub) CreateConnector(ctx context.Context, req *dataflow.ConnectorCreateReq) (*dataflow.ConnectorID, error) {
	if s.createConnector == nil {
		panic("unexpected CreateConnector call")
	}
	return s.createConnector(ctx, req)
}

func (s *connectorServiceStub) DeleteConnector(ctx context.Context, req *dataflow.ConnectorID) (*emptypb.Empty, error) {
	if s.deleteConnector == nil {
		panic("unexpected DeleteConnector call")
	}
	return s.deleteConnector(ctx, req)
}

func (s *connectorServiceStub) EditConnector(ctx context.Context, req *dataflow.ConnectorEditReq) (*emptypb.Empty, error) {
	if s.editConnector == nil {
		panic("unexpected EditConnector call")
	}
	return s.editConnector(ctx, req)
}

func (s *connectorServiceStub) GetConnector(ctx context.Context, req *dataflow.ConnectorID) (*dataflow.Connector, error) {
	if s.getConnector == nil {
		panic("unexpected GetConnector call")
	}
	return s.getConnector(ctx, req)
}

func (s *connectorServiceStub) GetChangefeedDetail(ctx context.Context, req *dataflow.ConnectorID) (*structpb.Struct, error) {
	if s.getChangefeedDetail == nil {
		panic("unexpected GetChangefeedDetail call")
	}
	return s.getChangefeedDetail(ctx, req)
}

func (s *connectorServiceStub) ListConnectors(ctx context.Context, req *dataflow.ConnectorListReq) (*dataflow.Connectors, error) {
	if s.listConnectors == nil {
		panic("unexpected ListConnectors call")
	}
	return s.listConnectors(ctx, req)
}

func (s *connectorServiceStub) ScaleChangefeed(ctx context.Context, req *dataflow.ChangefeedScaleReq) (*emptypb.Empty, error) {
	if s.scaleChangefeed == nil {
		panic("unexpected ScaleChangefeed call")
	}
	return s.scaleChangefeed(ctx, req)
}

func (s *connectorServiceStub) StartConnector(ctx context.Context, req *dataflow.ConnectorID) (*emptypb.Empty, error) {
	if s.startConnector == nil {
		panic("unexpected StartConnector call")
	}
	return s.startConnector(ctx, req)
}

func (s *connectorServiceStub) StopConnector(ctx context.Context, req *dataflow.ConnectorID) (*emptypb.Empty, error) {
	if s.stopConnector == nil {
		panic("unexpected StopConnector call")
	}
	return s.stopConnector(ctx, req)
}

func (s *connectorServiceStub) TestConnector(ctx context.Context, req *dataflow.ConnectorCreateReq) (*emptypb.Empty, error) {
	if s.testConnector == nil {
		panic("unexpected TestConnector call")
	}
	return s.testConnector(ctx, req)
}

func connectorForList(id uint64, name string, scope changefeedScope) *dataflow.Connector {
	return &dataflow.Connector{
		ConnectorId:            id,
		ConnectorType:          dataflow.ConnectorType_MYSQL,
		Status:                 dataflow.Connector_RUNNING,
		ChangefeedInstanceType: "4rcu",
		Name:                   name,
		Request: &dataflow.ConnectorCreateReq{
			ClusterId:     scope.clusterID(),
			ProjectId:     scope.projectID(),
			OrgId:         scope.tenantID(),
			ConnectorType: dataflow.ConnectorType_MYSQL,
			Name:          name,
			MysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{Endpoint: "mysql.example.com:3306", Username: "user", Password: "pass"},
			},
		},
	}
}

func TestListChangefeeds_PaginationTokenRoundTrip(t *testing.T) {
	scope := testChangefeedScope(10, 20, 30)

	t.Run("default page size", func(t *testing.T) {
		stub := &connectorServiceStub{
			listConnectors: func(ctx context.Context, req *dataflow.ConnectorListReq) (*dataflow.Connectors, error) {
				require.Equal(t, scope.clusterID(), req.GetClusterId())
				require.Equal(t, scope.projectID(), req.GetProjectId())
				require.Equal(t, scope.tenantID(), req.GetOrgId())
				require.Equal(t, uint32(1), req.GetPage())
				require.Equal(t, defaultListChangefeedsPageSize, req.GetPerPage())
				return &dataflow.Connectors{
					Total: 1,
					Items: []*dataflow.Connector{connectorForList(101, "cf-1", scope)},
				}, nil
			},
		}

		resp, err := newConnectorChangefeedAdapter(stub, nil, newNoopBizChangefeedModel(), nil, nil, nil).ListChangefeeds(context.Background(), &changefeedopenbp.ListChangefeedsRequest{
			ClusterId: fmt.Sprint(scope.clusterID()),
		}, scope)
		require.NoError(t, err)
		require.Len(t, resp.GetChangefeeds(), 1)
		require.Empty(t, resp.GetNextPageToken())
		require.Equal(t, int32(1), resp.GetTotalSize())
	})

	t.Run("token round trip", func(t *testing.T) {
		var pages []uint32
		stub := &connectorServiceStub{
			listConnectors: func(ctx context.Context, req *dataflow.ConnectorListReq) (*dataflow.Connectors, error) {
				pages = append(pages, req.GetPage())
				require.Equal(t, uint32(1), req.GetPerPage())
				switch req.GetPage() {
				case 1:
					return &dataflow.Connectors{
						Total: 2,
						Items: []*dataflow.Connector{connectorForList(101, "cf-1", scope)},
					}, nil
				case 2:
					return &dataflow.Connectors{
						Total: 2,
						Items: []*dataflow.Connector{connectorForList(102, "cf-2", scope)},
					}, nil
				default:
					t.Fatalf("unexpected page %d", req.GetPage())
					return nil, nil
				}
			},
		}
		adapter := newConnectorChangefeedAdapter(stub, nil, newNoopBizChangefeedModel(), nil, nil, nil)

		first, err := adapter.ListChangefeeds(context.Background(), &changefeedopenbp.ListChangefeedsRequest{
			ClusterId: fmt.Sprint(scope.clusterID()),
			PageSize:  1,
		}, scope)
		require.NoError(t, err)
		require.Len(t, first.GetChangefeeds(), 1)
		require.NotEmpty(t, first.GetNextPageToken())

		page, err := decodePageToken(first.GetNextPageToken())
		require.NoError(t, err)
		require.Equal(t, uint32(2), page)

		second, err := adapter.ListChangefeeds(context.Background(), &changefeedopenbp.ListChangefeedsRequest{
			ClusterId: fmt.Sprint(scope.clusterID()),
			PageSize:  1,
			PageToken: first.GetNextPageToken(),
		}, scope)
		require.NoError(t, err)
		require.Len(t, second.GetChangefeeds(), 1)
		require.Equal(t, "102", second.GetChangefeeds()[0].GetId())
		require.Empty(t, second.GetNextPageToken())
		require.Equal(t, []uint32{1, 2}, pages)
	})
}

func TestAdapterGetChangefeed_InvalidateCache(t *testing.T) {
	const (
		changefeedID       = "101"
		cachedCheckpoint   = uint64(11)
		realtimeCheckpoint = uint64(456)
	)

	newConnector := func(checkpointTSO uint64) *dataflow.Connector {
		var checkpointUTC *timestamppb.Timestamp
		if checkpointTSO > 0 {
			checkpointUTC = timestamppb.New(utils.GetTimeFromTS(checkpointTSO))
		}
		return &dataflow.Connector{
			ConnectorId:            101,
			ConnectorType:          dataflow.ConnectorType_MYSQL,
			Status:                 dataflow.Connector_RUNNING,
			ChangefeedInstanceType: "4rcu",
			Name:                   "cf-1",
			CheckpointTso:          checkpointTSO,
			CheckpointUtc:          checkpointUTC,
			Request: &dataflow.ConnectorCreateReq{
				ClusterId:     30,
				ProjectId:     20,
				OrgId:         10,
				ConnectorType: dataflow.ConnectorType_MYSQL,
				Name:          "cf-1",
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "mysql.example.com:3306",
						Username: "user",
						Password: "pass",
					},
				},
			},
		}
	}

	newDetailResp := func(fields map[string]*structpb.Value) *domain.GetBizChangefeedDetailResp {
		return &domain.GetBizChangefeedDetailResp{
			Changefeed: &structpb.Struct{Fields: fields},
		}
	}

	tests := []struct {
		name             string
		invalidateCache  bool
		detailResp       *domain.GetBizChangefeedDetailResp
		detailErr        bizerr.BizError
		wantErrSubstr    string
		wantCheckpointTS uint64
		wantRealtimeCall bool
	}{
		{
			name:             "uses cached checkpoint when invalidate_cache is false",
			invalidateCache:  false,
			wantCheckpointTS: cachedCheckpoint,
			wantRealtimeCall: false,
		},
		{
			name:            "overrides cached checkpoint when realtime detail has checkpoint_ts",
			invalidateCache: true,
			detailResp: newDetailResp(map[string]*structpb.Value{
				"checkpoint_ts": structpb.NewStringValue("456"),
			}),
			wantCheckpointTS: realtimeCheckpoint,
			wantRealtimeCall: true,
		},
		{
			name:            "keeps cached checkpoint when realtime detail has no checkpoint_ts",
			invalidateCache: true,
			detailResp: newDetailResp(map[string]*structpb.Value{
				"state": structpb.NewStringValue("normal"),
			}),
			wantCheckpointTS: cachedCheckpoint,
			wantRealtimeCall: true,
		},
		{
			name:             "keeps cached checkpoint when realtime detail is not found",
			invalidateCache:  true,
			detailErr:        bizerr.New(bizerr.ErrCodeRecordNotFound, "not found"),
			wantCheckpointTS: cachedCheckpoint,
			wantRealtimeCall: true,
		},
		{
			name:            "returns error when realtime detail uses unsupported key",
			invalidateCache: true,
			detailResp: newDetailResp(map[string]*structpb.Value{
				"checkpointTso": structpb.NewStringValue("456"),
			}),
			wantErrSubstr:    "parse realtime changefeed detail",
			wantRealtimeCall: true,
		},
		{
			name:            "returns error when realtime checkpoint_ts has invalid type",
			invalidateCache: true,
			detailResp: newDetailResp(map[string]*structpb.Value{
				"checkpoint_ts": structpb.NewBoolValue(true),
			}),
			wantErrSubstr:    "parse realtime changefeed detail",
			wantRealtimeCall: true,
		},
	}

	for _, tc := range tests {
		tc := tc
		t.Run(tc.name, func(t *testing.T) {
			realtimeCalled := false
			bizChangefeedModel := newNoopBizChangefeedModel()
			if tc.invalidateCache {
				bizChangefeedModel.On("GetChangefeedDetail", mock.Anything, mock.MatchedBy(func(req *domain.GetBizChangefeedDetailReq) bool {
					return req.BizChangefeedID == 101
				})).Run(func(args mock.Arguments) {
					realtimeCalled = true
				}).Return(tc.detailResp, tc.detailErr).Once()
			}

			adapter := newConnectorChangefeedAdapter(&connectorServiceStub{
				getConnector: func(ctx context.Context, req *dataflow.ConnectorID) (*dataflow.Connector, error) {
					require.Equal(t, uint64(101), req.GetConnectorId())
					return newConnector(cachedCheckpoint), nil
				},
			}, nil, bizChangefeedModel, nil, nil, nil)

			resp, err := adapter.GetChangefeed(context.Background(), &changefeedopenbp.GetChangefeedRequest{
				ChangefeedId:    changefeedID,
				InvalidateCache: tc.invalidateCache,
			}, changefeedScope{})

			if tc.wantErrSubstr != "" {
				require.Error(t, err)
				require.Contains(t, err.Error(), tc.wantErrSubstr)
				require.Nil(t, resp)
			} else {
				require.NoError(t, err)
				require.NotNil(t, resp)
				require.Equal(t, tc.wantCheckpointTS, resp.GetCheckpointTso())
				if tc.wantCheckpointTS > 0 {
					require.NotNil(t, resp.GetCheckpointTs())
					require.Equal(t, utils.GetTimeFromTS(tc.wantCheckpointTS), resp.GetCheckpointTs().AsTime())
				}
			}
			require.Equal(t, tc.wantRealtimeCall, realtimeCalled)
			bizChangefeedModel.AssertExpectations(t)
		})
	}
}

func TestCreateChangefeed_DryRunSkipsGetConnector(t *testing.T) {
	scope := testChangefeedScope(10, 20, 30)
	createCalled := false
	getCalled := false

	adapter := newConnectorChangefeedAdapter(&connectorServiceStub{
		createConnector: func(ctx context.Context, req *dataflow.ConnectorCreateReq) (*dataflow.ConnectorID, error) {
			createCalled = true
			require.Equal(t, scope.clusterID(), req.GetClusterId())
			require.Equal(t, scope.projectID(), req.GetProjectId())
			require.Equal(t, scope.tenantID(), req.GetOrgId())
			require.Equal(t, dataflow.ConnectorType_MYSQL, req.GetConnectorType())
			require.Equal(t, "my-cf", req.GetName())
			require.Equal(t, "4rcu", req.GetChangefeedInstanceType())
			require.NotNil(t, req.GetMysqlInfo())
			require.Equal(t, "db.example.com:3306", req.GetMysqlInfo().GetConnection().GetEndpoint())
			return &dataflow.ConnectorID{ConnectorId: 999}, nil
		},
		getConnector: func(ctx context.Context, req *dataflow.ConnectorID) (*dataflow.Connector, error) {
			getCalled = true
			return nil, errors.New("get should not be called on dry-run")
		},
	}, nil, newNoopBizChangefeedModel(), nil, nil, nil)

	resp, err := adapter.CreateChangefeed(context.Background(), &changefeedopenbp.CreateChangefeedRequest{
		DryRun: true,
		Changefeed: &changefeedopenbp.Changefeed{
			Name:                "my-cf",
			ClusterId:           fmt.Sprint(scope.clusterID()),
			ReplicationCapacity: "4rcu",
			DownstreamType:      changefeedopenbp.DownstreamType_MYSQL,
			DownstreamConfig: &changefeedopenbp.Changefeed_Mysql{Mysql: &changefeedopenbp.MySQLDownstreamConfig{
				Connection: &changefeedopenbp.Connection{
					Endpoint: "db.example.com:3306",
					Username: "user",
					Password: "pass",
				},
			}},
			NetworkInfo: &changefeedopenbp.NetworkInfo{
				NetworkType: changefeedopenbp.NetworkInfo_NETWORK_TYPE_PUBLIC,
			},
		},
	}, scope)
	require.NoError(t, err)
	require.True(t, createCalled)
	require.False(t, getCalled)
	require.NotNil(t, resp)
}

func TestListChangefeedRCUs_MapsInstanceTypes(t *testing.T) {
	resp, err := newConnectorChangefeedAdapter(&connectorServiceStub{}, &cdcInstanceTypeModelStub{}, newNoopBizChangefeedModel(), nil, nil, nil).ListChangefeedRCUs(context.Background(), &changefeedopenbp.ListChangefeedRCUsRequest{})
	require.NoError(t, err)
	require.Len(t, resp.GetRcus(), len(entity.DefaultChangefeedInstanceTypes))
	for i, item := range entity.DefaultChangefeedInstanceTypes {
		got := resp.GetRcus()[i]
		require.Equal(t, item.Name, got.GetName())
		require.Equal(t, fmt.Sprintf("%d RCUs", item.RCUNum), got.GetDisplayName())
		require.Equal(t, uint32(item.RCUNum), got.GetRcuNum())
		require.Equal(t, uint32(item.RPSLimit), got.GetRpsLimit())
	}
}

func TestDeleteChangefeed(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(10, 20, 30)
	authorized := false

	service := newTestService(&changefeedAdapterMock{
		resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
			require.Equal(t, changefeedID, gotChangefeedID)
			return scope, nil
		},
		deleteChangefeed: func(ctx context.Context, req *changefeedopenbp.DeleteChangefeedRequest, gotScope changefeedScope) (*emptypb.Empty, error) {
			require.True(t, authorized)
			require.Equal(t, changefeedID, req.GetChangefeedId())
			require.Equal(t, scope, gotScope)
			return &emptypb.Empty{}, nil
		},
	}, permissionRBACClient(t, DeleteChangefeedReq, scope, true, &authorized), nil)

	resp, err := service.DeleteChangefeed(permissionContext(), &changefeedopenbp.DeleteChangefeedRequest{ChangefeedId: changefeedID})
	require.NoError(t, err)
	require.NotNil(t, resp)
}

func TestScaleChangefeed(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(10, 20, 30)
	authorized := false

	service := newTestService(&changefeedAdapterMock{
		resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
			require.Equal(t, changefeedID, gotChangefeedID)
			return scope, nil
		},
		scaleChangefeed: func(ctx context.Context, req *changefeedopenbp.ScaleChangefeedRequest, gotScope changefeedScope) (*changefeedopenbp.Changefeed, error) {
			require.True(t, authorized)
			require.Equal(t, changefeedID, req.GetChangefeedId())
			require.Equal(t, "8rcu", req.GetReplicationCapacity())
			require.Equal(t, scope, gotScope)
			return &changefeedopenbp.Changefeed{Id: changefeedID, Name: "cf-1", ReplicationCapacity: "8rcu"}, nil
		},
	}, permissionRBACClient(t, ScaleChangefeedReq, scope, true, &authorized), nil)

	resp, err := service.ScaleChangefeed(permissionContext(), &changefeedopenbp.ScaleChangefeedRequest{
		ChangefeedId:        changefeedID,
		ReplicationCapacity: "8rcu",
	})
	require.NoError(t, err)
	require.Equal(t, changefeedID, resp.GetId())
	require.Equal(t, "8rcu", resp.GetReplicationCapacity())
}

func TestEditChangefeedDownstreamConfig(t *testing.T) {
	const changefeedID = "101"
	scope := testChangefeedScope(10, 20, 30)
	authorized := false

	service := newTestService(&changefeedAdapterMock{
		resolveChangefeedScope: func(ctx context.Context, gotChangefeedID string) (changefeedScope, error) {
			require.Equal(t, changefeedID, gotChangefeedID)
			return scope, nil
		},
		editChangefeed: func(ctx context.Context, req *changefeedopenbp.EditChangefeedDownstreamConfigRequest, gotScope changefeedScope) (*changefeedopenbp.Changefeed, error) {
			require.True(t, authorized)
			require.Equal(t, changefeedID, req.GetChangefeedId())
			require.Equal(t, changefeedopenbp.DownstreamType_MYSQL, req.GetDownstreamType())
			require.Equal(t, scope, gotScope)
			return &changefeedopenbp.Changefeed{Id: changefeedID, Name: "cf-1", DownstreamType: changefeedopenbp.DownstreamType_MYSQL}, nil
		},
	}, permissionRBACClient(t, EditChangefeedReq, scope, true, &authorized), nil)

	resp, err := service.EditChangefeedDownstreamConfig(permissionContext(), &changefeedopenbp.EditChangefeedDownstreamConfigRequest{
		ChangefeedId:     changefeedID,
		DownstreamType:   changefeedopenbp.DownstreamType_MYSQL,
		TableConfig:      &changefeedopenbp.TableConfig{FilterRules: []string{"db.tbl1"}},
		DownstreamConfig: &changefeedopenbp.EditChangefeedDownstreamConfigRequest_Mysql{Mysql: &changefeedopenbp.MySQLDownstreamConfig{Connection: &changefeedopenbp.Connection{Endpoint: "mysql.example.com:3306", Username: "user", Password: "pass"}}},
	})
	require.NoError(t, err)
	require.Equal(t, changefeedID, resp.GetId())
	require.Equal(t, changefeedopenbp.DownstreamType_MYSQL, resp.GetDownstreamType())
}

func TestListChangefeedsRequest_PageSizeOptional(t *testing.T) {
	req := &changefeedopenbp.ListChangefeedsRequest{
		ClusterId: "1234567890123",
	}
	require.NoError(t, req.Validate())
}

func TestCreateChangefeedRequest_MySQLPrivateLink_AllowsEmptyEndpoint(t *testing.T) {
	sinkEndpointID := uint64(12345)
	req := &changefeedopenbp.CreateChangefeedRequest{
		Changefeed: &changefeedopenbp.Changefeed{
			ClusterId:           "1234567890123",
			Name:                "cf-mysql-pl",
			ReplicationCapacity: "2rcu",
			DownstreamType:      changefeedopenbp.DownstreamType_MYSQL,
			NetworkInfo: &changefeedopenbp.NetworkInfo{
				NetworkType:    changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
				SinkEndpointId: &sinkEndpointID,
				Ports:          []uint32{3306},
			},
			TableConfig: &changefeedopenbp.TableConfig{
				FilterRules: []string{"*.*"},
				Mode:        changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE,
			},
			StartPosition: &changefeedopenbp.StartPosition{
				Mode: changefeedopenbp.StartPosition_FROM_NOW,
			},
			DownstreamConfig: &changefeedopenbp.Changefeed_Mysql{
				Mysql: &changefeedopenbp.MySQLDownstreamConfig{
					Connection: &changefeedopenbp.Connection{
						Endpoint: "",
						Username: "user",
						Password: "pass",
					},
				},
			},
		},
	}
	require.NoError(t, req.Validate())
}

func TestToConnectorCreateReq_KafkaPublicMissingBrokerEndpoints(t *testing.T) {
	adapter := &connectorChangefeedAdapter{}
	scope := testChangefeedScope(10, 20, 30)

	_, err := adapter.toConnectorCreateReq(&changefeedopenbp.CreateChangefeedRequest{
		Changefeed: &changefeedopenbp.Changefeed{
			ClusterId:           "30",
			Name:                "cf-kafka",
			ReplicationCapacity: "2rcu",
			DownstreamType:      changefeedopenbp.DownstreamType_KAFKA,
			NetworkInfo: &changefeedopenbp.NetworkInfo{
				NetworkType: changefeedopenbp.NetworkInfo_NETWORK_TYPE_PUBLIC,
			},
			TableConfig: &changefeedopenbp.TableConfig{
				FilterRules: []string{"*.*"},
				Mode:        changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE,
			},
			StartPosition: &changefeedopenbp.StartPosition{
				Mode: changefeedopenbp.StartPosition_FROM_NOW,
			},
			DownstreamConfig: &changefeedopenbp.Changefeed_Kafka{
				Kafka: &changefeedopenbp.KafkaDownstreamConfig{
					Broker: &changefeedopenbp.KafkaDownstreamConfig_Broker{},
					DataFormat: &changefeedopenbp.KafkaDownstreamConfig_DataFormat{
						Protocol: changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON,
					},
					TopicPartitionConfig: &changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig{
						DispatchType:      changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC,
						DefaultTopic:      "cdc-events",
						ReplicationFactor: 1,
						PartitionNum:      1,
					},
				},
			},
		},
	}, scope)
	require.Error(t, err)
	require.Contains(t, err.Error(), "kafka.broker.broker_endpoints")
}

func TestToConnectorCreateReq_KafkaPrivateLinkAllowsEmptyBrokerEndpoints(t *testing.T) {
	adapter := &connectorChangefeedAdapter{}
	scope := testChangefeedScope(10, 20, 30)
	sinkEndpointID := uint64(12345)

	_, err := adapter.toConnectorCreateReq(&changefeedopenbp.CreateChangefeedRequest{
		Changefeed: &changefeedopenbp.Changefeed{
			ClusterId:           "30",
			Name:                "cf-kafka-pl",
			ReplicationCapacity: "2rcu",
			DownstreamType:      changefeedopenbp.DownstreamType_KAFKA,
			NetworkInfo: &changefeedopenbp.NetworkInfo{
				NetworkType:    changefeedopenbp.NetworkInfo_NETWORK_TYPE_PRIVATE_LINK,
				SinkEndpointId: &sinkEndpointID,
				Ports:          []uint32{9092},
			},
			TableConfig: &changefeedopenbp.TableConfig{
				FilterRules: []string{"*.*"},
				Mode:        changefeedopenbp.TableConfig_IGNORE_NOT_SUPPORT_TABLE,
			},
			StartPosition: &changefeedopenbp.StartPosition{
				Mode: changefeedopenbp.StartPosition_FROM_NOW,
			},
			DownstreamConfig: &changefeedopenbp.Changefeed_Kafka{
				Kafka: &changefeedopenbp.KafkaDownstreamConfig{
					Broker: &changefeedopenbp.KafkaDownstreamConfig_Broker{},
					DataFormat: &changefeedopenbp.KafkaDownstreamConfig_DataFormat{
						Protocol: changefeedopenbp.KafkaDownstreamConfig_DataFormat_PROTOCOL_CANAL_JSON,
					},
					TopicPartitionConfig: &changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig{
						DispatchType:      changefeedopenbp.KafkaDownstreamConfig_TopicPartitionConfig_DISPATCH_TYPE_ONE_TOPIC,
						DefaultTopic:      "cdc-events",
						ReplicationFactor: 1,
						PartitionNum:      1,
					},
				},
			},
		},
	}, scope)
	require.NoError(t, err)
}

func TestEditChangefeedDownstreamConfigRequestValidate_AllowsEmptyMySQLPassword(t *testing.T) {
	req := &changefeedopenbp.EditChangefeedDownstreamConfigRequest{
		ChangefeedId:   "101",
		DownstreamType: changefeedopenbp.DownstreamType_MYSQL,
		DownstreamConfig: &changefeedopenbp.EditChangefeedDownstreamConfigRequest_Mysql{
			Mysql: &changefeedopenbp.MySQLDownstreamConfig{
				Connection: &changefeedopenbp.Connection{
					Endpoint: "mysql.example.com:3306",
					Username: "user",
					Password: "",
				},
			},
		},
	}
	require.NoError(t, req.Validate())
}

func TestEditChangefeedDownstreamConfig_DryRunSkipsGetConnector(t *testing.T) {
	const changefeedID = "101"
	currentServerlessClusterID := uint64(200)
	editCalled := false
	getCalled := false

	adapter := newConnectorChangefeedAdapter(&connectorServiceStub{
		getConnector: func(ctx context.Context, req *dataflow.ConnectorID) (*dataflow.Connector, error) {
			if getCalled {
				return nil, errors.New("get should not be called twice on dry-run")
			}
			getCalled = true
			require.Equal(t, uint64(101), req.GetConnectorId())
			return &dataflow.Connector{
				ConnectorId:            101,
				ConnectorType:          dataflow.ConnectorType_SERVERLESS_TIDB,
				Status:                 dataflow.Connector_RUNNING,
				ChangefeedInstanceType: "4rcu",
				Name:                   "current-cf",
				Request: &dataflow.ConnectorCreateReq{
					ClusterId:     30,
					ProjectId:     20,
					OrgId:         10,
					ConnectorType: dataflow.ConnectorType_SERVERLESS_TIDB,
					Name:          "current-cf",
					ServerlessTidbInfo: &dataflow.ServerlessTidbInfo{
						ClusterId: currentServerlessClusterID,
						MysqlInfo: &dataflow.MySQLInfo{
							Connection: &dataflow.Connection{
								Endpoint: "old.example.com:3306",
								Username: "user",
								Password: "oldpass",
							},
						},
					},
				},
			}, nil
		},
		editConnector: func(ctx context.Context, req *dataflow.ConnectorEditReq) (*emptypb.Empty, error) {
			editCalled = true
			return &emptypb.Empty{}, nil
		},
	}, nil, newNoopBizChangefeedModel(), nil, nil, nil)

	resp, err := adapter.EditChangefeedDownstreamConfig(context.Background(), &changefeedopenbp.EditChangefeedDownstreamConfigRequest{
		ChangefeedId:   changefeedID,
		DryRun:         true,
		DownstreamType: changefeedopenbp.DownstreamType_MYSQL,
		TableConfig:    &changefeedopenbp.TableConfig{FilterRules: []string{"db.tbl1"}},
		DownstreamConfig: &changefeedopenbp.EditChangefeedDownstreamConfigRequest_Mysql{
			Mysql: &changefeedopenbp.MySQLDownstreamConfig{
				Connection: &changefeedopenbp.Connection{
					Endpoint: "new.example.com:3306",
					Username: "user",
					Password: "pass",
				},
			},
		},
	}, changefeedScope{})
	require.NoError(t, err)
	require.True(t, editCalled)
	require.NotNil(t, resp)
}