// 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 coordinator

import (
	"context"
	"math"
	"time"

	"github.com/pingcap/failpoint"
	"github.com/pingcap/log"
	"github.com/pingcap/ticdc/coordinator/changefeed"
	"github.com/pingcap/ticdc/coordinator/gccleaner"
	"github.com/pingcap/ticdc/pkg/common"
	appcontext "github.com/pingcap/ticdc/pkg/common/context"
	"github.com/pingcap/ticdc/pkg/config"
	"github.com/pingcap/ticdc/pkg/config/kerneltype"
	"github.com/pingcap/ticdc/pkg/errors"
	"github.com/pingcap/ticdc/pkg/messaging"
	"github.com/pingcap/ticdc/pkg/metrics"
	"github.com/pingcap/ticdc/pkg/node"
	"github.com/pingcap/ticdc/pkg/pdutil"
	"github.com/pingcap/ticdc/pkg/server"
	"github.com/pingcap/ticdc/pkg/txnutil/gc"
	"github.com/pingcap/ticdc/pkg/util"
	"github.com/pingcap/ticdc/server/watcher"
	"github.com/pingcap/ticdc/utils/chann"
	"github.com/tikv/client-go/v2/oracle"
	pd "github.com/tikv/pd/client"
	"go.uber.org/atomic"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
)

// Message Flow in Coordinator:
// (from maintainer)
// External Messages         Coordinator                Controller              Storage
//      |                        |                          |                     |
//      |   ----message----->    |                          |                     |
//      |                        |                          |                     |
//      |                        |  ---event.message---->   |                     |
//      |                        |                          |                     |
//      |                        |  <---state change-----   |                     |
//      |                        |                          |                     |
//      |                        |  ----update state---------------->             |
//      |                        |                          |                     |
//      |                        |  <---checkpoint ts----   |                     |
//      |                        |                          |                     |
//      |                        |  ----save checkpoint ts------------->          |
//      |                        |                          |                     |
//
// Flow Description:
// 1. External messages arrive at Coordinator via MessageCenter
// 2. Coordinator forwards messages as events to Controller
// 3. Controller processes events and reports state changes back
// 4. Coordinator updates state in meta store
// 5. Controller reports checkpoint TS
// 6. Coordinator saves checkpoint TS to meta store

// coordinator implements the Coordinator interface
type coordinator struct {
	nodeInfo     *node.Info
	version      int64
	gcServiceID  string
	lastTickTime time.Time

	controller *Controller
	backend    changefeed.Backend

	mc             messaging.MessageCenter
	gcManager      gc.Manager
	gcTickInterval time.Duration
	gcCleaner      *gccleaner.Cleaner

	pdClient pd.Client
	pdClock  pdutil.Clock

	// eventCh is used to receive the event from message center, basically these messages
	// are from maintainer.
	eventCh *chann.DrainableChann[*Event]
	// changefeedChangeCh is used to receive the changefeed change from the controller
	changefeedChangeCh chan []*changefeedChange

	// msgGuardWaitGroup guards Add/Wait so Stop never races with new recv handlers.
	msgGuardWaitGroup util.GuardedWaitGroup

	cancel func()
	closed atomic.Bool
}

func New(node *node.Info,
	pdClient pd.Client,
	backend changefeed.Backend,
	gcServiceID string,
	version int64,
	batchSize int,
	balanceCheckInterval time.Duration,
) server.Coordinator {
	mc := appcontext.GetService[messaging.MessageCenter](appcontext.MessageCenter)
	c := &coordinator{
		version:        version,
		nodeInfo:       node,
		gcServiceID:    gcServiceID,
		gcManager:      gc.NewManager(gcServiceID, pdClient),
		gcCleaner:      gccleaner.New(pdClient, gcServiceID),
		gcTickInterval: time.Minute,
		lastTickTime:   time.Now(),

		eventCh:            chann.NewAutoDrainChann[*Event](),
		pdClient:           pdClient,
		pdClock:            appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock),
		mc:                 mc,
		changefeedChangeCh: make(chan []*changefeedChange, 1024),
		backend:            backend,
	}
	// handle messages from message center
	mc.RegisterHandler(messaging.CoordinatorTopic, c.recvMessages)
	c.controller = NewController(
		c.version,
		c.nodeInfo,
		c.changefeedChangeCh,
		c.backend,
		c.eventCh,
		batchSize,
		balanceCheckInterval,
		c.pdClient,
	)

	nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName)
	nodeManager.RegisterOwnerChangeHandler(
		string(c.nodeInfo.ID),
		func(newCoordinatorID string) {
			if newCoordinatorID != string(c.nodeInfo.ID) {
				log.Info("Coordinator changed, and I am not the coordinator, stop myself",
					zap.String("selfID", string(c.nodeInfo.ID)),
					zap.String("newCoordinatorID", newCoordinatorID))
				c.Stop()
			}
		})

	return c
}

func (c *coordinator) recvMessages(ctx context.Context, msg *messaging.TargetMessage) error {
	if !c.msgGuardWaitGroup.AddIf(func() bool {
		return !c.closed.Load()
	}) {
		return nil
	}
	defer c.msgGuardWaitGroup.Done()

	select {
	case <-ctx.Done():
		return context.Cause(ctx)
	default:
		c.eventCh.In() <- &Event{message: msg}
	}

	return nil
}

// Run spawns two goroutines to handle messages and run the coordinator.
func (c *coordinator) Run(ctx context.Context) error {
	ctx, cancel := context.WithCancel(ctx)
	c.cancel = cancel

	eg, ctx := errgroup.WithContext(ctx)
	eg.Go(func() error {
		return c.run(ctx)
	})
	eg.Go(func() error {
		return c.runHandleEvent(ctx)
	})
	eg.Go(func() error {
		return c.gcCleaner.Run(ctx)
	})

	eg.Go(func() error {
		return c.controller.collectMetrics(ctx)
	})

	return eg.Wait()
}

// run handles the following:
// 1. update the gc safepoint to PD
// 2. store the changefeed checkpointTs to meta store
// 3. handle the state changed event
func (c *coordinator) run(ctx context.Context) error {
	failpoint.Inject("InjectUpdateGCTickerInterval", func(val failpoint.Value) {
		c.gcTickInterval = time.Duration(val.(int) * int(time.Second))
	})
	failpoint.Inject("coordinator-run-with-error", func() error {
		return errors.New("coordinator run with error")
	})

	gcTicker := time.NewTicker(c.gcTickInterval)
	defer gcTicker.Stop()
	for {
		select {
		case <-ctx.Done():
			return context.Cause(ctx)
		case <-gcTicker.C:
			// BeginGCTick must happen before updateGCSafepoint so tasks added in this tick
			// will not be undone by the same tick.
			c.gcCleaner.BeginGCTick()
			err := c.updateGCSafepoint(ctx)
			if err == nil {
				c.gcCleaner.OnUpdateGCSafepointSucceeded()
			}
			now := time.Now()
			metrics.CoordinatorCounter.Add(float64(now.Sub(c.lastTickTime)) / float64(time.Second))
			c.lastTickTime = now
		case changes := <-c.changefeedChangeCh:
			if err := c.saveCheckpointTs(ctx, changes); err != nil {
				return errors.Trace(err)
			}
			for _, change := range changes {
				if change.changeType == ChangeState || change.changeType == ChangeStateAndTs {
					if err := c.handleStateChange(ctx, change); err != nil {
						return errors.Trace(err)
					}
				}
			}
		}
	}
}

// runHandleEvent handles messages from the other modules.
func (c *coordinator) runHandleEvent(ctx context.Context) error {
	for {
		select {
		case <-ctx.Done():
			return context.Cause(ctx)
		case event := <-c.eventCh.Out():
			c.controller.HandleEvent(ctx, event)
		}
	}
}

func (c *coordinator) handleStateChange(
	ctx context.Context,
	event *changefeedChange,
) error {
	cf := c.controller.getChangefeed(event.changefeedID)
	if cf == nil {
		log.Warn("changefeed not found", zap.String("changefeed", event.changefeedID.String()))
		return nil
	}
	cfInfo, err := cf.GetInfo().Clone()
	if err != nil {
		return errors.Trace(err)
	}
	cfInfo.State = event.state
	cfInfo.Error = event.err
	progress := config.ProgressNone
	if event.state == config.StateFailed || event.state == config.StateFinished {
		progress = config.ProgressStopping
	}
	if err = c.backend.UpdateChangefeed(ctx, cfInfo, cf.GetStatus().CheckpointTs, progress); err != nil {
		log.Error("failed to update changefeed state",
			zap.Error(err))
		return errors.Trace(err)
	}
	cf.SetInfo(cfInfo)

	switch event.state {
	case config.StateWarning:
		c.controller.operatorController.StopChangefeed(ctx, event.changefeedID, false)
		c.controller.updateChangefeedEpoch(ctx, event.changefeedID)
		c.controller.moveChangefeedToSchedulingQueue(event.changefeedID, false, false)
	case config.StateFailed, config.StateFinished:
		c.controller.operatorController.StopChangefeed(ctx, event.changefeedID, false)
	default:
	}
	return nil
}

// checkStaleCheckpointTs checks if the checkpointTs is stale, if it is, it will send a state change event to the stateChangedCh
func (c *coordinator) checkStaleCheckpointTs(ctx context.Context, changefeed *changefeed.Changefeed, reportedCheckpointTs uint64) {
	id := changefeed.ID
	err := c.gcManager.CheckStaleCheckpointTs(changefeed.GetKeyspaceID(), id, reportedCheckpointTs)
	if err == nil {
		return
	}

	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()

	errCode, _ := errors.RFCCode(err)
	state := config.StateWarning
	if errors.IsChangefeedGCFastFailErrorCode(errCode) {
		state = config.StateFailed
	}

	change := newChangefeedChange(changefeed, state, ChangeState, &config.RunningError{
		Code:    string(errCode),
		Message: err.Error(),
	})

	select {
	case <-ctx.Done():
		log.Warn("Failed to send state change event to stateChangedCh since context timeout, "+
			"there may be a lot of state need to be handled. Try next time",
			zap.String("changefeed", id.String()),
			zap.Error(context.Cause(ctx)))
	case c.changefeedChangeCh <- []*changefeedChange{change}:
	}
}

func (c *coordinator) saveCheckpointTs(ctx context.Context, changes []*changefeedChange) error {
	statusMap := make(map[common.ChangeFeedID]uint64)
	cfsMap := make(map[common.ChangeFeedID]*changefeed.Changefeed)
	for _, change := range changes {
		if change.changeType == ChangeState {
			continue
		}
		upCf := change.changefeed
		reportedCheckpointTs := upCf.GetStatus().CheckpointTs
		if upCf.GetLastSavedCheckPointTs() < reportedCheckpointTs {
			statusMap[upCf.ID] = reportedCheckpointTs
			cfsMap[upCf.ID] = upCf
			c.checkStaleCheckpointTs(ctx, upCf, reportedCheckpointTs)
		}
	}
	if len(statusMap) == 0 {
		return nil
	}
	err := c.controller.backend.UpdateChangefeedCheckpointTs(ctx, statusMap)
	if err != nil {
		log.Error("failed to update checkpointTs", zap.Error(err))
		return errors.Trace(err)
	}
	// update the last saved checkpoint ts and send checkpointTs to maintainer
	for id, cp := range statusMap {
		cf := cfsMap[id]
		cf.SetLastSavedCheckPointTs(cp)
		if cf.NeedCheckpointTsMessage() {
			msg := cf.NewCheckpointTsMessage(cf.GetLastSavedCheckPointTs())
			c.sendMessages([]*messaging.TargetMessage{msg})
		}
	}
	return nil
}

func (c *coordinator) CreateChangefeed(ctx context.Context, info *config.ChangeFeedInfo) error {
	if err := c.controller.CreateChangefeed(ctx, info); err != nil {
		return err
	}

	c.gcCleaner.Add(info.ChangefeedID, info.KeyspaceID, gc.EnsureGCServiceCreating)
	return nil
}

func (c *coordinator) RemoveChangefeed(ctx context.Context, id common.ChangeFeedID) (uint64, error) {
	return c.controller.RemoveChangefeed(ctx, id)
}

func (c *coordinator) PauseChangefeed(ctx context.Context, id common.ChangeFeedID) error {
	return c.controller.PauseChangefeed(ctx, id)
}

func (c *coordinator) ResumeChangefeed(ctx context.Context, id common.ChangeFeedID, newCheckpointTs uint64, overwriteCheckpointTs bool) error {
	if err := c.controller.ResumeChangefeed(ctx, id, newCheckpointTs, overwriteCheckpointTs); err != nil {
		return err
	}
	if overwriteCheckpointTs {
		cf := c.controller.getChangefeed(id)
		if cf != nil {
			c.gcCleaner.Add(id, cf.GetKeyspaceID(), gc.EnsureGCServiceResuming)
		}
	}
	return nil
}

func (c *coordinator) UpdateChangefeed(ctx context.Context, change *config.ChangeFeedInfo) error {
	return c.controller.UpdateChangefeed(ctx, change)
}

func (c *coordinator) ListChangefeeds(ctx context.Context, keyspace string) ([]*config.ChangeFeedInfo, []*config.ChangeFeedStatus, error) {
	return c.controller.ListChangefeeds(ctx, keyspace)
}

func (c *coordinator) GetChangefeed(ctx context.Context, changefeedDisplayName common.ChangeFeedDisplayName) (*config.ChangeFeedInfo, *config.ChangeFeedStatus, error) {
	return c.controller.GetChangefeed(ctx, changefeedDisplayName)
}

func (c *coordinator) Initialized() bool {
	return c.controller.initialized.Load()
}

func (c *coordinator) Stop() {
	if c.closed.CompareAndSwap(false, true) {
		c.mc.DeRegisterHandler(messaging.CoordinatorTopic)
		// Ensure no handler is still writing to eventCh before closing it.
		c.msgGuardWaitGroup.Wait()
		c.controller.Stop()
		c.cancel()
		// close eventCh after cancel, to avoid send or get event from the channel
		c.eventCh.CloseAndDrain()
	}
}

func (c *coordinator) RequestResolvedTsFromLogCoordinator(ctx context.Context, changefeedDisplayName common.ChangeFeedDisplayName) {
	c.controller.RequestResolvedTsFromLogCoordinator(ctx, changefeedDisplayName)
}

func (c *coordinator) sendMessages(msgs []*messaging.TargetMessage) {
	for _, msg := range msgs {
		err := c.mc.SendCommand(msg)
		if err != nil {
			log.Error("failed to send coordinator request", zap.Any("msg", msg), zap.Error(err))
			continue
		}
	}
}

func (c *coordinator) updateGlobalGcSafepoint(ctx context.Context) error {
	minCheckpointTs := c.controller.calculateGlobalGCSafepoint()
	// check if the upstream has a changefeed, if not we should update the gc safepoint
	if minCheckpointTs == math.MaxUint64 {
		ts := c.pdClock.CurrentTime()
		minCheckpointTs = oracle.GoTimeToTS(ts)
	}
	// When the changefeed starts up, CDC will do a snapshot read at
	// (checkpointTs - 1) from TiKV, so (checkpointTs - 1) should be an upper
	// bound for the GC safepoint.
	gcSafepointUpperBound := minCheckpointTs - 1
	err := c.gcManager.TryUpdateServiceGCSafepoint(ctx, gcSafepointUpperBound)
	return errors.Trace(err)
}

func (c *coordinator) updateAllKeyspaceGcBarriers(ctx context.Context) error {
	barrierMap := c.controller.calculateKeyspaceGCBarrier()

	var retErr error
	for meta, barrierTS := range barrierMap {
		if err := c.updateKeyspaceGcBarrier(ctx, meta, barrierTS); err != nil {
			log.Warn("update keyspace gc barrier failed",
				zap.Uint32("keyspaceID", meta.ID), zap.String("keyspaceName", meta.Name),
				zap.Uint64("barrierTS", barrierTS), zap.Error(err))
			retErr = err
		}
	}
	return retErr
}

func (c *coordinator) updateKeyspaceGcBarrier(
	ctx context.Context, meta common.KeyspaceMeta, barrierTS uint64,
) error {
	barrierTsUpperBound := barrierTS - 1
	err := c.gcManager.TryUpdateKeyspaceGCBarrier(ctx, meta.ID, meta.Name, barrierTsUpperBound)
	return errors.Trace(err)
}

// updateGCSafepoint update the gc safepoint
// On next gen, we should update the gc barrier for all keyspaces
// Otherwise we should update the global gc safepoint
func (c *coordinator) updateGCSafepoint(ctx context.Context) error {
	if kerneltype.IsNextGen() {
		return c.updateAllKeyspaceGcBarriers(ctx)
	}
	return c.updateGlobalGcSafepoint(ctx)
}