package dm_svc

import (
	"context"
	"encoding/json"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/tiflow/dm/pkg/checker"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"go.uber.org/zap"
)

type DMPrecheckWorker struct {
	dmTaskRepo      repo.DMTaskRepo
	mgmtSvcFacade   facade.MgmtSvcFacade
	encryptorFacade facade.EncryptorFacade
	// semaphore channel to limit concurrent tasks across Run invocations
	sem chan struct{}
}

const (
	// maxConcurrentPrechecks limits how many prechecks run concurrently in this process
	maxConcurrentPrechecks = 20
	// per-task timeout for precheck operations
	precheckTaskTimeout = 5 * time.Minute
)

func NewDMPrecheckWorker(dmTaskRepo repo.DMTaskRepo,
	mgmtSvcFacade facade.MgmtSvcFacade,
	encryptorFacade facade.EncryptorFacade,
) *DMPrecheckWorker {
	return &DMPrecheckWorker{
		dmTaskRepo:      dmTaskRepo,
		mgmtSvcFacade:   mgmtSvcFacade,
		encryptorFacade: encryptorFacade,
		sem:             make(chan struct{}, maxConcurrentPrechecks),
	}
}

// Run lists pending prechecks and attempts to start them without blocking.
func (w *DMPrecheckWorker) Run(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	startTime := time.Now()
	logger.Info("start to run dm task worker")

	tasks, err := w.listPendingPrechecks(ctx)
	if err != nil {
		logger.Error("failed to list dm precheck tasks", zap.Error(err))
		return err
	}

	for _, t := range tasks {
		w.tryStartTask(ctx, t, logger)
	}

	logger.Info("finish to run dm precheck worker", zap.Duration("duration", time.Since(startTime)))
	return nil
}

// listPendingPrechecks fetches pending precheck tasks from repository.
func (w *DMPrecheckWorker) listPendingPrechecks(ctx context.Context) ([]*entity.DMPrecheck, error) {
	return w.dmTaskRepo.ListPrechecks(ctx, &entity.DMPrecheckFilter{
		Stages: []constant.DMPrecheckStage{constant.DMPrecheckStagePending},
	})
}

// tryStartTask attempts a non-blocking acquire of semaphore and starts the task if slot available.
func (w *DMPrecheckWorker) tryStartTask(ctx context.Context, t *entity.DMPrecheck, parentLogger *zap.Logger) {
	logger := parentLogger.With(zap.String("precheck_id", t.ID), zap.String("cluster_id", t.ClusterID))

	select {
	case w.sem <- struct{}{}:
		// acquired slot, persist Running state
		t.Stage = constant.DMPrecheckStageRunning
		if err := w.dmTaskRepo.UpdatePrecheck(ctx, t.ID, map[string]any{"stage": t.Stage}); err != nil {
			logger.Error("failed to update dm precheck stage to running, skip starting task", zap.Error(err))
			// release slot and skip
			<-w.sem
			return
		}

		logger.Info("start to process dm precheck task")
		// start goroutine to process task
		go w.startTaskGoroutine(ctx, t, logger)
	default:
		logger.Info("concurrency limit reached, skip starting task for now")
		return
	}
}

// startTaskGoroutine runs the actual task logic inside a goroutine; it is responsible
// for releasing the semaphore slot and updating final task state.
func (w *DMPrecheckWorker) startTaskGoroutine(ctx context.Context, pre *entity.DMPrecheck, logger *zap.Logger) {
	defer func() { <-w.sem }()

	// create a per-task child context with timeout
	taskCtx, cancel := context.WithTimeout(ctx, precheckTaskTimeout)
	// store cancel func so Run can trigger cancellation via DB flag
	defer cancel()

	logger.Info("processing dm precheck task", zap.Duration("timeout", precheckTaskTimeout))

	if err := w.processPrecheck(taskCtx, pre); err != nil {
		logger.Error("failed to process dm precheck task", zap.Error(err))
		return
	}
}

// processPrecheck performs the full sequence of DM precheck items and persists each result.
func (w *DMPrecheckWorker) processPrecheck(ctx context.Context, precheck *entity.DMPrecheck) (err error) {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(zap.String("precheck_id", precheck.ID), zap.String("cluster_id", precheck.ClusterID))

	// Decrypt passwords in the request (reverse logic of encryptPasswords)
	if err := w.decryptPasswords(logger, precheck.Request); err != nil {
		logger.Error("failed to decrypt passwords", zap.Error(err))
		return w.markPrecheckAsFailed(ctx, precheck)
	}

	// Clear passwords after processing to ensure no plaintext passwords are stored in database
	defer func() {
		w.clearPasswordsInRequest(precheck.Request)
		// Update the database record with cleared passwords
		e := w.dmTaskRepo.UpdatePrecheck(ctx, precheck.ID, map[string]any{
			"request": precheck.Request,
		})
		if e != nil {
			logger.Error("failed to clear passwords in precheck request", zap.Error(e))
		}
	}()

	// iterate all defined precheck items and process them one by one
	// When mgmt returns a user-facing execution error (e.g. invalid route rules), cache it and
	// apply the same failure to remaining items without extra RPC calls. This keeps total and
	// per-status counts aligned while avoiding redundant calls that would fail the same way.
	var executionFailureReason, executionFailureWorkaround string
	for _, item := range vo.GetPrecheckItemList(precheck.Request) {
		name := item.String()
		logger = logger.With(zap.String("current_item", name))
		logger.Info("start to process dm precheck item")

		// if parent context is canceled, stop processing
		if ctx.Err() != nil {
			logger.Warn("task context canceled, stop processing precheck items", zap.Error(ctx.Err()))
			return w.markPrecheckAsFailed(ctx, precheck)
		}

		precheckItem := &entity.DMPrecheckItem{
			ID:           entity.GenerateIDWithPrefix(constant.DMPrecheckItemIDPrefix),
			DMPrecheckID: precheck.ID,
			Stage:        constant.DMPrecheckStageRunning,
			Description:  name,
			Type:         item,
		}
		bizErr := w.dmTaskRepo.CreatePrecheckItem(ctx, precheckItem)
		if bizErr != nil {
			logger.Error("failed to create dm precheck item record", zap.String("item", name), zap.Error(bizErr))
			return w.markPrecheckAsFailed(ctx, precheck)
		}

		precheck.Request.DmPrecheck.Item = item
		if executionFailureReason != "" {
			// Subsequent items: skip mgmt RPC; mark as not executed due to shared init failure.
			skippedReason := buildSkippedDueToInitFailureReason(executionFailureReason)
			if err := w.persistUserFacingPrecheckItemFailure(
				ctx, precheckItem, name, skippedReason, executionFailureWorkaround, logger,
			); err != nil {
				return w.markPrecheckAsFailed(ctx, precheck, precheckItem)
			}
			continue
		}

		resp, err := w.mgmtSvcFacade.DMPrecheck(
			ctx,
			(*globalserverlessv1.ServerlessDMServicePrecheckRequest)(precheck.Request),
		)
		if err != nil {
			if reason, workaround, ok := extractDMPrecheckUserError(err); ok {
				// User-facing config/validation error: persist as check failure (stage=finished)
				// and continue so every planned item appears in the report with consistent counts.
				executionFailureReason = reason
				executionFailureWorkaround = workaround
				logger.Info(
					"dm precheck returned user-facing error",
					zap.String("item", name),
					zap.String("reason", reason),
				)
				if updErr := w.persistUserFacingPrecheckItemFailure(
					ctx, precheckItem, name, reason, workaround, logger,
				); updErr != nil {
					return w.markPrecheckAsFailed(ctx, precheck, precheckItem)
				}
				continue
			}
			logger.Error("failed to call dm precheck", zap.String("item", name), zap.Error(err))
			return w.markPrecheckAsFailed(ctx, precheck, precheckItem)
		}

		var results entity.Results
		if err := json.Unmarshal(resp.Result, &results); err != nil {
			logger.Error("failed to unmarshal dm precheck result", zap.String("item", name), zap.Error(err))
			return w.markPrecheckAsFailed(ctx, precheck, precheckItem)
		}

		// overwrite certain check results to align with cloud expectations
		if err := overwritePrecheckResult(item, &results); err != nil {
			logger.Error("failed to overwrite precheck result", zap.String("item", name), zap.Error(err))
			return w.markPrecheckAsFailed(ctx, precheck, precheckItem)
		}

		err = w.dmTaskRepo.UpdatePrecheckItem(ctx, precheckItem.ID, map[string]any{
			"stage":   constant.DMPrecheckStageFinished,
			"results": results,
		})
		if err != nil {
			logger.Error("failed to update dm precheck item record", zap.String("item", name), zap.Error(err))
			return w.markPrecheckAsFailed(ctx, precheck, precheckItem)
		}
	}

	// all items processed successfully, mark finished
	precheck.Stage = constant.DMPrecheckStageFinished
	if err := w.dmTaskRepo.UpdatePrecheck(context.WithoutCancel(ctx), precheck.ID, map[string]any{
		"stage": precheck.Stage,
	}); err != nil {
		logger.Error("failed to update dm precheck stage to finished", zap.Error(err))
		return w.markPrecheckAsFailed(ctx, precheck)
	}
	return nil
}

// persistUserFacingPrecheckItemFailure stores a synthetic checker failure. Stage is finished
// (not failed) so API formatters expose reason/solution instead of the generic internal message.
func (w *DMPrecheckWorker) persistUserFacingPrecheckItemFailure(
	ctx context.Context,
	precheckItem *entity.DMPrecheckItem,
	name, reason, workaround string,
	logger *zap.Logger,
) error {
	results := buildDMPrecheckFailureResults(name, reason, workaround)
	if err := w.dmTaskRepo.UpdatePrecheckItem(ctx, precheckItem.ID, map[string]any{
		"stage":   constant.DMPrecheckStageFinished,
		"results": results,
	}); err != nil {
		logger.Error(
			"failed to update dm precheck item with user-facing error",
			zap.String("item", name),
			zap.Error(err),
		)
		return err
	}
	return nil
}

func (w *DMPrecheckWorker) markPrecheckAsFailed(
	ctx context.Context,
	precheck *entity.DMPrecheck,
	items ...*entity.DMPrecheckItem,
) error {
	precheck.Stage = constant.DMPrecheckStageFailed

	err := w.dmTaskRepo.UpdatePrecheck(context.WithoutCancel(ctx), precheck.ID, map[string]any{
		"stage": precheck.Stage,
	})
	if err != nil {
		return err
	}

	for _, item := range items {
		item.Stage = constant.DMPrecheckStageFailed
		err = w.dmTaskRepo.UpdatePrecheckItem(context.WithoutCancel(ctx), item.ID, map[string]any{
			"stage": item.Stage,
		})
		if err != nil {
			return err
		}
	}
	return nil
}

// decryptPasswords decrypts passwords in DMPrecheckRequest (reverse logic of encryptPasswords)
func (w *DMPrecheckWorker) decryptPasswords(logger *zap.Logger,
	req *entity.PrecheckRequest,
) error {
	if req == nil || req.DmPrecheck == nil {
		return nil
	}

	var err error

	// Decrypt target database password
	if req.DmPrecheck.TargetDb != nil && req.DmPrecheck.TargetDb.Password != "" {
		req.DmPrecheck.TargetDb.Password, err = w.encryptorFacade.DecryptBase64(req.DmPrecheck.TargetDb.Password)
		if err != nil {
			logger.Error("failed to decrypt target database password", zap.Error(err))
			return err
		}
	}

	// Decrypt source database passwords
	for i, source := range req.DmPrecheck.Sources {
		if source != nil && source.ConnProfile != nil && source.ConnProfile.Password != "" {
			source.ConnProfile.Password, err = w.encryptorFacade.DecryptBase64(source.ConnProfile.Password)
			if err != nil {
				logger.Error("failed to decrypt source database password", zap.Int("source_index", i), zap.Error(err))
				return err
			}
		}
	}

	return nil
}

func (w *DMPrecheckWorker) clearPasswordsInRequest(req *entity.PrecheckRequest) {
	if req == nil || req.DmPrecheck == nil {
		return
	}

	// Clear target database password
	if req.DmPrecheck.TargetDb != nil {
		req.DmPrecheck.TargetDb.Password = ""
	}

	// Clear source database passwords
	for _, source := range req.DmPrecheck.Sources {
		if source != nil && source.ConnProfile != nil {
			source.ConnProfile.Password = ""
		}
	}
}

func overwritePrecheckResult(checkItem regionalserverlessv1.DMPrecheck_PrecheckItem, pcResult *entity.Results) error {
	// currently, almost all precheck items in cloud have one result, but 'conn_num' may have two results
	if len(pcResult.Results) < 1 {
		return errors.Errorf("unexpected check result:%+v", pcResult.Results)
	}

	if checkItem == regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_TARGET_DB_PRIVILEGE_CHECKING {
		if pcResult.Results[0] == nil {
			return errors.Errorf("unexpected check result:%+v", pcResult.Results)
		}
		overwriteTargetPrivilegeResult(pcResult.Results[0])
	}

	if checkItem == regionalserverlessv1.DMPrecheck_PRECHECK_ITEM_PRIMARY_KEY_CHECKING {
		pcResult.Results = overwritePrimaryKeyResult(pcResult.Results)
	}

	// TODO: enable this when physical import is supported in serverless dm
	// if checkItem == dmconfig.LightningTableEmptyChecking {
	//	if pcResult.Results[0] == nil {
	//		return errors.Errorf("unexpected check result:%+v", pcResult.Results)
	//	}
	//	overwriteTableEmptyResult(pcResult.Results[0])
	// }

	return nil
}

func overwritePrimaryKeyResult(results []*checker.Result) []*checker.Result {
	for i := range results {
		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 results
}

func overwriteTargetPrivilegeResult(result *checker.Result) {
	// Currently, dm checker will return 'warning' if target DB doesn't have some privileges on '*.*' since it can't recognizes
	// some specified schema/table-level privileges.
	// For cloud dm, we will return error simply and handle the error manually.
	if result.State == checker.StateWarning {
		result.State = checker.StateFailure
	}
}

// func overwriteTableEmptyResult(result *checker.Result) {
//	// For cloud dm, table empty check can be skipped through ignore, so we set this failure only to warning.
//	if result.State == checker.StateFailure {
//		result.State = checker.StateWarning
//	}