// Copyright 2024 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package changefeed

import (
	"context"
	"fmt"
	"testing"

	"github.com/golang/mock/gomock"
	"github.com/pingcap/errors"
	"github.com/pingcap/ticdc/pkg/common"
	"github.com/pingcap/ticdc/pkg/config"
	"github.com/pingcap/ticdc/pkg/etcd"
	"github.com/stretchr/testify/require"
	"go.etcd.io/etcd/api/v3/mvccpb"
	clientv3 "go.etcd.io/etcd/client/v3"
)

func TestGetAllChangefeeds(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()

	// get changefeeds failed
	backend := NewEtcdBackend(cdcClient)
	cdcClient.EXPECT().GetChangefeedInfoAndStatus(gomock.Any()).Return(int64(0), nil, nil, errors.New("get key failed")).Times(1)
	resp, err := backend.GetAllChangefeeds(context.Background())
	require.Nil(t, resp)
	require.NotNil(t, err)

	// info unmarshal failed, changefeed will be ignored
	cdcClient.EXPECT().GetChangefeedInfoAndStatus(gomock.Any()).Return(
		int64(0),
		map[common.ChangeFeedDisplayName]*mvccpb.KeyValue{
			{Name: "test", Keyspace: "default"}: {Key: []byte("/tidb/cdc/default/default/changefeed/info/test"), Value: []byte("{}")},
		},
		map[common.ChangeFeedDisplayName]*mvccpb.KeyValue{
			{Name: "test", Keyspace: "default"}: {Key: []byte("/tidb/cdc/default/default/changefeed/info/test"), Value: []byte("invalid json")},
		},
		nil,
	).Times(1)
	resp, err = backend.GetAllChangefeeds(context.Background())
	require.NotNil(t, resp)
	require.Nil(t, err)
	require.Len(t, resp, 0)

	// status unmarshal failed, changefeed will not be ignored, and the checkpiont ts will be the start ts
	// the old version of changefeed without gid
	cdcClient.EXPECT().GetChangefeedInfoAndStatus(gomock.Any()).Return(
		int64(0),
		map[common.ChangeFeedDisplayName]*mvccpb.KeyValue{
			{Name: "test", Keyspace: "default"}: {Key: []byte("/tidb/cdc/default/default/changefeed/info/test"), Value: []byte("}{")},
		},
		map[common.ChangeFeedDisplayName]*mvccpb.KeyValue{
			{Name: "test", Keyspace: "default"}: {Key: []byte("/tidb/cdc/default/default/changefeed/info/test"), Value: []byte(`{"changefeed-id":"test", "start-ts": 1}`)},
		},
		nil,
	).Times(1)
	// put the gid and status
	etcdClient.EXPECT().Put(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).Times(2)
	resp, err = backend.GetAllChangefeeds(context.Background())
	require.NotNil(t, resp)
	require.Nil(t, err)
	require.Len(t, resp, 1)
	for _, v := range resp {
		require.Equal(t, uint64(1), v.Status.CheckpointTs)
		require.Equal(t, config.ProgressNone, v.Status.Progress)
	}

	// has no info, changefeed will be ignored
	cdcClient.EXPECT().GetChangefeedInfoAndStatus(gomock.Any()).Return(
		int64(0),
		map[common.ChangeFeedDisplayName]*mvccpb.KeyValue{
			{Name: "test", Keyspace: "default"}: {Key: []byte("/tidb/cdc/default/default/changefeed/info/test"), Value: []byte("{}")},
		},
		nil,
		nil,
	).Times(1)
	resp, err = backend.GetAllChangefeeds(context.Background())
	require.NotNil(t, resp)
	require.Nil(t, err)
	require.Len(t, resp, 0)
}

func TestCreateChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	// create changefeeds failed
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(nil, errors.New("txn failed")).Times(1)
	require.NotNil(t, backend.CreateChangefeed(context.Background(), &config.ChangeFeedInfo{}))

	// txn fail
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&clientv3.TxnResponse{Succeeded: false}, nil).Times(1)
	require.NotNil(t, backend.CreateChangefeed(context.Background(), &config.ChangeFeedInfo{}))

	// txn success
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Len(2), gomock.Len(2), gomock.Any()).
		Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(1)
	require.Nil(t, backend.CreateChangefeed(context.Background(), &config.ChangeFeedInfo{}))
}

func TestUpdateChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errors.New("txn failed")).Times(1)
	require.NotNil(t, backend.UpdateChangefeed(context.Background(), &config.ChangeFeedInfo{}, 0, config.ProgressStopping))

	// txn fail
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&clientv3.TxnResponse{Succeeded: false}, nil).Times(1)
	require.NotNil(t, backend.UpdateChangefeed(context.Background(), &config.ChangeFeedInfo{}, 0, config.ProgressStopping))

	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Len(0), NewFuncMatcher(func(i interface{}) bool {
		ops := i.([]clientv3.Op)
		require.Len(t, ops, 2)
		require.True(t, ops[0].IsPut())
		require.True(t, ops[1].IsPut())
		return true
	}), gomock.Any()).Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(1)
	require.Nil(t, backend.UpdateChangefeed(context.Background(), &config.ChangeFeedInfo{}, 2, config.ProgressStopping))
}

func TestPauseChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	changefeedID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	info := &config.ChangeFeedInfo{State: config.StateNormal}
	status := &config.ChangeFeedStatus{Progress: config.ProgressStopping}

	cdcClient.EXPECT().GetChangeFeedInfo(gomock.Any(), changefeedID.DisplayName).Return(info, nil).Times(1)
	cdcClient.EXPECT().GetChangeFeedStatus(gomock.Any(), changefeedID).Return(status, int64(0), nil).Times(1)
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(1)

	err := backend.PauseChangefeed(context.Background(), changefeedID)
	require.Nil(t, err)
}

func TestDeleteChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	changefeedID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)

	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), NewFuncMatcher(func(i interface{}) bool {
		ops := i.([]clientv3.Op)
		require.Len(t, ops, 2)
		require.True(t, ops[0].IsDelete())
		require.True(t, ops[1].IsDelete())
		return true
	}), gomock.Any()).Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(1)

	err := backend.DeleteChangefeed(context.Background(), changefeedID)
	require.Nil(t, err)
}

func TestResumeChangefeed(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	changefeedID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	info := &config.ChangeFeedInfo{State: config.StateStopped}
	status := &config.ChangeFeedStatus{CheckpointTs: 100}

	cdcClient.EXPECT().GetChangeFeedInfo(gomock.Any(), changefeedID.DisplayName).Return(info, nil).Times(1)
	cdcClient.EXPECT().GetChangeFeedStatus(gomock.Any(), changefeedID).Return(status, int64(0), nil).Times(1)
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(1)

	err := backend.ResumeChangefeed(context.Background(), changefeedID, 200)
	require.Nil(t, err)
}

func TestSetChangefeedProgress(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	changefeedID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	status := &config.ChangeFeedStatus{Progress: config.ProgressNone}

	cdcClient.EXPECT().GetChangeFeedStatus(gomock.Any(), changefeedID).Return(status, int64(0), nil).Times(1)
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(1)

	err := backend.SetChangefeedProgress(context.Background(), changefeedID, config.ProgressRemoving)
	require.Nil(t, err)
}

func TestSetChangefeedProgressRetriesOnCASConflict(t *testing.T) {
	// Scenario: SetChangefeedProgress races with another writer updating the same etcd key.
	// Steps:
	// 1) First CAS attempt fails (TxnResponse.Succeeded=false) due to ModRevision mismatch.
	// 2) The function retries (re-reads status + re-attempts Txn) and succeeds.
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	changefeedID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)

	// The first read observes modRevision=1; CAS fails. The second read observes modRevision=2; CAS succeeds.
	cdcClient.EXPECT().GetChangeFeedStatus(gomock.Any(), changefeedID).
		Return(&config.ChangeFeedStatus{Progress: config.ProgressNone}, int64(1), nil).Times(1)
	cdcClient.EXPECT().GetChangeFeedStatus(gomock.Any(), changefeedID).
		Return(&config.ChangeFeedStatus{Progress: config.ProgressNone}, int64(2), nil).Times(1)

	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&clientv3.TxnResponse{Succeeded: false}, nil).Times(1)
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
		Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(1)

	err := backend.SetChangefeedProgress(context.Background(), changefeedID, config.ProgressRemoving)
	require.NoError(t, err)
}

func TestUpdateChangefeedCheckpointTs(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	cdcClient := etcd.NewMockCDCEtcdClient(ctrl)
	etcdClient := etcd.NewMockClient(ctrl)
	cdcClient.EXPECT().GetEtcdClient().Return(etcdClient).AnyTimes()
	cdcClient.EXPECT().GetClusterID().Return("test-cluster-id").AnyTimes()
	backend := NewEtcdBackend(cdcClient)

	cps := map[common.ChangeFeedID]uint64{
		common.NewChangeFeedIDWithName("test1", common.DefaultKeyspaceName): 100,
	}
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&clientv3.TxnResponse{Succeeded: false}, nil).Times(1)
	err := backend.UpdateChangefeedCheckpointTs(context.Background(), cps)
	require.NotNil(t, err)

	cps = make(map[common.ChangeFeedID]uint64)
	for i := 0; i < 129; i++ {
		cps[common.NewChangeFeedIDWithName(fmt.Sprintf("%d", i), common.DefaultKeyspaceName)] = 100
	}
	etcdClient.EXPECT().Txn(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(&clientv3.TxnResponse{Succeeded: true}, nil).Times(2)
	err = backend.UpdateChangefeedCheckpointTs(context.Background(), cps)
	require.Nil(t, err)
}

type FuncMarcher struct {
	m func(interface{}) bool
}

func NewFuncMatcher(m func(interface{}) bool) gomock.Matcher {
	return &FuncMarcher{
		m: m,
	}
}

func (f *FuncMarcher) Matches(x interface{}) bool {
	return f.m(x)
}

func (f *FuncMarcher) String() string {
	return "func"
}