package branch

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

	"github.com/lithammer/shortuuid/v3"
	"github.com/pingcap/errors"
	core_event "github.com/tidbcloud/eventbus-datastruct/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/pkgv2/password"
	"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/service"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"github.com/tidbcloud/serverless-service/pkg/event"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.opentelemetry.io/otel"
	"go.uber.org/zap"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
)

var (
	Tracer                = otel.Tracer("branch-reconciler")
	focusedPhases         = []cloudv1.BranchPhase{cloudv1.BranchPhaseRunning}
	syncingBranchStatuses = []constant.BranchStatus{
		constant.BranchStatusPending,
		constant.BranchStatusCreating,
		constant.BranchStatusRestoring,
	}
)

const (
	MaxConcurrentBranch     = 10
	branchFastPollingWindow = 2 * time.Minute
)

type BranchReconciler struct {
	infraFacade                facade.InfraFacade
	branchRepo                 repo.BranchRepo
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade
	encryptorFacade            facade.EncryptorFacade
	accountFacade              facade.AccountFacade
	userroleService            *service.UserRoleService
	eventbusService            *eventbus.EventbusService
	now                        func() time.Time
	executor                   *branchSyncExecutor
}

func NewBranchReconciler(
	infraFacade facade.InfraFacade,
	branchRepo repo.BranchRepo,
	clusterProxyProviderFacade facade.ClusterProxyProviderFacade,
	encryptorFacade facade.EncryptorFacade,
	accountFacade facade.AccountFacade,
	userroleService *service.UserRoleService,
	eventbusService *eventbus.EventbusService,
) *BranchReconciler {
	return &BranchReconciler{
		infraFacade:                infraFacade,
		branchRepo:                 branchRepo,
		clusterProxyProviderFacade: clusterProxyProviderFacade,
		encryptorFacade:            encryptorFacade,
		accountFacade:              accountFacade,
		userroleService:            userroleService,
		eventbusService:            eventbusService,
		now:                        time.Now,
		executor:                   newBranchSyncExecutor(MaxConcurrentBranch),
	}
}

func (r *BranchReconciler) RunFast(ctx context.Context) error {
	cutoff := r.now().Add(-branchFastPollingWindow)
	return r.run(ctx, "fast", repo.BranchFilter{
		Statuses:     syncingBranchStatuses,
		CreatedAfter: &cutoff,
	})
}

func (r *BranchReconciler) Run(ctx context.Context) error {
	cutoff := r.now().Add(-branchFastPollingWindow)
	return r.run(ctx, "regular", repo.BranchFilter{
		Statuses:          syncingBranchStatuses,
		CreatedAtOrBefore: &cutoff,
	})
}

func (r *BranchReconciler) run(ctx context.Context, stage string, filter repo.BranchFilter) error {
	logger := utils.GetWorkerLoggerFromContext(ctx).With(zap.String("polling_stage", stage))
	logger.Info("start to sync branches")

	branches, bizErr := r.branchRepo.ListBranchesByFilter(ctx, filter)
	if bizErr != nil {
		logger.Error("failed to list syncing branches", zap.Error(bizErr))
		return bizErr
	}

	result := r.executor.Execute(ctx, branches, r.syncBranch)
	failedCount := 0
	for _, outcome := range result.outcomes {
		if outcome.err == nil {
			continue
		}
		logger.Error("failed to sync branch",
			zap.String("branch_id", outcome.branch.ID),
			zap.Error(outcome.err))
		failedCount++
	}

	logger.Info("finish syncing branches",
		zap.Int("branch_count", len(branches)),
		zap.Int("admitted_count", len(result.outcomes)),
		zap.Int("skipped_count", result.skipped),
		zap.Int("failed_count", failedCount))
	if failedCount > 0 {
		return errors.New("some branches failed to sync, check logs for details")
	}
	return nil
}

func (r *BranchReconciler) syncBranch(ctx context.Context, branch *entity.Branch) error {
	ctx = logid.WithLogid(ctx, logid.GenLogid())
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branch.ID))

	if branch.Status == constant.BranchStatusPending {
		return r.createCR(ctx, branch)
	}

	branchCR, err := r.infraFacade.GetBranch(ctx, branch.ProjectID, branch.ID)
	if err != nil {
		if apierrors.IsNotFound(err) {
			logger.Info("branch CR is not found")
			return nil
		}
		logger.Error("can't get branch from infra", zap.Error(err))
		return errors.Errorf("can't get branch from infra: %s", err)
	}

	return r.reconcileBranch(ctx, branch, branchCR)
}

func (r *BranchReconciler) createCR(ctx context.Context, branch *entity.Branch) error {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branch.ID))

	if branch.CRSpec == nil {
		logger.Error("branch CR spec is nil")
		return errors.New("branch CR spec is nil")
	}

	var branchCR cloudv1.Branch
	if err := json.Unmarshal(branch.CRSpec, &branchCR); err != nil {
		logger.Error("failed to unmarshal branch CR spec", zap.Error(err))
		return errors.Errorf("failed to unmarshal branch CR spec: %s", err)
	}

	err := r.infraFacade.CreateBranch(ctx, &branchCR)
	if err != nil {
		if !apierrors.IsAlreadyExists(err) {
			logger.Error("failed to create branch CR", zap.Error(err))
			return errors.Errorf("failed to create branch CR: %s", err)
		}
		logger.Info("branch CR already exists, skip creation")
		return nil
	}
	logger.Info("[branch lifecycle] branch CR created successfully",
		zap.Int64("age_ms", time.Since(branch.CreatedAt).Milliseconds()))
	return r.updateBranchStatus(ctx, branch, constant.BranchStatusCreating)
}

func (r *BranchReconciler) reconcileBranch(
	ctx context.Context,
	branch *entity.Branch,
	branchCR *cloudv1.Branch,
) error {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branchCR.Name))

	logger.Info("start to reconcile branch")

	if !slices.Contains(focusedPhases, branchCR.Status.Phase) {
		logger.Info("ignore event", zap.String("branch_phase", string(branchCR.Status.Phase)))
		return nil
	}

	if branchCR.DeletionTimestamp != nil {
		logger.Info("ignore event, branch is going to delete")
		return nil
	}

	switch branchCR.Status.Phase {
	case cloudv1.BranchPhaseRunning:
		if branchCR.Annotations[constant.AnnoKeyKeyspaceRestore] == "" ||
			branchCR.Annotations[constant.AnnoKeyKeyspaceRestore] == constant.AnnoKeyspaceRestoreFinish {
			err := r.initBranchUsers(ctx, branch, branchCR)
			if err != nil {
				logger.Error("init branch delegate users failed", zap.Error(err))
				return err
			}

			err = r.updateBranchStatus(ctx, branch, constant.BranchStatusRunning)
			if err != nil {
				logger.Error("update branch status to running failed", zap.Error(err))
				return err
			}
		}
	}

	return nil
}

func (r *BranchReconciler) updateBranchStatus(
	ctx context.Context,
	branch *entity.Branch,
	status constant.BranchStatus,
) error {
	if branch.Status != status {
		prevStatus := branch.Status
		branch.Status = status
		now := time.Now()
		branch.UpdatedAt = &now
		if status == constant.BranchStatusRunning && branch.CreationFinishedTime == nil {
			branch.CreationFinishedTime = &now
		}
		bizErr := r.branchRepo.Update(
			ctx, branch,
			[]string{"status", "updated_at", "creation_finished_time"},
			// in case the status is changed during this reconciler round,
			// for example deleted, NoRecord is returned, next reconciler round will handle it
			map[string]any{"status": prevStatus},
		)
		if bizErr != nil {
			log.Error(ctx, "updateBranchStatus failed",
				zap.String("branch_id", branch.ID),
				zap.String("prev_status", string(prevStatus)), zap.String("status", string(status)),
				zap.Error(bizErr))
			return bizerr.New(bizerr.ErrCodeDB, "save branch to db failed: %s", bizErr)
		}
		log.Info(ctx, "[branch lifecycle] update branch status successfully",
			zap.String("branch_id", branch.ID),
			zap.String("prev_status", string(prevStatus)),
			zap.String("status", string(status)),
			zap.Int64("e2e_ms", time.Since(branch.CreatedAt).Milliseconds()))

		if status == constant.BranchStatusRunning {
			annotationsMap, err := branch.UnmarshalAnnotations()
			// Event should not break the main flow, just log error
			if err != nil {
				log.Error(ctx, "failed to unmarshal annotation", zap.Error(err))
			}
			clusterID, err := strconv.ParseUint(branch.MainBranchID, 10, 64)
			if err != nil {
				log.Error(
					ctx,
					"failed to parse cluster id",
					zap.Error(err),
					zap.String("cluster_id", branch.MainBranchID),
				)
			}
			detail := event.BranchDetail{
				ID:             branch.ID,
				Name:           branch.Name,
				TenantID:       branch.TenantID,
				ProjectID:      branch.ProjectID,
				MainBranchID:   branch.MainBranchID,
				ParentBranchID: branch.ParentID,
				Annotations:    annotationsMap,
			}
			detailString, err := json.Marshal(detail)
			if err != nil {
				log.Error(ctx, "failed to marshal branch detail", zap.Error(err))
			}

			var displayEventType string
			var summaryMsg string
			switch prevStatus {
			case constant.BranchStatusCreating:
				displayEventType = event.EventTypeCreateBranch
				summaryMsg = fmt.Sprintf("Branch %s successfully created.", branch.Name)
			case constant.BranchStatusRestoring:
				displayEventType = event.EventTypeResetBranch
				summaryMsg = fmt.Sprintf("Branch %s successfully reset.", branch.Name)
			default:
				displayEventType = event.EventTypeUnknown
			}

			data := &core_event.DisplayEvent{
				DisplayEventType: displayEventType,
				DisplayStatus:    core_event.EventStatusSucceeded,
				Severity:         core_event.EventSeverityInfo,
				Summary:          summaryMsg,
				HappenedAtMs:     uint64(time.Now().UTC().UnixMilli()),
				AuthorID:         branch.CreatedByID,
				ClusterID:        clusterID,
				ProjectID:        branch.ProjectID,
				TenantID:         branch.TenantID,
				Detail:           detailString,
			}
			go func(data *core_event.DisplayEvent) {
				if err := r.eventbusService.Publish(&tieventbus.Event{
					ID:   shortuuid.New(),
					Type: core_event.DisplayEventTopic,
					Data: data,
				}); err != nil {
					log.Error(ctx, "failed to publish events center", zap.Error(err))
					return
				}
			}(data)
		}
	}
	return nil
}

func (r *BranchReconciler) initBranchUsers(
	ctx context.Context,
	branch *entity.Branch,
	branchCR *cloudv1.Branch,
) (err error) {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branch.ID))

	if branch.CloudAdminPwdSynced {
		logger.Info("init branch users has already runned, skip")
		return nil
	}

	proxy, err := r.clusterProxyProviderFacade.GetBranchProxy(ctx, branchCR, nil)
	if err != nil {
		logger.Error("get cluster proxy failed")
		return err
	}

	if branch.EncryptedCloudAdminPwd != "" {
		// cloud_admin_synced flag not set but cloud_admin pwd saved, means syncCloudAdminPwd() failed last time
		logger.Info("cloud_admin_synced flag not set but cloud_admin pwd saved")
		cloudAdminPwd, err := r.encryptorFacade.DecryptBase64(branch.EncryptedCloudAdminPwd)
		if err != nil {
			return err
		}
		return r.syncCloudAdminPwd(ctx, proxy, branch, cloudAdminPwd)
	}

	// 1. init root user password
	operator := &clusterproxy.ClusterOperator{
		Username:   constant.CloudAdminUsername,
		AuthMethod: clusterproxy.AuthMethodPassword,
		Credential: "",
	}
	proxy = proxy.SetOperator(operator)

	var rootPwd string
	if branch.EncryptedRootPassword != "" {
		rootPwd, err = r.encryptorFacade.DecryptBase64(branch.EncryptedRootPassword)
		if err != nil {
			return err
		}
	} else {
		rootPwd = password.Decorate(utils.RandomString(32))
	}
	status, err := branchCR.AwsFreeTierClusterStatusFromBranch()
	if err != nil {
		return err
	}
	rootUser := status.KeyspaceToken + ".root"
	err = proxy.ChangePassword(ctx, rootUser, rootPwd)
	if err != nil {
		logger.Error("change root password using cloud_admin failed")
		return err
	}
	logger.Info("change root password using cloud_admin succeeded")

	// 2. init delegate user
	svcOperator := &service.SqlOperator{
		Username:   rootUser,
		AuthMethod: constant.AuthMethodPassword,
		Credential: rootPwd,
	}
	dbusers, err := r.accountFacade.GetDBUsers(ctx, branch.TenantID, branch.ProjectID, status.KeyspaceToken)
	if err != nil {
		return err
	}
	if len(dbusers) == 0 {
		return errors.New("initial delegate users is empty")
	}
	var userInfos []*service.UserInfo
	for _, user := range dbusers {
		userInfos = append(userInfos, &service.UserInfo{
			Username:   user.Username,
			UserEmail:  user.Email,
			UserType:   constant.SqlUserTypeDelegate,
			AuthMethod: constant.AuthMethodToken,
			Credential: "",
			Roles:      []string{string(user.BuiltinRole)},
		})
	}
	// update status temporarily, createUsers could only apply on running branch
	entityStatus := branch.Status
	branch.Status = constant.BranchStatusRunning
	err = r.userroleService.CreateUsers(ctx, svcOperator, branch, userInfos)
	branch.Status = entityStatus
	if err != nil {
		logger.Error("create delegate users failed")
		return err
	}
	logger.Info("create delegate users succeeded")

	// 3. init cloud admin password
	cloudAdminPwd := password.Decorate(utils.RandomString(32))
	branch.EncryptedCloudAdminPwd, err = r.encryptorFacade.EncryptBase64(cloudAdminPwd)
	if err != nil {
		return err
	}
	branch.EncryptedRootPassword = "" // clear root password
	err = r.branchRepo.Update(ctx, branch, []string{"encrypted_cloud_admin_pwd", "encrypted_root_password"}, nil)
	if err != nil {
		logger.Error("save cloud_admin password and clear root password failed")
		return err
	}

	return r.syncCloudAdminPwd(ctx, proxy, branch, cloudAdminPwd)
}

func (r *BranchReconciler) syncCloudAdminPwd(
	ctx context.Context,
	proxy facade.ClusterProxyFacade,
	branch *entity.Branch,
	cloudAdminPwd string,
) error {
	logger := log.WithContextL(ctx).With(zap.String("branch_id", branch.ID))

	// InitCloudAdminPassword is idempotent if you call with the same password so we are relieved to call it multiple times.
	err := proxy.InitCloudAdminPassword(ctx, cloudAdminPwd)
	if err != nil {
		logger.Error("init cloud_admin password failed")
		return err
	}
	logger.Info("init cloud_admin password succeeded")

	branch.CloudAdminPwdSynced = true
	err = r.branchRepo.Update(ctx, branch, []string{"cloud_admin_pwd_synced"}, nil)
	if err != nil {
		logger.Error("set cloud_admin_synced flag failed")
		return err
	}
	logger.Info("set cloud_admin_synced flag succeed")

	return nil
}