package action

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"time"

	meteringsdkcommon "github.com/pingcap/metering_sdk/common"
	meteringsdkwriter "github.com/pingcap/metering_sdk/writer"
	meteringsdkmetawriter "github.com/pingcap/metering_sdk/writer/meta"
	apiv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/clustergateway"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
)

const (
	SendMeteringMetadataActionName         = "send-metering-metadata"
	SendPausingMeteringMetadataActionName  = "send-pausing-metering-metadata"
	SendPausedMeteringMetadataActionName   = "send-paused-metering-metadata"
	SendResumingMeteringMetadataActionName = "send-resuming-metering-metadata"
	SendActiveMeteringMetadataActionName   = "send-active-metering-metadata"

	SharedPoolMeteringStateActive   = "active"
	SharedPoolMeteringStatePausing  = "pausing"
	SharedPoolMeteringStatePaused   = "paused"
	SharedPoolMeteringStateResuming = "resuming"
)

type SendMeteringMetadataAction struct {
	enable             bool
	env                string
	region             string
	plan               string
	clusterGatewayCli  clustergateway.Client
	sharedPoolRepo     repo.SharedPoolRepo
	tidbClusterRepo    repo.TiDBClusterRepo
	meteringMetaWriter *meteringsdkmetawriter.MetaWriter
}

func NewSendMeteringMetadataAction(
	tidbClusterRepo repo.TiDBClusterRepo,
	sharedPoolRepo repo.SharedPoolRepo,
	clusterGatewayCli clustergateway.Client,
	meteringMetaWriter *meteringsdkmetawriter.MetaWriter,
	env string, // dev, staging, production
	region string, // phyical region where the executor is deployed
	plan string,
	enable bool,
) *SendMeteringMetadataAction {
	return &SendMeteringMetadataAction{
		sharedPoolRepo:     sharedPoolRepo,
		tidbClusterRepo:    tidbClusterRepo,
		clusterGatewayCli:  clusterGatewayCli,
		meteringMetaWriter: meteringMetaWriter,
		env:                env,
		region:             region,
		plan:               plan,
		enable:             enable,
	}
}

type SendSharedPoolLifecycleMetadataAction struct {
	name               string
	pauseState         string
	terminal           bool
	operationRepo      repo.PauseResumeOperationRepo
	meteringMetaAction *SendMeteringMetadataAction
}

func NewSendSharedPoolLifecycleMetadataAction(
	name string,
	pauseState string,
	terminal bool,
	operationRepo repo.PauseResumeOperationRepo,
	meteringMetaAction *SendMeteringMetadataAction,
) actions.Action {
	return &SendSharedPoolLifecycleMetadataAction{
		name:               name,
		pauseState:         pauseState,
		terminal:           terminal,
		operationRepo:      operationRepo,
		meteringMetaAction: meteringMetaAction,
	}
}

func (a *SendSharedPoolLifecycleMetadataAction) Name() string {
	return a.name
}

func (a *SendSharedPoolLifecycleMetadataAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	if !a.meteringMetaAction.enable {
		tilog.Info(ctx, "metering is not enabled, skip this action")
		return nil
	}

	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return fmt.Errorf("failed to get shared pool id: %w", err)
	}
	operationID, err := args.GetString(workflow.PauseResumeOperationIdKey)
	if err != nil {
		return fmt.Errorf("failed to get pause/resume operation id: %w", err)
	}
	operation, err := a.operationRepo.LoadByID(ctx, sharedPoolID, operationID)
	if err != nil {
		return fmt.Errorf("failed to load pause/resume operation: %w", err)
	}
	modifyTS, err := pauseResumeMeteringModifyTS(operation, a.terminal)
	if err != nil {
		return err
	}
	sharedPool, err := a.meteringMetaAction.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return fmt.Errorf("failed to load shared pool: %w", err)
	}
	err = a.meteringMetaAction.sendSharedPoolMeta(ctx, sharedPool, false, modifyTS, a.pauseState, operationID)
	if errors.Is(err, meteringsdkwriter.ErrFileExists) {
		modifyTS = max(modifyTS+1, time.Now().Unix())
		err = a.meteringMetaAction.sendSharedPoolMeta(ctx, sharedPool, false, modifyTS, a.pauseState, operationID)
	}
	if err != nil {
		return fmt.Errorf("failed to send shared pool lifecycle metering metadata: %w", err)
	}
	return nil
}

func pauseResumeMeteringModifyTS(operation *entity.PauseResumeOperation, terminal bool) (int64, error) {
	if operation.GetCreatedAt() == nil || !operation.GetCreatedAt().IsValid() {
		return 0, errors.New("pause/resume operation has invalid created_at")
	}
	createdTS := operation.GetCreatedAt().AsTime().Unix()
	if !terminal {
		return createdTS, nil
	}
	return max(time.Now().Unix(), createdTS+1), nil
}

func (a *SendMeteringMetadataAction) Name() string {
	return SendMeteringMetadataActionName
}

func (a *SendMeteringMetadataAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	if !a.enable {
		tilog.Info(ctx, "metering is not enabled, skip this action")
		return nil
	}
	val, _ := args.GetString(workflow.MeteringTypeKey)
	switch val {
	case workflow.MeteringTypeValueLogicMetaByTiDBCluster:
		tidbCluterId, err := args.GetString(workflow.TiDBClusterIdKey)
		if err != nil {
			tilog.Error(ctx, "failed to get tidb cluster id", zap.Error(err))
			return fmt.Errorf("failed to get tidb cluster id: %w", err)
		}
		deleteTiDBCluster, _ := args.GetString(workflow.MeteringDeleteTiDBClusterKey)
		tidbCluster, err := a.tidbClusterRepo.Load(ctx, tidbCluterId, true)
		if err != nil {
			tilog.Error(ctx, "failed to load tidb cluster", zap.Error(err))
			return fmt.Errorf("failed to load tidb cluster: %w", err)
		}

		sharedPoolId := tidbCluster.GetSharedPoolId()
		sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, true)
		if err != nil {
			tilog.Error(ctx, "failed to load shared pool", zap.Error(err))
			return fmt.Errorf("failed to load shared pool: %w", err)
		}
		if err := a.sendLogicMetaByTiDBCluster(ctx, tidbCluster, sharedPool, deleteTiDBCluster == workflow.MeteringDeleteTiDBClusterValue); err != nil {
			tilog.Error(ctx, "failed to send metering metadata", zap.String("tidbGroupId", tidbCluterId), zap.Error(err))
			return err
		}
	case workflow.MeteringTypeValueLogicMetaBySharedPool:
		sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
		if err != nil {
			tilog.Error(ctx, "failed to get shared pool id", zap.Error(err))
			return fmt.Errorf("failed to get shared pool id: %w", err)
		}
		sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, false)
		if err != nil {
			tilog.Error(ctx, "failed to load shared pool", zap.Error(err))
			return fmt.Errorf("failed to load shared pool: %w", err)
		}
		if err := a.sendLogicMetaBySharedPool(ctx, sharedPool); err != nil {
			tilog.Error(ctx, "failed to send metering metadata", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
			return err
		}
	case workflow.MeteringTypeValueSharedPoolMeta:
		sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
		if err != nil {
			tilog.Error(ctx, "failed to get shared pool id", zap.Error(err))
			return fmt.Errorf("failed to get shared pool id: %w", err)
		}
		deleteSharedPool, _ := args.GetString(workflow.MeteringDeleteSharedPoolKey)
		isDeleteSharedPoolMetering := isDeleteSharedPoolMetering(deleteSharedPool)
		sharedPool, err := a.loadSharedPoolForMetering(ctx, args, sharedPoolId, isDeleteSharedPoolMetering)
		if err != nil {
			tilog.Error(ctx, "failed to load shared pool", zap.Error(err))
			return fmt.Errorf("failed to load shared pool: %w", err)
		}

		tenantId, _ := args.GetString(workflow.TenantIdKey)
		if tenantId != "" {
			sharedPool.TenantId = tenantId
		}
		modifyTS, err := sharedPoolMetaModifyTS(args, isDeleteSharedPoolMetering)
		if err != nil {
			tilog.Error(ctx, "failed to resolve shared pool metering modify ts", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
			return err
		}
		if err := a.sendSharedPoolMeta(ctx, sharedPool, isDeleteSharedPoolMetering, modifyTS, "", ""); err != nil {
			tilog.Error(ctx, "failed to send metering metadata", zap.String("sharedPoolId", sharedPoolId), zap.Error(err))
			return err
		}
	default:
		// do nothind
	}

	return nil
}

func isDeleteSharedPoolMetering(deleteSharedPool string) bool {
	return deleteSharedPool == workflow.MeteringDeleteSharedPoolValue
}

func (a *SendMeteringMetadataAction) loadSharedPoolForMetering(ctx context.Context, args actions.ActionArgs, sharedPoolID string, preferSnapshot bool) (*entity.SharedPool, error) {
	if preferSnapshot {
		if sharedPoolStr, err := args.GetString(workflow.SharedPoolKey); err == nil && sharedPoolStr != "" {
			sharedPool := &entity.SharedPool{}
			// Delete workflows remove the row before the DAG runs, so only the delete tombstone
			// path reconstructs the shared pool from the serialized snapshot in DAG args.
			if err := protojson.Unmarshal([]byte(sharedPoolStr), &sharedPool.SharedPool); err != nil {
				return nil, fmt.Errorf("failed to unmarshal shared pool snapshot: %w", err)
			}
			return sharedPool, nil
		}
	}

	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return nil, err
	}
	return sharedPool, nil
}

func (a *SendMeteringMetadataAction) sendLogicMetaBySharedPool(ctx context.Context, sharedPool *entity.SharedPool) error {
	sharedPoolId := sharedPool.GetId()
	filter := &repo.TiDBClusterListFilter{
		SharedPoolIds: []string{sharedPoolId},
	}
	tidbClusters, err := a.tidbClusterRepo.List(ctx, filter)
	if err != nil {
		tilog.Error(ctx, "failed to list tidb cluster", zap.Error(err))
		return err
	}
	for _, tidbCluster := range tidbClusters {
		if err := a.sendLogicMetaByTiDBCluster(ctx, tidbCluster, sharedPool, false); err != nil {
			return err
		}
	}
	return nil
}

func (a *SendMeteringMetadataAction) sendLogicMetaByTiDBCluster(ctx context.Context, tidbCluster *entity.TiDBCluster, sharedPool *entity.SharedPool, deleteTiDBCluster bool) error {
	now := time.Now()

	storageEncryptionEnabled, err := a.getStorageEncryptionEnabled(ctx, sharedPool, tidbCluster.GetId())
	if err != nil {
		if deleteTiDBCluster && clustergateway.IsNotFoundError(err) {
			tilog.Warn(ctx, "keyspace not found while sending delete metering, fallback to storageEncryptionEnabled=false",
				zap.String("tidbClusterId", tidbCluster.GetId()),
				zap.String("sharedPoolId", sharedPool.GetId()),
			)
			storageEncryptionEnabled = false
		} else {
			tilog.Error(ctx, "failed to get storage encryption enabled", zap.String("tidbClusterId", tidbCluster.GetId()), zap.Error(err))
			return err
		}
	}

	// ru_version: safe fallback to "1" for legacy clusters where RuVersion may be 0
	ruVersionStr := "1"
	if tidbCluster.GetRuVersion() > 0 {
		ruVersionStr = strconv.Itoa(int(tidbCluster.GetRuVersion()))
	}

	// check if the tidbgroup is deleted
	metadata := map[string]any{
		"env":                      a.getEnv(),
		"region":                   a.getRegion(),
		"version":                  "1",
		"ru_version":               ruVersionStr,
		"plan":                     a.getPlan(sharedPool, tidbCluster.GetCapacityMode()),
		"capacityMode":             tidbCluster.GetCapacityMode().String(),
		"physicalID":               tidbCluster.GetSharedPoolId(),
		"tenantId":                 tidbCluster.GetTenantId(),
		"minRcu":                   strconv.FormatInt(int64(tidbCluster.GetRequestUnit().GetMin()), 10),
		"maxRcu":                   strconv.FormatInt(int64(tidbCluster.GetRequestUnit().GetMax()), 10),
		"deleted":                  deleteTiDBCluster,
		"storageEncryptionEnabled": storageEncryptionEnabled,
	}
	// for tikv
	var dataDiskType string
	var dataIops int64
	var dataThroughput int64
	var columnComputeDataDiskType string
	var columnWriteDataDiskType string
	for _, comp := range sharedPool.GetComponentGroups() {
		if comp.GetComponentType() == apiv1beta.ComponentType_TIKV {
			dataStorage := getDataStorage(comp.GetStorages())
			if dataStorage != nil {
				dataDiskType = dataStorage.GetType()
				dataIops = int64(dataStorage.GetIops())
				dataThroughput = int64(dataStorage.GetThroughputMbs())
			}
		}
		if comp.GetComponentType() == apiv1beta.ComponentType_TIFLASH_COMPUTE {
			dataStorage := getDataStorage(comp.GetStorages())
			if dataStorage != nil {
				columnComputeDataDiskType = dataStorage.GetType()
			}
		}
		if comp.GetComponentType() == apiv1beta.ComponentType_TIFLASH_WRITE {
			dataStorage := getDataStorage(comp.GetStorages())
			if dataStorage != nil {
				columnWriteDataDiskType = dataStorage.GetType()
			}
		}
	}

	metadata["dataDiskType"] = dataDiskType
	metadata["dataIops"] = dataIops
	if dataThroughput != 0 {
		metadata["dataThroughput"] = dataThroughput
	}
	metadata["columnComputeDataDiskType"] = columnComputeDataDiskType
	metadata["columnWriteDataDiskType"] = columnWriteDataDiskType

	data := &meteringsdkcommon.MetaData{
		ClusterID: tidbCluster.GetId(),
		ModifyTS:  now.Unix(),
		Type:      meteringsdkcommon.MetaTypeLogic,
		Metadata:  metadata,
	}
	if err := a.meteringMetaWriter.Write(ctx, data); err != nil {
		tilog.Error(ctx, "failed to write metering data", zap.Error(err))
		return err
	}
	return nil
}

func (a *SendMeteringMetadataAction) sendSharedPoolMeta(
	ctx context.Context,
	sharedPool *entity.SharedPool,
	deleted bool,
	modifyTS int64,
	pauseState string,
	operationID string,
) error {
	// check if the tidbgroup is deleted
	metadata := map[string]any{
		"env":          a.getEnv(),
		"region":       a.getRegion(),
		"version":      "1",
		"plan":         a.getPlan(sharedPool, sharedPool.GetCapacityMode()),
		"capacityMode": sharedPool.GetCapacityMode().String(),
		"physicalID":   sharedPool.GetId(),
		"tenantId":     sharedPool.TenantId,
		"deleted":      deleted,
	}
	if pauseState != "" {
		metadata["pauseState"] = pauseState
	}
	if operationID != "" {
		metadata["operationId"] = operationID
	}

	data := &meteringsdkcommon.MetaData{
		ClusterID: sharedPool.GetId(),
		ModifyTS:  modifyTS,
		Type:      meteringsdkcommon.MetaTypeSharedpool,
		Metadata:  metadata,
	}
	if err := a.meteringMetaWriter.Write(ctx, data); err != nil {
		tilog.Error(ctx, "failed to write metering metadata", zap.Error(err))
		return err
	}
	return nil
}

func sharedPoolMetaModifyTS(args actions.ActionArgs, deleted bool) (int64, error) {
	if !deleted {
		return time.Now().Unix(), nil
	}

	modifyTSArg, ok := args[workflow.MeteringModifyTSKey]
	if !ok {
		return time.Now().Unix(), nil
	}
	modifyTSStr := modifyTSArg.String()
	if modifyTSStr == "" {
		return time.Now().Unix(), nil
	}

	modifyTS, err := strconv.ParseInt(modifyTSStr, 10, 64)
	if err != nil {
		return 0, fmt.Errorf("failed to parse metering modify ts: %w", err)
	}
	return modifyTS, nil
}

func (a *SendMeteringMetadataAction) getEnv() string {
	return a.env
}

// getRegion return the real region but not the logical region
// us-west-2, us-east-2, etc.
func (a *SendMeteringMetadataAction) getRegion() string {
	return a.region
}

func (a *SendMeteringMetadataAction) getPlan(sharedPool *entity.SharedPool, capacityMode apiv1beta.CapacityMode) string {
	return entity.ResolveServicePlan(sharedPool, capacityMode, a.plan)
}

type keyspaceEncryptionConfig struct {
	Enabled *bool `json:"enabled"`
}

func (a *SendMeteringMetadataAction) getStorageEncryptionEnabled(ctx context.Context, sharedPool *entity.SharedPool, tidbClusterId string) (bool, error) {
	keyspaceMeta, err := a.clusterGatewayCli.GetKeySpace(ctx, sharedPool, tidbClusterId)
	if err != nil {
		// Fail fast on any GetKeySpace error (including NotFound) to avoid incorrect billing signals.
		return false, err
	}

	keyspaceConfig := keyspaceMeta.GetConfig()
	if keyspaceConfig == nil {
		tilog.Debug(ctx, "keyspace config is nil, storage encryption disabled", zap.String("tidbClusterId", tidbClusterId))
		return false, nil
	}

	encryptionJSON := keyspaceConfig["encryption"]
	if encryptionJSON == "" {
		tilog.Debug(ctx, "keyspace encryption config missing or empty, storage encryption disabled", zap.String("tidbClusterId", tidbClusterId))
	}
	return parseKeyspaceEncryptionEnabled(encryptionJSON)
}

func parseKeyspaceEncryptionEnabled(encryptionJSON string) (bool, error) {
	if encryptionJSON == "" {
		return false, nil
	}

	var cfg keyspaceEncryptionConfig
	if err := json.Unmarshal([]byte(encryptionJSON), &cfg); err != nil {
		return false, fmt.Errorf("unmarshal keyspace encryption config failed: %w", err)
	}
	if cfg.Enabled == nil {
		return false, errors.New("keyspace encryption config missing enabled field")
	}
	return *cfg.Enabled, nil
}

func getDataStorage(storages []*apiv1beta.Component_Storage) *apiv1beta.Component_Storage {
	for i := range storages {
		storage := storages[i]
		if storage.GetName() == apiv1beta.Component_Storage_DATA {
			return storage
		}
	}
	return nil
}