package dm

import (
	"context"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/pingcap/errors"
	"github.com/pingcap/tiflow/dm/openapi"
	"github.com/tidbcloud/dataflow-service/internal/common"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/handler/agent"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/eks-provider/api/v1alpha1"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

func (m Model) NewDMMetering(ctx context.Context) error {
	startTs := time.Now().Truncate(time.Hour).Add(-1 * time.Hour)
	endTs := startTs.Add(time.Hour)
	logger := m.Log().With(zap.String("meter new dm jobs at", startTs.String()), zap.String("end at", endTs.String()))

	// get all running dm jobs
	// TODO need split with original meter worker
	jobs, err := GetJobsForMetering(entity.DMWorkerVersionV2, &startTs, &endTs)
	if err != nil {
		logger.Error("fail to get new metering jobs", zap.Error(err))
		return err
	}

	success := 0
	for _, job := range jobs {
		m.meterSingleJob(ctx, &job, logger, startTs, endTs)
		success++
	}
	logger.Info("finish metering", zap.String("success job", fmt.Sprintf("%d/%d", success, len(jobs))))
	return nil
}

func (m Model) meterSingleJob(ctx context.Context, dfJob *entity.DFJob, logger *zap.Logger, startTs, endTs time.Time) {

	resourceAPI, err := NewRegionalResourceAPI(ctx, dfJob.Provider, dfJob.Region, dfJob.TenantID.Uint64())
	if err != nil {
		logger.Error("failed to get resource api", zap.Error(err))
		return
	}
	m.resourceAPI = resourceAPI

	var (
		meteringKind2Fn = map[entity.MeteringKind]func(context.Context, *zap.Logger, *entity.DMJob, entity.DMMetering) (*entity.DMMetering, error){
			entity.MeteringKindNode:   m.meterDMNodeUsage,
			entity.MeteringKindInput:  m.meterDMInputUsage,
			entity.MeteringKindOutput: m.meterDMOutputUsage,
			entity.MeteringKindNat:    m.meterDMNatUsage,
		}
		kind entity.MeteringKind
		fn   func(context.Context, *zap.Logger, *entity.DMJob, entity.DMMetering) (*entity.DMMetering, error)
	)
	defer func() {
		if err := recover(); err != nil {
			logger.Error("metering panic occurred", zap.Uint64("job-id", uint64(dfJob.ID)),
				zap.String("kind", string(kind)), zap.Any("err", err))
			return
		}
	}()

	meterings := []*entity.DMMetering{}
	for kind, fn = range meteringKind2Fn {
		if dfJob.Provider == infraapi.ProviderTypeGKE && kind == entity.MeteringKindInput {
			// ingress traffic is free on GCP, see https://cloud.google.com/vpc/network-pricing#vpc-pricing
			// we only meter NAT for ingress traffic if conn-method is DIRECT
			continue
		}
		logger := logger.With(zap.Uint64("job-id", uint64(dfJob.ID)), zap.String("kind", string(kind)))
		if _, err := getCurMeteringRecord(dfJob.ID, startTs, endTs, kind); err == nil {
			logger.Debug("skip metering job")
			continue
		} else if err != gorm.ErrRecordNotFound {
			logger.Error("fail to get prev metering record", zap.Error(err))
			continue
		}
		// get dm job
		dmJob, err := GetDMJobByID(dfJob.ID)
		if err != nil {
			logger.Error("fail to get dm job", zap.Error(err))
			continue
		}
		dmMeteringTemplate := entity.DMMetering{
			JobID:     dfJob.ID,
			StartTS:   startTs,
			EndTS:     endTs,
			Synced:    false,
			ClusterID: dfJob.ClusterID,
			ProjectID: dfJob.ProjectID,
			TenantID:  dfJob.TenantID,
		}
		metering, err := fn(ctx, logger, dmJob, dmMeteringTemplate)
		if err != nil || metering == nil {
			// if the metering == nil
			// 	this metering should be skipped
			// if err != nil
			// 	error has logged already, no need to log again
			continue
		}
		metering.CostID = uuid.New().String()
		meterings = append(meterings, metering)
	}
	if err := recordNewMeterings(meterings); err != nil {
		logger.Error("fail to record new meterings", zap.Error(err))
	}
}

// CostDMSyncJob is exported for testing.
func (m Model) CostDMSyncJob(ctx context.Context,
	logger *zap.Logger,
	dmJob *entity.DMJob,
	jobSource entity.DMJobSource,
	dmMeteringKind entity.MeteringKind) (cost float64, err error) {

	if !needCheckDMSyncJob(dmJob, jobSource) {
		return
	}

	jobSourceName := generateDMDataSourceName(&jobSource)
	taskStatus, err := m.getDMTaskStatus(ctx, dmJob, logger, entity.DMStageSync, jobSourceName)
	if err != nil {
		err = errors.Annotatef(err, "failed to get dm sync task status:%s", jobSourceName)
		return 0, err
	}
	if taskStatus == nil || len(taskStatus.Data) == 0 {
		logger.Warn("get dm job sync task status is empty")
		return
	}

	for _, subTaskStatus := range taskStatus.Data {
		if subTaskStatus.SourceName != jobSourceName ||
			subTaskStatus.Unit != DMTaskUintSync {
			continue
		}
		if dmMeteringKind == entity.MeteringKindOutput {
			cost += float64(subTaskStatus.SyncStatus.IoTotalBytes)
		} else {
			cost += float64(subTaskStatus.SyncStatus.DumpIoTotalBytes)
		}
	}
	return
}

func needCheckDMSyncJob(job *entity.DMJob, jobSource entity.DMJobSource) bool {
	switch job.Mode {
	case entity.DMJobModeIncr:
		return true
	case entity.DMJobModeAll:
		return jobSource.LoadStatus.IsFinished()
	default:
		return false
	}
}

//nolint:unused
func (m Model) meterDMNatUsage(ctx context.Context, logger *zap.Logger, dmJob *entity.DMJob, dmMetering entity.DMMetering) (*entity.DMMetering, error) {
	// only for public IP
	var (
		sources []entity.DMJobSource
		err     error
	)
	if sources, err = GetDMJobSources(dmJob.ID); err != nil {
		logger.Error("fail to get dm job source", zap.Error(err))
		return nil, err
	}
	if len(sources) < 1 {
		// no source
		logger.Error("no source for dm job")
		return nil, errors.New("no source for dm job")
	}

	dmMetering.Kind = entity.MeteringKindNat
	for _, source := range sources {
		if source.ConnProfile.ConnMethod != entity.ConnMethodDirect {
			return nil, nil
		}
		if source.DumpStatus.IsFinished() {
			dmMetering.Cost += source.DumpStatus.FinishedBytes
		}
		cost, err := m.CostDMSyncJob(ctx, logger, dmJob, source, dmMetering.Kind)
		if err != nil {
			return nil, err
		}
		dmMetering.Cost = util.ByteToGBFloat64(dmMetering.Cost + cost)
	}

	return &dmMetering, nil
}

//nolint:unused
func (m Model) meterDMInputUsage(ctx context.Context, logger *zap.Logger, dmJob *entity.DMJob, dmMetering entity.DMMetering) (*entity.DMMetering, error) {

	var (
		sources []entity.DMJobSource
		err     error
	)
	if sources, err = GetDMJobSources(dmJob.ID); err != nil {
		logger.Error("fail to get dm job source", zap.Error(err))
		return nil, err
	}
	if len(sources) < 1 {
		// no source
		logger.Error("no source for dm job")
		return nil, errors.New("no source for dm job")
	}
	dmMetering.Kind = entity.MeteringKindInput

	for _, source := range sources {
		if source.DumpStatus.IsFinished() {
			dmMetering.Cost += source.DumpStatus.FinishedBytes
		}
		cost, err := m.CostDMSyncJob(ctx, logger, dmJob, source, dmMetering.Kind)
		if err != nil {
			return nil, err
		}
		dmMetering.Cost = util.ByteToGBFloat64(dmMetering.Cost + cost)
	}
	return &dmMetering, nil
}

//nolint:unused
func (m Model) meterDMOutputUsage(ctx context.Context, logger *zap.Logger, dmJob *entity.DMJob, dmMetering entity.DMMetering) (*entity.DMMetering, error) {

	var (
		sources []entity.DMJobSource
		err     error
	)
	if sources, err = GetDMJobSources(dmJob.ID); err != nil {
		logger.Error("fail to get dm job source", zap.Error(err))
		return nil, err
	}
	if len(sources) < 1 {
		// no source
		logger.Error("no source for dm job")
		return nil, errors.New("no source for dm job")
	}
	dmMetering.Kind = entity.MeteringKindOutput

	for _, source := range sources {
		if source.LoadStatus.IsFinished() {
			dmMetering.Cost += float64(source.LoadStatus.FinishedBytes)
		}
		cost, err := m.CostDMSyncJob(ctx, logger, dmJob, source, dmMetering.Kind)
		if err != nil {
			return nil, err
		}
		dmMetering.Cost = util.ByteToGBFloat64(dmMetering.Cost + cost)
	}

	return &dmMetering, nil
}

func (m Model) meterDMNodeUsage(ctx context.Context, logger *zap.Logger, dmJob *entity.DMJob, dmMetering entity.DMMetering) (*entity.DMMetering, error) {
	// get tcu number
	tcu, err := getRCUBySpecID(dmJob.SpecID)
	if err != nil {
		logger.Error("fail to get tcu number", zap.Error(err))
		return nil, err
	}
	dmMetering.Kind = entity.MeteringKindNode
	dmMetering.Cost = tcu
	return &dmMetering, nil
}

func (m Model) getDMTaskStatus(ctx context.Context, dmJob *entity.DMJob, logger *zap.Logger, stage entity.DMStageType, jobSourceName string) (*openapi.GetTaskStatusResponse, error) {

	_, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, uint64(dmJob.ClusterID))
	if err != nil {
		logger.Error("fail to get cluster network agent", zap.Error(err))
		return nil, err
	}

	dmJobCR, err := m.resourceAPI.GetDMJob(ctx, dmJob.ProjectID.Uint64(), dmJob.ID.Uint64())
	if err != nil {
		logger.Error("failed to get dm job cr", zap.Error(err))
		return nil, err
	}

	if dmJobCR.Status.Phase != v1alpha1.DMJobPhaseRunning {
		// should not happen
		logger.Error("dm job CR not in running phase", zap.Reflect("phase", dmJobCR.Status.Phase))
		return nil, errors.New("dm job CR in invalid phase")
	}

	taskName := generateDMTaskName(dmJob.ID, stage)
	dmOpenapiClient := agentClient.GetDMOpenapiClient(dmJob.ClusterID.Uint64(), dmJobCR.Status.MasterServerName)
	taskStatus, err := dmOpenapiClient.QueryTaskStatus(ctx, taskName, []string{jobSourceName})
	if err == agent.ErrorDMResourceNotFound {
		return nil, nil
	} else if err != nil {
		logger.Error("fail to get engine job", zap.Error(err))
		return nil, err
	}
	return taskStatus, nil
}