// 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 (
	"math"
	"testing"

	"github.com/pingcap/ticdc/heartbeatpb"
	"github.com/pingcap/ticdc/pkg/common"
	"github.com/pingcap/ticdc/pkg/config"
	"github.com/pingcap/ticdc/pkg/errors"
	"github.com/pingcap/ticdc/pkg/node"
	"github.com/stretchr/testify/require"
	"go.uber.org/atomic"
)

func TestAddAbsentChangefeed(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}
	cf.backoff = NewBackoff(cf.ID, 0, 0)

	db.AddAbsentChangefeed(cf)

	require.Contains(t, db.GetAbsent(), cf)
	require.Contains(t, db.changefeeds, cf.ID)
}

func TestAddStoppedChangefeed(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}

	db.AddStoppedChangefeed(cf)

	require.Contains(t, db.stopped, cf.ID)
	require.Contains(t, db.changefeeds, cf.ID)
}

func TestAddReplicatingMaintainer(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}
	nodeID := node.ID("node-1")

	db.AddReplicatingMaintainer(cf, nodeID)

	require.Contains(t, db.GetReplicating(), cf)
	require.Contains(t, db.changefeeds, cf.ID)
	require.Equal(t, nodeID, cf.GetNodeID())
}

func TestStopByChangefeedID(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}
	db.AddReplicatingMaintainer(cf, node.ID("node-1"))
	require.Contains(t, db.GetByNodeID("node-1"), cf)

	nodeID := db.StopByChangefeedID(cf.ID, false)

	require.Contains(t, db.stopped, cf.ID)
	require.Contains(t, db.changefeeds, cf.ID)
	require.Equal(t, node.ID("node-1"), nodeID)
	require.Lenf(t, db.GetReplicating(), 0, "")
	require.NotContains(t, db.GetByNodeID("node-1"), cf)

	sizeMap := db.GetTaskSizePerNode()
	_, ok := sizeMap["node-1"]
	require.False(t, ok)

	require.Equal(t, "", db.StopByChangefeedID(common.NewChangeFeedIDWithName("a", common.DefaultKeyspaceName), false).String())
}

func TestMoveToSchedulingQueue(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}
	db.AddStoppedChangefeed(cf)
	cf.backoff = NewBackoff(cf.ID, 0, 0)
	cf.status = atomic.NewPointer(&heartbeatpb.MaintainerStatus{
		CheckpointTs: 100,
	})

	db.MoveToSchedulingQueue(cf.ID, true, false)

	require.Contains(t, db.GetAbsent(), cf)
	require.NotContains(t, db.stopped, cf.ID)
}

func TestMoveToSchedulingQueueResetBackoffAlsoResetsCheckpointBaseline(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test-baseline", common.DefaultKeyspaceName)}
	db.AddStoppedChangefeed(cf)
	cf.backoff = NewBackoff(cf.ID, 0, 200)
	cf.status = atomic.NewPointer(&heartbeatpb.MaintainerStatus{
		CheckpointTs: 120,
	})

	db.MoveToSchedulingQueue(cf.ID, true, false)

	require.Equal(t, uint64(120), cf.backoff.checkpointTs)
}

func TestRemoveChangefeed(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}
	db.AddAbsentChangefeed(cf)

	db.StopByChangefeedID(cf.ID, false)
	require.NotContains(t, db.GetAbsent(), cf)
	require.Contains(t, db.changefeeds, cf.ID)
	require.Contains(t, db.stopped, cf.ID)

	cf2 := &Changefeed{ID: common.NewChangeFeedIDWithName("test2", common.DefaultKeyspaceName)}
	db.AddReplicatingMaintainer(cf2, "node1")
	require.Contains(t, db.GetByNodeID("node1"), cf2)
	require.Equal(t, node.ID("node1"), db.StopByChangefeedID(cf2.ID, true))
	require.NotContains(t, db.GetAbsent(), cf2)
	require.NotContains(t, db.changefeeds, cf2.ID)
	require.NotContains(t, db.stopped, cf2.ID)
	require.Equal(t, "", cf2.nodeID.String())

	sizeMap := db.GetTaskSizePerNode()
	_, ok := sizeMap["node1"]
	require.False(t, ok)
}

func TestGetByID(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf := &Changefeed{ID: common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)}
	db.AddStoppedChangefeed(cf)

	result := db.GetByID(cf.ID)
	require.Equal(t, cf, result)
}

func TestChangefeedDBGetAllChangefeeds(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf1 := &Changefeed{ID: common.NewChangeFeedIDWithName("test1", common.DefaultKeyspaceName)}
	cf2 := &Changefeed{ID: common.NewChangeFeedIDWithName("test2", common.DefaultKeyspaceName)}
	db.AddAbsentChangefeed(cf1)
	db.AddAbsentChangefeed(cf2)

	result := db.GetAllChangefeeds()

	require.Contains(t, result, cf1)
	require.Contains(t, result, cf2)
}

func TestGetWaitingSchedulingChangefeeds(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf1 := &Changefeed{ID: common.NewChangeFeedIDWithName("test1", common.DefaultKeyspaceName)}
	cf2 := &Changefeed{ID: common.NewChangeFeedIDWithName("test2", common.DefaultKeyspaceName)}
	cf1.backoff = NewBackoff(cf1.ID, 0, 0)
	cf1.backoff.failed.Store(true)
	cf2.backoff = NewBackoff(cf2.ID, 0, 0)
	db.AddAbsentChangefeed(cf1)
	db.AddReplicatingMaintainer(cf2, "node1")
	cf3 := &Changefeed{ID: common.NewChangeFeedIDWithName("test3", common.DefaultKeyspaceName)}
	cf3.backoff = NewBackoff(cf3.ID, 0, 0)
	db.AddAbsentChangefeed(cf3)

	result, nMap := db.GetWaitingSchedulingChangefeeds(3)
	require.Equal(t, 1, nMap["node1"])
	require.NotContains(t, result, cf1)
	require.NotContains(t, result, cf2)
	require.Contains(t, result, cf3)

	result, nMap = db.GetWaitingSchedulingChangefeeds(1)
	require.Equal(t, 1, nMap["node1"])
	require.Len(t, result, 1)
}

func TestGetAllStoppedChangefeeds(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf1 := &Changefeed{ID: common.NewChangeFeedIDWithName("test1", common.DefaultKeyspaceName)}
	cf2 := &Changefeed{ID: common.NewChangeFeedIDWithName("test2", common.DefaultKeyspaceName)}
	db.AddStoppedChangefeed(cf1)
	db.AddStoppedChangefeed(cf2)

	require.Equal(t, 2, db.GetStoppedSize())
}

func TestGetAllReplicatingMaintainers(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf1 := &Changefeed{ID: common.NewChangeFeedIDWithName("test1", common.DefaultKeyspaceName)}
	cf2 := &Changefeed{ID: common.NewChangeFeedIDWithName("test2", common.DefaultKeyspaceName)}
	nodeID1 := node.ID("node-1")
	nodeID2 := node.ID("node-2")
	db.AddReplicatingMaintainer(cf1, nodeID1)
	db.AddReplicatingMaintainer(cf2, nodeID2)

	result := db.GetReplicating()
	require.Contains(t, result, cf1)
	require.Contains(t, result, cf2)
}

func TestGetSize(t *testing.T) {
	db := NewChangefeedDB(1216)
	cf1 := &Changefeed{ID: common.NewChangeFeedIDWithName("test1", common.DefaultKeyspaceName)}
	cf2 := &Changefeed{ID: common.NewChangeFeedIDWithName("test2", common.DefaultKeyspaceName)}
	db.AddReplicatingMaintainer(cf1, "node-1")
	db.AddAbsentChangefeed(cf2)
	db.AddStoppedChangefeed(&Changefeed{ID: common.NewChangeFeedIDWithName("test2", common.DefaultKeyspaceName)})
	require.Equal(t, 1, db.GetReplicatingSize())
	require.Equal(t, 1, db.GetStoppedSize())
	require.Equal(t, 1, db.GetAbsentSize())
	require.Equal(t, 3, db.GetSize())
	db.BindChangefeedToNode("", "node-1", cf2)
	require.Equal(t, 0, db.GetAbsentSize())
	require.Equal(t, 1, len(db.GetScheduling()))
	require.Contains(t, db.GetByNodeID("node-1"), cf1)
	require.Contains(t, db.GetByNodeID("node-1"), cf2)

	db.BindChangefeedToNode("node-1", "node-2", cf2)
	require.NotContains(t, db.GetByNodeID("node-1"), cf2)
	require.Contains(t, db.GetByNodeID("node-2"), cf2)

	db.BindChangefeedToNode("node-2", "node-3", cf2)
	require.NotContains(t, db.GetByNodeID("node-2"), cf2)
	require.Contains(t, db.GetByNodeID("node-3"), cf2)

	sizeMap := db.GetTaskSizePerNode()
	require.Equal(t, 1, sizeMap["node-1"])
	require.Equal(t, 1, sizeMap["node-3"])
	_, ok := sizeMap["node-2"]
	require.False(t, ok)
}

func TestReplaceStoppedChangefeed(t *testing.T) {
	db := NewChangefeedDB(1216)
	cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	cf := &Changefeed{
		ID: cfID,
		info: atomic.NewPointer(&config.ChangeFeedInfo{
			ChangefeedID: cfID,
			Config:       config.GetDefaultReplicaConfig(),
			SinkURI:      "mysql://127.0.0.1:3306",
		}),
		status: atomic.NewPointer(&heartbeatpb.MaintainerStatus{}),
	}
	db.AddStoppedChangefeed(cf)
	require.Contains(t, db.stopped, cf.ID)

	cf2 := &config.ChangeFeedInfo{
		ChangefeedID: cfID,
		Config:       config.GetDefaultReplicaConfig(),
		SinkURI:      "kafka://127.0.0.1:9092",
	}
	db.ReplaceStoppedChangefeed(cf2)
	require.Contains(t, db.stopped, cf.ID)

	cf3 := db.GetByID(cf.ID)
	require.Equal(t, true, cf3.NeedCheckpointTsMessage())

	cf4ID := common.NewChangeFeedIDWithName("test4", common.DefaultKeyspaceName)
	cf4 := &config.ChangeFeedInfo{
		ChangefeedID: cf4ID,
		SinkURI:      "kafka://127.0.0.1:9092",
	}
	db.ReplaceStoppedChangefeed(cf4)

	require.NotContains(t, db.changefeeds, cf4ID)
}

func TestScheduleChangefeed(t *testing.T) {
	db := NewChangefeedDB(1216)
	cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	cf := NewChangefeed(cfID, &config.ChangeFeedInfo{
		ChangefeedID: cfID,
		Config:       config.GetDefaultReplicaConfig(),
		SinkURI:      "mysql://127.0.0.1:3306",
	},
		10, true)
	db.AddAbsentChangefeed(cf)
	db.MarkMaintainerScheduling(cf)
	db.BindChangefeedToNode("", "node-1", cf)
	cf.backoff.isRestarting.Store(true)
	db.MarkMaintainerReplicating(cf)
	require.Contains(t, db.GetReplicating(), cf)
	require.False(t, cf.backoff.isRestarting.Load())

	cf2 := db.GetByChangefeedDisplayName(cf.ID.DisplayName)
	require.Equal(t, cf, cf2)

	db.MarkMaintainerAbsent(cf2)
	require.Contains(t, db.GetAbsent(), cf)
	require.NotContains(t, db.GetReplicating(), cf)
	require.NotContains(t, db.GetScheduling(), cf)
	require.Equal(t, "", cf.nodeID.String())
	require.NotContains(t, db.GetByNodeID("node-1"), cf)
}

func TestCalculateGCSafepoint(t *testing.T) {
	db := NewChangefeedDB(1216)
	require.True(t, math.MaxUint64 == db.CalculateGlobalGCSafepoint())

	cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	cf1 := NewChangefeed(cfID,
		&config.ChangeFeedInfo{
			ChangefeedID: cfID,
			Config:       config.GetDefaultReplicaConfig(),
			State:        config.StateStopped,
		}, 11, true)
	db.AddStoppedChangefeed(cf1)
	require.Equal(t, uint64(11), db.CalculateGlobalGCSafepoint())

	cf2ID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	cf2 := NewChangefeed(cf2ID,
		&config.ChangeFeedInfo{
			ChangefeedID: cf2ID,
			Config:       config.GetDefaultReplicaConfig(),
			State:        config.StateFinished,
		}, 9, true)
	db.AddStoppedChangefeed(cf2)
	require.Equal(t, uint64(11), db.CalculateGlobalGCSafepoint())

	cf3ID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	cf3 := NewChangefeed(cf3ID,
		&config.ChangeFeedInfo{
			ChangefeedID: cf3ID,
			Config:       config.GetDefaultReplicaConfig(),
			State:        config.StateNormal,
		}, 10, true)
	db.AddStoppedChangefeed(cf3)
	require.Equal(t, uint64(10), db.CalculateGlobalGCSafepoint())

	cf4ID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	cf4 := NewChangefeed(cf4ID,
		&config.ChangeFeedInfo{
			ChangefeedID: cf4ID,
			Config:       config.GetDefaultReplicaConfig(),
			State:        config.StateFailed,
			Error: &config.RunningError{
				Code: string(errors.ErrGCTTLExceeded.ID()),
			},
		}, 7, true)
	db.AddStoppedChangefeed(cf4)
	require.Equal(t, uint64(10), db.CalculateGlobalGCSafepoint())

	cf5ID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName)
	cf5 := NewChangefeed(cf5ID,
		&config.ChangeFeedInfo{
			ChangefeedID: cf5ID,
			Config:       config.GetDefaultReplicaConfig(),
			State:        config.StateFailed,
			Error:        &config.RunningError{},
		}, 7, true)
	db.AddStoppedChangefeed(cf5)
	require.Equal(t, uint64(7), db.CalculateGlobalGCSafepoint())
}