package export_svc

import (
	"context"
	"errors"
	"strconv"
	"time"

	"github.com/lithammer/shortuuid/v3"
	"github.com/tidbcloud/eventbus-datastruct/event"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/internal/config"
	"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/errorcode"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	"github.com/tidbcloud/serverless-service/internal/metric"
	"github.com/tidbcloud/serverless-service/internal/utils"
	event_constant "github.com/tidbcloud/serverless-service/pkg/event"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
)

const (
	exportCRNotFoundReason = "export cr not found"
)

type ExportSyncerService struct {
	exportRepo      repo.ExportRepo
	infraFacade     facade.InfraFacade
	mgmtSvc         facade.MgmtSvcFacade
	exportSvc       service.ExportService
	eventbusService *eventbus.EventbusService
}

func NewExportSyncerService(
	exportRepo repo.ExportRepo,
	infraFacade facade.InfraFacade,
	mgmtSvc facade.MgmtSvcFacade,
	exportSvc service.ExportService,
	eventbusService *eventbus.EventbusService,
) *ExportSyncerService {
	return &ExportSyncerService{
		exportRepo:      exportRepo,
		infraFacade:     infraFacade,
		mgmtSvc:         mgmtSvc,
		exportSvc:       exportSvc,
		eventbusService: eventbusService,
	}
}

func (s *ExportSyncerService) Run(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	logger.Info("start to sync export")

	exports, bizErr := s.exportRepo.ListByFilter(ctx, entity.ExportFilter{States: []constant.ExportState{
		constant.ExportStatePreparing,
		constant.ExportStateRunning,
	}})
	if bizErr != nil {
		logger.Error("failed to list active exports", zap.Error(bizErr))
		return bizErr
	}

	startTime := time.Now()
	logger.Info("[global]sync export with infra, start.", zap.Int("exportCount", len(exports)))

	var errs error
	var group errgroup.Group
	activeErrs := make([]error, len(exports))
	concurrency := config.Global().BizConfig.ExportSyncerConcurrency
	if concurrency < 1 {
		concurrency = 20
	}
	group.SetLimit(concurrency)
	for i, export := range exports {
		group.Go(func() error {
			activeErrs[i] = s.syncActiveExport(ctx, export)
			return nil
		})
	}
	_ = group.Wait()
	errs = multierr.Append(errs, multierr.Combine(activeErrs...))
	if err := s.reconcileTerminalClusterStatuses(ctx); err != nil {
		errs = multierr.Append(errs, err)
	}

	logger.Info(
		"[global]sync export with infra, done.",
		zap.Any("costSeconds", time.Since(startTime).Seconds()),
		zap.Error(errs),
	)
	return errs
}

func (s *ExportSyncerService) syncActiveExport(ctx context.Context, export *entity.Export) error {
	switch export.State {
	case constant.ExportStatePreparing:
		return s.createPreparingExport(ctx, export)
	case constant.ExportStateRunning:
		return s.syncExport(ctx, export)
	default:
		return nil
	}
}

func (s *ExportSyncerService) createPreparingExport(ctx context.Context, export *entity.Export) error {
	logger := utils.GetWorkerLoggerFromContext(ctx).With(
		zap.String("export_id", export.ID),
		zap.String("cluster_id", export.ClusterID),
	)

	current, bizErr := s.exportRepo.GetById(ctx, export.ClusterID, export.ID)
	if bizErr != nil {
		logger.Error("failed to load preparing export", zap.Error(bizErr))
		return bizErr
	}
	if current.State != constant.ExportStatePreparing {
		return nil
	}

	cluster, resolvedProxyEndpoint, err := s.resolveClusterRuntime(ctx, current.ClusterID)
	if err != nil {
		logger.Error("failed to resolve export cluster runtime", zap.Error(err))
		return err
	}
	proxyEndpoint := current.ProxyEndpoint
	if proxyEndpoint == "" {
		proxyEndpoint = resolvedProxyEndpoint
		if err := s.persistProxyEndpoint(ctx, current, proxyEndpoint); err != nil {
			logger.Error("failed to persist export proxy endpoint", zap.Error(err))
			return err
		}
	}
	exportCR, buildErr := s.exportSvc.NewInfraAPIExportObject(ctx, current, cluster)
	if buildErr != nil {
		if buildErr.ErrCode() == bizerr.ErrCodeParams &&
			buildErr.ErrMsg() == service.InvalidExportCreatePayloadMessage {
			current.Reason = service.InvalidExportCreatePayloadMessage
			return s.failPreparingExport(ctx, current, cluster)
		}
		logger.Error("failed to build export CR", zap.Error(buildErr))
		return buildErr
	}

	// Cancellation can win while credentials or target metadata are being resolved.
	current, bizErr = s.exportRepo.GetById(ctx, export.ClusterID, export.ID)
	if bizErr != nil {
		logger.Error("failed to reload export before creating CR", zap.Error(bizErr))
		return bizErr
	}
	if current.State != constant.ExportStatePreparing {
		return nil
	}

	if err := s.infraFacade.CreateExport(ctx, proxyEndpoint, exportCR); err != nil {
		logger.Error("failed to create export CR", zap.Error(err))
		return err
	}

	current.State = constant.ExportStateRunning
	transitioned, bizErr := s.exportRepo.TransitionPreparingState(ctx, current)
	if bizErr != nil {
		logger.Error("failed to transition export to running", zap.Error(bizErr))
		return bizErr
	}
	if transitioned {
		return nil
	}

	latest, bizErr := s.exportRepo.GetById(ctx, export.ClusterID, export.ID)
	if bizErr != nil {
		logger.Error("failed to reload export after transition race", zap.Error(bizErr))
		return bizErr
	}
	if latest.State == constant.ExportStateRunning {
		return nil
	}
	projectID, err := strconv.ParseUint(current.ProjectID, 10, 64)
	if err != nil {
		return err
	}
	if err := s.infraFacade.DeleteExport(context.WithoutCancel(ctx), proxyEndpoint, projectID, current.ID); err != nil {
		logger.Error("failed to compensate export CR after state changed", zap.Error(err))
		return err
	}
	return nil
}

func (s *ExportSyncerService) failPreparingExport(
	ctx context.Context,
	export *entity.Export,
	cluster *tidbmgmtservicev1.Cluster,
) error {
	export.State = constant.ExportStateFailed
	transitioned, bizErr := s.exportRepo.TransitionPreparingState(ctx, export)
	if bizErr != nil {
		log.Error(ctx, "failed to mark invalid preparing export as failed",
			zap.String("export_id", export.ID), zap.Error(bizErr))
		return bizErr
	}
	if !transitioned {
		return nil
	}
	if export.ClusterStatusID != 0 {
		if err := s.cleanupClusterStatus(ctx, export); err != nil {
			return err
		}
	}
	metric.ExportFailedGauge.WithLabelValues(export.ClusterID, export.ID, export.Reason).Set(1)
	go s.publishTerminalEvent(ctx, export, cluster)
	return nil
}

func (s *ExportSyncerService) reconcileTerminalClusterStatuses(ctx context.Context) error {
	hasClusterStatus := true
	terminalStates := []constant.ExportState{
		constant.ExportStateSucceed,
		constant.ExportStateFailed,
		constant.ExportStateCanceled,
		constant.ExportStateDeleted,
	}

	var errs error
	for _, state := range terminalStates {
		exports, bizErr := s.exportRepo.ListByFilter(ctx, entity.ExportFilter{
			State:            state,
			HasClusterStatus: &hasClusterStatus,
			IncludeDeleted:   true,
		})
		if bizErr != nil {
			errs = multierr.Append(errs, bizErr)
			continue
		}
		for _, export := range exports {
			if err := s.cleanupClusterStatus(ctx, export); err != nil {
				errs = multierr.Append(errs, err)
			}
		}
	}
	return errs
}

func (s *ExportSyncerService) syncExport(
	ctx context.Context,
	export *entity.Export,
) error {
	logger := utils.GetWorkerLoggerFromContext(ctx).With(
		zap.String("export_id", export.ID),
		zap.String("cluster_id", export.ClusterID),
	)

	proxyEndpoint := export.ProxyEndpoint
	var cluster *tidbmgmtservicev1.Cluster
	if proxyEndpoint == "" {
		var resolvedProxyEndpoint string
		var err error
		cluster, resolvedProxyEndpoint, err = s.resolveClusterRuntime(ctx, export.ClusterID)
		if err != nil {
			logger.Error("failed to resolve export cluster runtime", zap.Error(err))
			return err
		}
		proxyEndpoint = resolvedProxyEndpoint
		if err := s.persistProxyEndpoint(ctx, export, proxyEndpoint); err != nil {
			logger.Error("failed to persist export proxy endpoint", zap.Error(err))
			return err
		}
	}

	projectID, err := strconv.ParseUint(export.ProjectID, 10, 64)
	if err != nil {
		logger.Error("failed to parse export project id", zap.Error(err))
		return err
	}

	exportCR, err := s.infraFacade.GetExport(ctx, proxyEndpoint, projectID, export.ID)
	if err != nil {
		if apierrors.IsNotFound(err) {
			export.Reason = exportCRNotFoundReason
			return s.updateExportTerminalStatusWithCluster(
				ctx, export, constant.ExportStateFailed, cluster,
			)
		}
		logger.Error("failed to get export CR from infra", zap.Error(err))
		return err
	}

	if exportCR.DeletionTimestamp != nil {
		logger.Info("ignore export sync because export CR is deleting")
		return nil
	}

	switch exportCR.Status.Phase {
	case cloudv1.ExportPhaseFailed:
		export.Reason = exportCR.Status.Message
		return s.updateExportTerminalStatusWithCluster(
			ctx, export, constant.ExportStateFailed, cluster,
		)
	case cloudv1.ExportPhaseCompleted:
		return s.updateExportTerminalStatusWithCluster(
			ctx, export, constant.ExportStateSucceed, cluster,
		)
	case cloudv1.ExportPhaseCancel:
		return s.updateExportTerminalStatusWithCluster(
			ctx, export, constant.ExportStateCanceled, cluster,
		)
	default:
		logger.Debug(
			"ignore export sync because phase is not terminal",
			zap.String("phase", string(exportCR.Status.Phase)),
		)
		return nil
	}
}

func (s *ExportSyncerService) updateExportTerminalStatusWithCluster(
	ctx context.Context,
	export *entity.Export,
	state constant.ExportState,
	cluster *tidbmgmtservicev1.Cluster,
) error {
	if cluster == nil {
		var err error
		cluster, _, err = s.resolveClusterRuntime(ctx, export.ClusterID)
		if err != nil {
			return err
		}
	}
	return s.updateExportTerminalStatus(ctx, export, cluster, state)
}

func (s *ExportSyncerService) persistProxyEndpoint(
	ctx context.Context,
	export *entity.Export,
	proxyEndpoint string,
) error {
	if bizErr := s.exportRepo.UpdateProxyEndpoint(ctx, export.ID, export.ClusterID, proxyEndpoint); bizErr != nil {
		return bizErr
	}
	export.ProxyEndpoint = proxyEndpoint
	return nil
}

func (s *ExportSyncerService) resolveClusterRuntime(
	ctx context.Context,
	clusterID string,
) (*tidbmgmtservicev1.Cluster, string, error) {
	clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return nil, "", err
	}

	cluster, bizErr := s.mgmtSvc.GetCluster(ctx, clusterIDUint64, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if bizErr != nil {
		return nil, "", bizErr
	}

	serverlessCluster := cluster.GetServerlessCluster()
	if serverlessCluster == nil || serverlessCluster.GetCluster() == nil {
		return nil, "", errors.New("serverless cluster is nil")
	}
	proxyEndpoint := serverlessCluster.GetCluster().GetProxyEndpoint()
	if proxyEndpoint == "" {
		return nil, "", errors.New("proxy endpoint is empty")
	}

	return cluster, proxyEndpoint, nil
}

func (s *ExportSyncerService) updateExportTerminalStatus(
	ctx context.Context,
	export *entity.Export,
	cluster *tidbmgmtservicev1.Cluster,
	state constant.ExportState,
) error {
	if state == export.State {
		return nil
	}

	export.State = state
	exportID := export.ID
	updatedExport, bizErr := s.exportRepo.UpdateRunningState(ctx, export)
	if bizErr != nil {
		if bizErr.ErrCode() == errorcode.ErrCodeExportUpdateStateAffectedNoRows {
			log.Warn(
				ctx,
				"update export status failed. Maybe the export status is not running now",
				zap.String("export_id", exportID),
			)
			return nil
		}
		log.Error(ctx, "update export status failed", zap.Error(bizErr))
		return bizErr
	}
	if updatedExport != nil {
		export = updatedExport
	}

	if export.ClusterStatusID != 0 {
		if err := s.cleanupClusterStatus(ctx, export); err != nil {
			return err
		}
	}

	if state == constant.ExportStateFailed {
		metric.ExportFailedGauge.WithLabelValues(export.ClusterID, export.ID, export.Reason).Set(1)
	}

	go s.publishTerminalEvent(ctx, export, cluster)
	return nil
}

func (s *ExportSyncerService) cleanupClusterStatus(ctx context.Context, export *entity.Export) error {
	clusterIDUint64, parseErr := strconv.ParseUint(export.ClusterID, 10, 64)
	if parseErr != nil {
		return parseErr
	}

	rctx := context.WithoutCancel(ctx)
	delErr := s.mgmtSvc.DeleteClusterCustomStatus(rctx, clusterIDUint64, export.ClusterStatusID)
	if delErr != nil && delErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
		log.Error(ctx, "failed to delete export cluster custom status",
			zap.String("export_id", export.ID),
			zap.Uint64("cluster_status_id", export.ClusterStatusID),
			zap.Error(delErr),
		)
		return delErr
	}

	if err := s.exportRepo.ClearClusterStatusID(ctx, export.ID, export.ClusterID, export.ClusterStatusID); err != nil {
		log.Error(ctx, "failed to clear export cluster status id",
			zap.String("export_id", export.ID),
			zap.Uint64("cluster_status_id", export.ClusterStatusID),
			zap.Error(err),
		)
		return err
	}
	export.ClusterStatusID = 0
	return nil
}

func (s *ExportSyncerService) publishTerminalEvent(
	ctx context.Context,
	export *entity.Export,
	cluster *tidbmgmtservicev1.Cluster,
) {
	logger := log.WithContextL(ctx)
	status := event.EventStatusSucceeded
	if export.State == constant.ExportStateFailed {
		status = event.EventStatusFailed
	}

	exportEvent, err := service.BuildExportEvent(export, cluster, status, event_constant.EventTypeCreateExport)
	if err != nil {
		logger.Error("failed to build export event", zap.Error(err))
		return
	}
	if err := s.eventbusService.Publish(&tieventbus.Event{
		ID:   shortuuid.New(),
		Type: event.DisplayEventTopic,
		Data: exportEvent,
	}); err != nil {
		logger.Error("failed to publish events center", zap.Error(err))
	}
}