package premium_dm

import (
	"context"
	"encoding/json"
	"strings"

	"github.com/BurntSushi/toml"
	dmconfig "github.com/pingcap/tiflow/dm/config"
	"github.com/pingcap/tiflow/dm/pkg/checker"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/pkg/agent"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/bizerr-wrapper"
	agentpb "github.com/tidbcloud/dataflow-service/dataflow-agent-ng/idl/pbgen/proto/ng_agent"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

const premiumPrecheckItemIDPrefix = "pci"

func generatePremiumPrecheckItemID() string {
	return entity.GenerateDMIDWithPrefix(premiumPrecheckItemIDPrefix)
}

func filterAllExceptPremium(checkItem string) []string {
	items := make([]string, 0)
	for item := range dmconfig.AllCheckingItems {
		if item != checkItem && item != dmconfig.AllChecking {
			items = append(items, item)
		}
	}
	return items
}

func (w *PremiumDMPrecheckWorker) doPremiumPrecheckDMJob(
	ctx context.Context,
	agentClient *agent.NGRegionalClient,
	precheckID string,
	logger *zap.Logger,
	subTaskCfgs []*dmconfig.SubTaskConfig,
	checkItem string,
	item *entity.PremiumDMPrecheckItem,
	agentReq *agentpb.PrecheckDMJobReq,
) (userFacingInitFailureMessage *premiumDMPrecheckUserMessage, bizErr bizerr.BizError) {
	if item == nil {
		return nil, bizerr_wrapper.NewInternalErr("%s", "precheck item is nil")
	}

	precheckItem, itemID, createErr := w.createRunningPrecheckItem(ctx, precheckID, checkItem, item, logger)
	if createErr != nil {
		return nil, createErr
	}

	defer func() {
		w.finishPremiumPrecheckDMJob(ctx, itemID, precheckItem, logger, &bizErr, recover())
	}()

	reqCopy, reqErr := buildPremiumPrecheckAgentReq(agentReq, subTaskCfgs, checkItem)
	if reqErr != nil {
		return nil, failPremiumPrecheckItem(precheckItem, logger, reqErr)
	}

	agentResp, err := agentClient.PrecheckDMJob(ctx, &reqCopy)
	if userFacingInitFailureMessage, bizErr = handlePremiumPrecheckAgentError(
		err,
		agentResp,
		checkItem,
		precheckItem,
		logger,
	); bizErr != nil || userFacingInitFailureMessage != nil {
		return userFacingInitFailureMessage, bizErr
	}

	results, resultsErr := decodePremiumPrecheckResults(agentResp.CheckResults, checkItem)
	if resultsErr != nil {
		return nil, failPremiumPrecheckItem(precheckItem, logger, resultsErr)
	}
	precheckItem.Results = results
	precheckItem.Stage = string(entity.PremiumDMPrecheckStageFinished)
	userFacingInitFailureMessage, _ = extractPremiumDMPrecheckUserMessageFromResults(&results)
	return userFacingInitFailureMessage, nil
}

type premiumPrecheckResultError struct {
	msg string
	err error
}

func (w *PremiumDMPrecheckWorker) finishPremiumPrecheckDMJob(
	ctx context.Context,
	itemID string,
	precheckItem *entity.PremiumDMPrecheckItem,
	logger *zap.Logger,
	bizErr *bizerr.BizError,
	panicValue any,
) {
	if panicValue != nil {
		logger.Error("doPremiumPrecheckDMJob panic", zap.Any("err", panicValue))
		precheckItem.Stage = string(entity.PremiumDMPrecheckStageFailed)
		precheckItem.ErrMsg = "internal error"
		if *bizErr == nil {
			*bizErr = bizerr_wrapper.NewInternalErr("panic: %v", panicValue)
		}
	}
	if updateErr := w.dao.UpdatePrecheckItem(ctx, itemID, map[string]any{
		"stage":   precheckItem.Stage,
		"err_msg": precheckItem.ErrMsg,
		"results": precheckItem.Results,
	}); updateErr != nil {
		logger.Error("failed to update precheck item", zap.Error(updateErr))
	}
}

func failPremiumPrecheckItem(
	precheckItem *entity.PremiumDMPrecheckItem,
	logger *zap.Logger,
	resultErr *premiumPrecheckResultError,
) bizerr.BizError {
	logger.Error(resultErr.msg, zap.Error(resultErr.err))
	precheckItem.Stage = string(entity.PremiumDMPrecheckStageFailed)
	precheckItem.ErrMsg = resultErr.msg
	return bizerr_wrapper.NewInternalErr("%s", resultErr.msg)
}

func (w *PremiumDMPrecheckWorker) createRunningPrecheckItem(
	ctx context.Context,
	precheckID string,
	checkItem string,
	item *entity.PremiumDMPrecheckItem,
	logger *zap.Logger,
) (*entity.PremiumDMPrecheckItem, string, bizerr.BizError) {
	itemID := item.ID
	if itemID == "" {
		itemID = generatePremiumPrecheckItemID()
	}
	precheckItem := &entity.PremiumDMPrecheckItem{
		ID:           itemID,
		DMPrecheckID: precheckID,
		Description:  checkItem,
		Type:         item.Type,
		Stage:        string(entity.PremiumDMPrecheckStageRunning),
	}
	if bizErr := w.dao.CreatePrecheckItem(ctx, precheckItem); bizErr != nil {
		logger.Error("failed to create precheck item", zap.Error(bizErr))
		return nil, "", bizErr
	}
	return precheckItem, itemID, nil
}

func buildPremiumPrecheckAgentReq(
	agentReq *agentpb.PrecheckDMJobReq,
	subTaskCfgs []*dmconfig.SubTaskConfig,
	checkItem string,
) (agentpb.PrecheckDMJobReq, *premiumPrecheckResultError) {
	reqCopy := *agentReq
	reqCopy.ExtraCheckingItems = nil
	if checkItem == PrimaryKeyChecking {
		reqCopy.ExtraCheckingItems = []string{PrimaryKeyChecking}
	}
	reqCopy.TaskConfig = make([]string, 0, len(subTaskCfgs))
	for i := range subTaskCfgs {
		taskConfig, err := encodePremiumPrecheckTaskConfig(subTaskCfgs[i], checkItem)
		if err != nil {
			return agentpb.PrecheckDMJobReq{}, err
		}
		if taskConfig == "" {
			continue
		}
		reqCopy.TaskConfig = append(reqCopy.TaskConfig, taskConfig)
	}
	if len(reqCopy.TaskConfig) == 0 {
		return agentpb.PrecheckDMJobReq{}, &premiumPrecheckResultError{
			msg: "no valid subtask config for precheck",
			err: nil,
		}
	}
	return reqCopy, nil
}

func encodePremiumPrecheckTaskConfig(
	subTaskCfg *dmconfig.SubTaskConfig,
	checkItem string,
) (string, *premiumPrecheckResultError) {
	if subTaskCfg == nil {
		return "", nil
	}
	cfgCopy := *subTaskCfg
	cfgCopy.IgnoreCheckingItems = filterAllExceptPremium(checkItem)
	var strBuilder strings.Builder
	if err := toml.NewEncoder(&strBuilder).Encode(&cfgCopy); err != nil {
		return "", &premiumPrecheckResultError{
			msg: "failed to encode subtask config",
			err: err,
		}
	}
	return strBuilder.String(), nil
}

func handlePremiumPrecheckAgentError(
	err error,
	agentResp *agentpb.PrecheckDMJobResp,
	checkItem string,
	precheckItem *entity.PremiumDMPrecheckItem,
	logger *zap.Logger,
) (*premiumDMPrecheckUserMessage, bizerr.BizError) {
	if err == nil && agentResp != nil {
		return nil, nil
	}
	if reason, ok := extractPremiumDMPrecheckUserError(err); ok {
		logger.Info("dm precheck returned user-facing initialization error", zap.String("reason", reason))
		precheckItem.Results = buildPremiumDMPrecheckFailureResults(checkItem, reason, "")
		precheckItem.Stage = string(entity.PremiumDMPrecheckStageFinished)
		return &premiumDMPrecheckUserMessage{Reason: reason}, nil
	}
	msg := "failed to precheck dm job"
	logger.Error(msg, zap.Error(err))
	precheckItem.Stage = string(entity.PremiumDMPrecheckStageFailed)
	precheckItem.ErrMsg = msg
	return nil, bizerr_wrapper.NewInternalErr("%s", msg)
}

func decodePremiumPrecheckResults(
	checkResults []byte,
	checkItem string,
) (entity.PremiumResults, *premiumPrecheckResultError) {
	var results entity.PremiumResults
	if err := json.Unmarshal(checkResults, &results); err != nil {
		return entity.PremiumResults{}, &premiumPrecheckResultError{
			msg: "failed to unmarshal check result",
			err: err,
		}
	}
	normalizePremiumDMPrecheckUserErrorResults(&results)
	if err := overwritePremiumPrecheckResult(checkItem, &results); err != nil {
		return entity.PremiumResults{}, &premiumPrecheckResultError{
			msg: "failed to overwrite precheck result",
			err: err,
		}
	}
	if len(results.Results) > 0 &&
		results.Results[0] != nil &&
		results.Results[0].Desc == premiumDMPrecheckInitializationDesc {
		results.Results[0].Desc = checkItem
	}
	return results, nil
}

func (w *PremiumDMPrecheckWorker) persistPremiumPrecheckUserFailure(
	ctx context.Context,
	precheckID string,
	item *entity.PremiumDMPrecheckItem,
	message *premiumDMPrecheckUserMessage,
	logger *zap.Logger,
) bizerr.BizError {
	if item == nil {
		return bizerr_wrapper.NewInternalErr("%s", "precheck item is nil")
	}
	itemID := item.ID
	if itemID == "" {
		itemID = generatePremiumPrecheckItemID()
	}

	precheckItem := &entity.PremiumDMPrecheckItem{
		ID:           itemID,
		DMPrecheckID: precheckID,
		Description:  item.Description,
		Type:         item.Type,
		Stage:        string(entity.PremiumDMPrecheckStageRunning),
	}
	if bizErr := w.dao.CreatePrecheckItem(ctx, precheckItem); bizErr != nil {
		logger.Error("failed to create precheck item", zap.Error(bizErr))
		return bizErr
	}

	precheckItem.Results = buildPremiumDMSkippedPrecheckFailureResults(item.Description, message)
	precheckItem.Stage = string(entity.PremiumDMPrecheckStageFinished)
	if bizErr := w.dao.UpdatePrecheckItem(ctx, itemID, map[string]any{
		"stage":   precheckItem.Stage,
		"err_msg": precheckItem.ErrMsg,
		"results": precheckItem.Results,
	}); bizErr != nil {
		logger.Error("failed to update precheck item", zap.Error(bizErr))
		return bizErr
	}
	return nil
}

func buildPremiumDMSkippedPrecheckFailureResults(
	desc string,
	message *premiumDMPrecheckUserMessage,
) entity.PremiumResults {
	return buildPremiumDMPrecheckFailureResults(
		desc,
		"This check was not executed because precheck initialization failed: "+message.Reason,
		message.Solution,
	)
}

func overwritePremiumPrecheckResult(checkItem string, result *entity.PremiumResults) error {
	if result == nil || len(result.Results) < 1 {
		return bizerr_wrapper.NewInternalErr("unexpected check result")
	}

	if checkItem == dmconfig.TargetDBPrivilegeChecking {
		if result.Results[0] == nil {
			return bizerr_wrapper.NewInternalErr("unexpected check result")
		}
		if result.Results[0].State == checker.StateWarning {
			result.Results[0].State = checker.StateFailure
		}
	}

	if checkItem == PrimaryKeyChecking {
		for i := range result.Results {
			if result.Results[i] == nil {
				continue
			}
			if result.Results[i].Desc == premiumDMPrecheckInitializationDesc {
				continue
			}
			result.Results[i].Instruction = "For Alibaba Cloud RDS, only tables with primary keys are" +
				" supported for migration. Please add primary keys to the tables and try again."
		}
	}

	return nil
}