package action

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

	"braces.dev/errtrace"
	"github.com/tidbcloud/fastflow/pkg/entity/run"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/constant"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/cvt"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/metrics"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/sqluser"
	workflowcommon "github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

const (
	PostCreateCluster                 = "post_create_cluster"
	forbidDelegateUsersForStarterFlag = "forbid_delegate_users_for_starter"
)

type PostCreateClusterAction struct {
	clusterRepo          repository.ClusterRepo
	transactioner        db.Transactioner
	regionalService      adapter.RegionalService
	sys9Lifecycle        adapter.Sys9Lifecycle
	accountSvc           adapter.AccountService
	encryptorService     adapter.EncryptorService
	dbUserSvc            sqluser.Service
	eventCommitter       adapter.WALCommitter
	clusterProxyProvider adapter.ClusterProxyProvider
}

func NewPostCreateClusterAction(
	clusterRepo repository.ClusterRepo,
	regionalService adapter.RegionalService,
	sys9Lifecycle adapter.Sys9Lifecycle,
	accountSvc adapter.AccountService,
	encryptorService adapter.EncryptorService,
	dbUserSvc sqluser.Service,
	eventCommitter adapter.WALCommitter,
	transactioner db.Transactioner,
	clusterProxyProvider adapter.ClusterProxyProvider) *PostCreateClusterAction {
	return &PostCreateClusterAction{
		clusterRepo:          clusterRepo,
		regionalService:      regionalService,
		sys9Lifecycle:        sys9Lifecycle,
		accountSvc:           accountSvc,
		encryptorService:     encryptorService,
		dbUserSvc:            dbUserSvc,
		eventCommitter:       eventCommitter,
		transactioner:        transactioner,
		clusterProxyProvider: clusterProxyProvider,
	}
}

type PostCreateClusterParameter struct {
	ClusterId string `json:"cluster_id"`
	// InitSQLs is a JSON-encoded []string of SQL statements to execute after cluster initialization.
	InitSQLs string `json:"init_sqls"`
}

func (a *PostCreateClusterAction) Name() string {
	return PostCreateCluster
}

func (a *PostCreateClusterAction) ParameterNew() any {
	return &PostCreateClusterParameter{}
}

func (a *PostCreateClusterAction) Run(aCtx run.ExecuteContext, params any) error {
	cctx, span := workflow.GenerateContext(aCtx)

	createClusterParam, ok := params.(*PostCreateClusterParameter)
	if !ok {
		tilog.Error(cctx, "expect PostCreateClusterParameter parameter")
		return errors.New("expect PostCreateClusterParameter parameter")
	}

	var err error
	tilog.Info(cctx, "start to create post cluster", zap.String("cluster_id", createClusterParam.ClusterId))
	defer func() {
		if err != nil {
			tilog.Error(
				cctx,
				"failed to post create cluster",
				zap.String("cluster_id", createClusterParam.ClusterId),
				zap.Error(err),
			)
		} else {
			tilog.Info(cctx, "end to post create cluster", zap.String("cluster_id", createClusterParam.ClusterId))
		}
		span.End()
	}()

	err = a.transactioner.Transaction(cctx, func(ctx context.Context) error {
		var en *entity.Cluster
		en, err = a.clusterRepo.Get(cctx, createClusterParam.ClusterId, true)
		if err != nil {
			metrics.ClusterInitializedFailedTotal.Inc()
			tilog.Error(cctx, "get cluster failed", zap.String("cluster", createClusterParam.ClusterId), zap.Error(err))
			return err
		}

		if en == nil {
			tilog.Error(cctx, "cluster not found", zap.String("cluster", createClusterParam.ClusterId))
			return nil
		}

		if !en.DeletedAt.IsZero() {
			tilog.Warn(
				cctx,
				"the cluster is deleted, interrupt initialize task",
				zap.String("cluster", createClusterParam.ClusterId),
			)
			return nil
		}

		if en.CreationFinishedAt != nil && !en.CreationFinishedAt.IsZero() {
			tilog.Warn(cctx, "the cluster initialize task is done", zap.String("cluster", en.ClusterId))
			return nil
		}

		// Parse init_sqls from the JSON-encoded string parameter
		var initSQLs []string
		if createClusterParam.InitSQLs != "" {
			if unmarshalErr := json.Unmarshal([]byte(createClusterParam.InitSQLs), &initSQLs); unmarshalErr != nil {
				// Log and proceed without init SQLs to avoid blocking cluster creation
				// due to a JSON parsing failure.
				tilog.Warn(cctx, "failed to unmarshal init sqls, proceeding without init sqls",
					zap.String("cluster", createClusterParam.ClusterId),
					zap.Error(unmarshalErr),
				)
				initSQLs = nil
			}
		}

		if isPGStarterClusterForPostCreate(en) {
			opID, ok := aCtx.ShareData().Get(workflowcommon.KeyOperationId)
			if !ok || opID == "" {
				err := errors.New("missing operation id in share data for pg cluster")
				tilog.Error(cctx, "missing operation id in share data for pg cluster", zap.String("cluster", en.ClusterId))
				return workflow.NewTerminalError(err)
			}
			if err := a.initPGCluster(cctx, en, opID); err != nil {
				tilog.Warn(
					cctx,
					"init pg cluster failed",
					zap.Uint64("tenant", en.TenantId),
					zap.String("cluster", en.ClusterId),
					zap.Error(err),
				)
				metrics.ClusterInitializedFailedTotal.Inc()
				return err
			}
		} else {
			var regionalClusters []*regionalserverlessv1.ServerlessRegionalServiceClusterInfo
			regionalClusters, err = a.regionalService.ListCluster(
				cctx,
				en.Region,
				&globalserverlessv1.ServerlessServiceListRequest{
					Filter: &globalserverlessv1.ClusterFilter{
						ClusterIds: []string{en.ClusterId},
						ProjectIds: []uint64{en.ProjectId},
					},
				},
			)
			if err != nil || regionalClusters == nil || len(regionalClusters) == 0 {
				metrics.ClusterInitializedFailedTotal.Inc()
				tilog.Warn(
					cctx,
					"get cluster failed",
					zap.Uint64("tenant", en.TenantId),
					zap.String("cluster", en.ClusterId),
					zap.Error(err),
				)
				return errors.New("get cluster failed")
			}

			regionalCluster := regionalClusters[0]
			if regionalCluster == nil {
				tilog.Warn(cctx, "regional cluster info is nil", zap.String("cluster", en.ClusterId))
				return errors.New("regional cluster info is nil")
			}

			if err := a.initCluster(cctx, en, regionalCluster, initSQLs); err != nil {
				tilog.Warn(
					cctx,
					"init cluster failed",
					zap.Uint64("tenant", en.TenantId),
					zap.String("cluster", regionalCluster.ClusterId),
					zap.Error(err),
				)
				metrics.ClusterInitializedFailedTotal.Inc()
				return err
			}
		}

		costTime := time.Since(en.CreatedAt).Seconds()
		metrics.ClusterInitializedTotalTime.Observe(costTime)

		if costTime > constant.LongPendingSec {
			metrics.ClusterInitializedLongTime.Inc()
			tilog.Warn(
				cctx,
				"the cluster initialize task cost long time",
				zap.String("cluster", en.ClusterId),
				zap.Float64("cost", costTime),
			)
		}

		tilog.Info(
			cctx,
			"the cluster initialize task is done",
			zap.String("cluster", en.ClusterId),
			zap.Float64("cost", costTime),
		)
		return nil
	})

	return err
}

func (a *PostCreateClusterAction) initPGCluster(ctx context.Context, cluster *entity.Cluster, operationID string) error {
	if operationID == "" {
		return errors.New("missing operation id for pg cluster")
	}
	if cluster.ExternalInstanceID == "" {
		return errors.New("missing external instance id for pg cluster")
	}

	operation, err := a.sys9Lifecycle.GetOperation(ctx, cluster, operationID)
	if err != nil {
		return err
	}
	if operation == nil {
		return errors.New("empty sys9 operation response")
	}

	if operation.Phase != "OPERATION_PHASE_SUCCEEDED" {
		return errors.New("wait sys9 operation succeeded")
	}

	connectionInfo, err := a.sys9Lifecycle.GetManagedDatabaseConnectionInfo(ctx, cluster, cluster.ExternalInstanceID)
	if err != nil {
		return err
	}
	if cluster.Cluster == nil {
		cluster.Cluster = &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{}
	}
	if cluster.Cluster.Endpoints == nil {
		cluster.Cluster.Endpoints = &regionalserverlessv1.Endpoints{}
	}
	if cluster.Cluster.Endpoints.Public == nil {
		cluster.Cluster.Endpoints.Public = &regionalserverlessv1.PublicEndpoint{}
	}
	cluster.Cluster.Endpoints.Public.Host = connectionInfo.Host
	cluster.Cluster.Endpoints.Public.Port = int64(connectionInfo.Port)

	cluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL
	now := time.Now()
	cluster.CreationFinishedAt = &now
	cluster.ServicePlan = globalcommonv1.ServicePlan_Starter
	cluster.EncryptedRootPassword = ""
	cluster.EncryptedCloudAdminPassword = ""
	cluster.Version = "sys9"

	if err := a.clusterRepo.Save(ctx, cluster); err != nil {
		return fmt.Errorf("save pg cluster failed: %w", err)
	}

	events, err := cluster.ReadyEvent(ctx)
	if err != nil {
		metrics.ClusterInitializedFailedTotal.Inc()
		return errtrace.Wrap(err)
	}

	return errtrace.Wrap(a.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, events))
}

func isPGStarterClusterForPostCreate(cluster *entity.Cluster) bool {
	if cluster == nil || cluster.Region == nil {
		return false
	}
	if cluster.ServicePlan != globalcommonv1.ServicePlan_Starter {
		return false
	}
	return strings.EqualFold(cluster.EngineType, "POSTGRESQL_COMPATIBLE")
}

func (a *PostCreateClusterAction) initCluster(
	ctx context.Context,
	cluster *entity.Cluster,
	regionalCluster *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	initSQLs []string,
) error {
	if !entity.IsReady4Launch(ctx, regionalCluster) {
		return errors.New("wait regional cluster normal")
	}

	attributes := entity.ClusterAttributes{}
	if regionalCluster.TokenBasedAuthEnabled {
		attributes.TokenBasedAuthEnabled = true
	}

	cid, err := strconv.ParseUint(regionalCluster.ClusterId, 10, 64)
	if err != nil {
		return fmt.Errorf("parse cluster id failed: %w", err)
	}

	proxy, err := a.clusterProxyProvider.GetClusterProxy(
		ctx,
		cid,
		regionalCluster.ProxyEndpoint,
		regionalCluster.Version,
		nil,
	)
	if err != nil {
		metrics.ClusterInitializedFailedTotal.Inc()
		return fmt.Errorf("get cluster proxy failed: %w", err)
	}

	cloudAdminPwd, err := a.encryptorService.DecryptBase64(ctx, cluster.EncryptedCloudAdminPassword)
	if err != nil {
		metrics.ClusterInitializedFailedTotal.Inc()
		return fmt.Errorf("decrypt cloud admin password failed: %w", err)
	}

	rootPwd, err := a.encryptorService.DecryptBase64(ctx, cluster.EncryptedRootPassword)
	if err != nil {
		metrics.ClusterInitializedFailedTotal.Inc()
		return fmt.Errorf("decrypt root password failed: %w", err)
	}

	if err := a.initClusterUsers(ctx, regionalCluster, attributes, proxy, cluster, cloudAdminPwd, rootPwd); err != nil {
		metrics.ClusterInitializedFailedTotal.Inc()
		return fmt.Errorf("init cluster user failed: %w", err)
	}

	if len(initSQLs) > 0 {
		tilog.Info(ctx, "executing init sqls", zap.String("cluster", cluster.ClusterId), zap.Int("count", len(initSQLs)))
		if err := proxy.ExecuteMultiSQL(ctx, initSQLs); err != nil {
			metrics.ClusterInitializedFailedTotal.Inc()
			return errtrace.Wrap(bizerr.WrapError(err, "execute init sqls failed"))
		}
		tilog.Info(ctx, "init sqls executed successfully", zap.String("cluster", cluster.ClusterId))
	}

	cluster.Version = regionalCluster.Version
	cluster.UserStatus = cvt.ClusterStatusRegionalToGlobal(regionalCluster.State)
	now := time.Now()
	cluster.CreationFinishedAt = &now

	cluster.Attributes = attributes
	cluster.EncryptedRootPassword = "" // clear root password

	if err := a.clusterRepo.Save(ctx, cluster); err != nil {
		return fmt.Errorf("save cluster failed: %w", err)
	}
	cluster.ServicePlan = globalcommonv1.ServicePlan_Starter
	if regionalCluster.GetAutoScaling().GetMaxRcu() > 0 {
		cluster.ServicePlan = globalcommonv1.ServicePlan_Essential
	}
	events, err := cluster.ReadyEvent(ctx)
	if err != nil {
		metrics.ClusterInitializedFailedTotal.Inc()
		return errtrace.Wrap(err)
	}

	return errtrace.Wrap(a.eventCommitter.BatchWrite(ctx, clusterevent.TopicClusterEvent, events))
}

func (a *PostCreateClusterAction) initClusterUsers(
	ctx context.Context,
	cluster *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	attributes entity.ClusterAttributes,
	proxy adapter.ClusterProxy,
	en *entity.Cluster,
	cloudAdminPwd, rootPwd string,
) error {
	// change password for cloud_admin
	if cloudAdminPwd != "" {
		if err := proxy.InitCloudAdminPassword(ctx, cloudAdminPwd); err != nil {
			return errtrace.Wrap(bizerr.WrapError(err, "create cloud_admin"))
		}
		// use cloud_admin to manage other users

		if err := proxy.SetOperator(&clusterproxy.ClusterOperator{
			Username:   constant.DBUserCloudAdmin,
			AuthMethod: clusterproxy.AuthMethodPassword,
			Credential: cloudAdminPwd,
		}); err != nil {
			return err
		}
	} else {
		// use privilege db to manage other users
		if err := proxy.SetOperator(&clusterproxy.ClusterOperator{
			Username:   "",
			AuthMethod: "", // this field will be amended to "privilegeDB" in clusterproxy/tidbcloud/clusterproxy.go:amendOperator
			Credential: "",
		}); err != nil {
			return err
		}
		tilog.Warn(ctx, "this cluster didn't enable SEM")
	}
	tilog.Info(ctx, "user password is set", zap.String("user", constant.DBUserCloudAdmin))

	if err := proxy.ChangePassword(ctx, constant.DBUserRoot, rootPwd); err != nil {
		return errtrace.Wrap(bizerr.WrapError(err, "change password for %s", constant.DBUserRoot))
	}
	tilog.Info(ctx, "user password is set", zap.String("user", constant.DBUserRoot))

	if attributes.TokenBasedAuthEnabled {
		if err := a.initDelegateUsers(ctx, cluster, attributes, en, rootPwd); err != nil {
			return errtrace.Wrap(err)
		}
	}
	return nil
}

func (a *PostCreateClusterAction) initDelegateUsers(
	ctx context.Context,
	cluster *regionalserverlessv1.ServerlessRegionalServiceClusterInfo,
	attributes entity.ClusterAttributes,
	en *entity.Cluster,
	rootPwd string,
) (err error) {
	hitFeatureFlag, err := a.accountSvc.IsHitFeatureFlag(ctx, en.TenantId, forbidDelegateUsersForStarterFlag)
	if err != nil {
		tilog.Warn(
			ctx,
			"failed to check feature flag, continue to init delegate users",
			zap.String("cluster", en.ClusterId),
			zap.Uint64("tenant", en.TenantId),
			zap.String("flag_name", forbidDelegateUsersForStarterFlag),
			zap.Error(err),
		)
	} else if hitFeatureFlag {
		tilog.Info(
			ctx,
			"skip init delegate users by feature flag",
			zap.String("cluster", en.ClusterId),
			zap.Uint64("tenant", en.TenantId),
			zap.String("flag_name", forbidDelegateUsersForStarterFlag),
		)
		return nil
	}

	// 2130684 is dev-us-east-1-f02 TestInfra's tenant id.
	if en.TenantId == 1372813089209302377 || en.TenantId == 2130684 {
		tilog.Info(ctx, "ignore to init delegate users", zap.String("cluster", en.ClusterId))
		return nil
	}
	tilog.Info(ctx, "try to init delegate users by root", zap.String("cluster id", en.ClusterId))
	candidates, err := a.accountSvc.GetClusterInitialDelegateUsers(ctx, en.TenantId, en.ProjectId, cluster.UserPrefix)
	if err != nil {
		return errtrace.Wrap(err)
	}

	tilog.Info(ctx, "candidates length", zap.Int("len", len(candidates)), zap.String("cluster id", en.ClusterId))
	out := make([]*adapter.DelegateDBUserInfo, 0)
	for _, each := range candidates {
		if each.BuiltinRole != globalserverlessv1.SqlUser_BUILTIN_ROLE_USER_ADMIN {
			continue
		}

		out = append(out, each)
	}

	tilog.Info(
		ctx,
		"initial admin delegate users",
		zap.Int("user length", len(out)),
		zap.String("cluster id", en.ClusterId),
	)
	if len(out) == 0 {
		tilog.Warn(ctx, "initial delegate users is empty", zap.String("cluster id", en.ClusterId))
		return nil
	}

	params := &sqluser.BatchCreateUserParams{
		// The initial delegate users are created by root.
		// The password of root was set same as cloud_admin.
		Operator: &globalserverlessv1.SqlUserOperator{
			Username:   constant.DBUserRoot,
			AuthMethod: globalserverlessv1.SqlUser_AUTH_METHOD_PASSWORD,
			Credential: rootPwd,
		},
		Cluster:           cluster,
		ClusterAttributes: attributes,
		Users:             nil, // fill later
	}
	for _, user := range out {
		params.Users = append(params.Users, &sqluser.CreateUserInfo{
			UserType:    globalserverlessv1.SqlUser_TYPE_DELEGATE,
			UserName:    user.Username,
			UserEmail:   user.Email,
			AuthMethod:  globalserverlessv1.SqlUser_AUTH_METHOD_TOKEN,
			Credential:  "",
			BuiltinRole: user.BuiltinRole,
			CustomRoles: nil, // no custom roles for initial delegate user
		})
	}
	if err := a.dbUserSvc.CreateSqlUsers(ctx, params); err != nil {
		return fmt.Errorf("create delegate users failed: %w", err)
	}
	tilog.Info(ctx, "all delegate users are created", zap.String("cluster id", en.ClusterId))
	return nil
}