package impl

import (
	"context"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service/app/internal/backuperr"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	restoreRepo "github.com/tidbcloud/dataflow-service/app/repo/restore"
	"github.com/tidbcloud/dataflow-service/internal/model/util"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
)

type logBackupLongTermSynchronizationFromInfraImpl struct {
	transProvider         db.TransProvider
	logBackupRepo         backupRepo.LogBackupRepo
	logBackupTruncateRepo backupRepo.LogBackupTruncateRepo
	restoreRepo           restoreRepo.RestoreRepo
	bucketRepo            repo.BucketRepo

	mgmtCli              grpcli.MgmtSvcGrpcClient
	backupRegionalFacade infraapi.BackupFacade
}

func NewLogBackupLongTermSynchronizationFromInfra(
	transProvider db.TransProvider,
	logBackupRepo backupRepo.LogBackupRepo,
	logBackupTruncateRepo backupRepo.LogBackupTruncateRepo,
	restoreRepo restoreRepo.RestoreRepo,
	bucketRepo repo.BucketRepo,
	coreCli grpcli.CoreGrpcClient, // kept for signature compat; non-cluster core APIs may be used elsewhere
	backupRegionalFacade infraapi.BackupFacade,
	mgmtCli grpcli.MgmtSvcGrpcClient,
) backup.LogBackupLongTermSynchronizationFromInfra {
	return &logBackupLongTermSynchronizationFromInfraImpl{
		transProvider:         transProvider,
		logBackupRepo:         logBackupRepo,
		logBackupTruncateRepo: logBackupTruncateRepo,
		restoreRepo:           restoreRepo,
		bucketRepo:            bucketRepo,

		mgmtCli:              mgmtCli,
		backupRegionalFacade: backupRegionalFacade,
	}
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) SyncAllLogBackup(ctx context.Context) bizerr.BizError {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)
	logBackups, err := s.logBackupRepo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.PreloadBackupSchedule().DeletedAtIsNull()
	})
	if err != nil {
		logger.Error("fail to list log backups", zap.Error(err))
		return err
	}
	// it will both sync log backups from infra and regional cluster server
	logBackupInfraCRs, iErr := s.backupRegionalFacade.ListLogBackups(ctx)
	if iErr != nil {
		logger.Error("fail to list log backups from infra", zap.Error(iErr))
		return bizerr_wrapper.NewInternalErr("%s", iErr)
	}
	logBackupInfraCRsMap := make(map[string]*cloudv1.Backup, len(logBackupInfraCRs))
	for i := range logBackupInfraCRs {
		logBackupInfraCRsMap[logBackupInfraCRs[i].Name] = &logBackupInfraCRs[i]
	}

	logger.Info("sync all log backups",
		zap.Int("log backup count", len(logBackups)),
		zap.Int("log backup infra cr count", len(logBackupInfraCRs)))
	for _, logBackup := range logBackups {
		if err := s.syncOneLogBackup(ctx, &logBackup, logBackupInfraCRsMap); err != nil {
			logger.Error("fail to sync log backup", zap.Error(err))
		}
	}
	return nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) syncOneLogBackup(
	ctx context.Context, logBackup *entity.LogBackup, logBackupInfraCRsMap map[string]*cloudv1.Backup,
) bizerr.BizError {
	ctx = log.WithLogger(ctx, log.GetLogger(ctx).With(zap.Uint64("log backup id", logBackup.ID.Uint64())))
	logger := log.GetLogger(ctx)
	if logBackup.BackupSchedule == nil {
		logger.Debug("log backup has no backup schedule")
		return nil
	}
	cr := logBackupInfraCRsMap[logBackup.CRName()]
	if cr == nil {
		logger.Debug("log backup not found in infra")
		return nil
	}

	if err := s.compareAndUpdateLogBackup(ctx, logBackup, logBackupInfraCRsMap[logBackup.CRName()]); err != nil {
		return err
	}
	return nil
}

//nolint:gocognit,gocyclo
func (s *logBackupLongTermSynchronizationFromInfraImpl) compareAndUpdateLogBackup(
	ctx context.Context, logBackup *entity.LogBackup, logBackupCR *cloudv1.Backup,
) bizerr.BizError {
	logger := log.GetLogger(ctx).With(zap.Uint64("log backup id", logBackup.ID.Uint64()))
	logger.Info("compare and update log backup from infra")

	updated := false
	status := s.convertLogBackupStatus(logBackupCR)
	needWaitSyncStoppedStatus := status == entity.LogBackupStatusRunning &&
		logBackup.Status == entity.LogBackupStatusStopping
	if status != "" && logBackup.Status != status {
		if !logBackup.Status.IsFinishedStatus() && !needWaitSyncStoppedStatus {
			logBackup.Status = status
			updated = true
		}
	}

	if logBackupCR.Status.BackupTS != "" && *logBackup.StartTS != logBackupCR.Status.BackupTS {
		logBackup.StartTS = &logBackupCR.Status.BackupTS
		updated = true
	}
	if logBackupCR.Status.BackupPath != "" && logBackup.Storage != logBackupCR.Status.BackupPath {
		logBackup.Storage = logBackupCR.Status.BackupPath
		updated = true
	}
	if logBackupCR.Status.LogTruncateUntil != "" &&
		logBackup.TruncateUntil != logBackupCR.Status.LogTruncateUntil {
		logBackup.TruncateUntil = logBackupCR.Status.LogTruncateUntil
		updated = true
		if err := s.updateLogBackupTruncateUntil(ctx, logBackup.ID, logBackupCR.Status.LogTruncateUntil); err != nil {
			return err
		}
	}
	truncateStatus, ok := logBackupCR.Status.LogSubCommandStatuses[cloudv1.LogTruncateCommand]
	if ok {
		if err := s.updateLogBackupTruncateStatus(ctx, logBackup.ID, &truncateStatus); err != nil {
			return err
		}
	}
	if logBackupCR.Status.LogCheckpointTs != "" && logBackup.CheckpointTS != logBackupCR.Status.LogCheckpointTs {
		logBackup.CheckpointTS = logBackupCR.Status.LogCheckpointTs
		updated = true
	}
	if status == entity.LogBackupStatusFailed && logBackup.ExtraInfo != nil && logBackup.ExtraInfo.On3LayerArch {
		cluster, err := s.getCluster(ctx, logBackup.ClusterID)
		if err != nil {
			logger.Warn("get cluster for failed log backup error", zap.Error(err))
		} else {
			rawCR, rawErr := s.backupRegionalFacade.RawGetBackupCR(
				ctx,
				cluster.GetCloudProvider(), // NOTE(ideascf): fix that previous code mistakenly used infraProviderType here.
				cluster.GetRegion(),
				uint64(logBackup.TenantID),
				uint64(logBackup.ProjectID),
				logBackup.ClusterID.Uint64(),
				logBackup.CRName(),
			)
			if rawErr != nil {
				logger.Warn("get raw backup cr for failed log backup error", zap.Error(rawErr))
			} else if errMsg := backuperr.ExtractFailedErrorFromBackupCR(rawCR); errMsg != "" && logBackup.ErrMsg != errMsg {
				logBackup.ErrMsg = errMsg
				updated = true
			}
		}
	}

	if updated {
		if err := s.logBackupRepo.Save(ctx, logBackup); err != nil {
			logger.Error("save log backup", zap.Error(err))
			return err
		}
	}
	return nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) convertLogBackupStatus(
	logBackupObj *cloudv1.Backup,
) entity.LogBackupStatus {
	switch logBackupObj.Status.Phase {
	case cloudv1.BackupStopped, cloudv1.BackupComplete:
		return entity.LogBackupStatusStopped
	case cloudv1.BackupRunning:
		return entity.LogBackupStatusRunning
	case cloudv1.BackupFailed:
		return entity.LogBackupStatusFailed
	default:
		return ""
	}
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) updateLogBackupTruncateUntil(
	ctx context.Context, logBackupID entity.ID, truncateUntil string,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	truncates, err := s.logBackupTruncateRepo.List(ctx,
		func(qs entity.LogBackupTruncateQuerySet) entity.LogBackupTruncateQuerySet {
			return qs.DeletedAtIsNull().
				StatusNotIn(entity.LogBackupTruncateStatusSuccess, entity.LogBackupTruncateStatusFailed).
				LogBackupIDEq(logBackupID).
				OrderAscByCreatedAt()
		})
	if err != nil {
		logger.Error("list log backup truncate", zap.Error(err))
		return err
	}

	now := time.Now()
	for _, each := range truncates {
		truncate := &each
		compareResult, err := util.CompareTSOStr(truncateUntil, truncate.Until)
		if err != nil {
			logger.Warn("compare truncate until error",
				zap.String("truncate until", truncateUntil),
				zap.String("truncate until of log backup", truncate.Until),
				zap.Uint64("log backup truncate id", uint64(truncate.ID)), zap.Error(err))
			continue
		}
		if compareResult < 0 {
			continue
		}

		if err := truncate.Finish(ctx, now, entity.LogBackupTruncateStatusSuccess, ""); err != nil {
			return err
		}
		if err := s.logBackupTruncateRepo.Save(ctx, truncate); err != nil {
			logger.Error("save log backup truncate", zap.Error(err))
			return err
		}
	}
	return nil
}

//nolint:funlen
func (s *logBackupLongTermSynchronizationFromInfraImpl) updateLogBackupTruncateStatus(
	ctx context.Context, logBackupID entity.ID, truncateStatus *cloudv1.LogSubCommandStatus,
) bizerr.BizError {
	logger := log.GetLogger(ctx).With(zap.Uint64("log backup id", logBackupID.Uint64()))
	logger.Info("update log backup truncate status from infra")
	if truncateStatus.Phase != cloudv1.BackupFailed {
		return nil
	}

	var (
		failedAt   time.Time
		failedWith string
	)
	for _, each := range truncateStatus.Conditions {
		if each.Type != cloudv1.BackupFailed || each.Status != corev1.ConditionTrue {
			continue
		}
		failedAt = each.LastTransitionTime.Time
		failedWith = each.Message
		break
	}
	if failedAt.IsZero() {
		logger.Warn("truncate status is failed but no failed at time found in conditions")
		return nil
	}

	truncates, err := s.logBackupTruncateRepo.List(ctx,
		func(qs entity.LogBackupTruncateQuerySet) entity.LogBackupTruncateQuerySet {
			return qs.DeletedAtIsNull().
				StatusIn(entity.LogBackupTruncateStatusRunning).
				LogBackupIDEq(logBackupID).
				OrderAscByCreatedAt()
		})
	if err != nil {
		logger.Error("list running log backup truncates", zap.Error(err))
		return err
	}

	for _, each := range truncates {
		truncate := &each
		if truncate.RunningAt == nil {
			logger.Warn("running truncate without runningAt",
				zap.Uint64("log backup truncate id", uint64(truncate.ID)),
			)
			continue
		}

		if truncate.RunningAt.Before(failedAt) {
			logger.Warn("mark running truncate as failed since infra truncate status is failed",
				zap.String("runningAt", truncate.RunningAt.String()),
				zap.String("failedAt", failedAt.String()),
				zap.Uint64("log backup truncate id", uint64(truncate.ID)),
			)
			if err := truncate.Finish(ctx, failedAt, entity.LogBackupTruncateStatusFailed, failedWith); err != nil {
				return err
			}
			if err := s.logBackupTruncateRepo.Save(ctx, truncate); err != nil {
				logger.Error("save log backup truncate", zap.Error(err))
				return err
			}
		}
	}
	return nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) TruncateAllPendingLogBackups(
	ctx context.Context,
) bizerr.BizError {
	ctx = log.WithLogID(ctx)
	logger := log.GetLogger(ctx)
	truncates, err := s.listLogBackupTruncate(ctx, logger)
	if err != nil {
		return err
	}
	if len(truncates) == 0 {
		logger.Info("no pending log backup truncate")
		return nil
	}

	pendingTruncates := pie.Of(truncates).Filter(func(truncate entity.LogBackupTruncate) bool {
		return truncate.Status == entity.LogBackupTruncateStatusPending
	}).Result

	if len(pendingTruncates) == 0 {
		logger.Info("no pending log backup truncate")
		return nil
	}

	runningLogBackupTruncatingSet := make(map[entity.ID]bool)
	pie.Of(truncates).Filter(func(truncate entity.LogBackupTruncate) bool {
		return truncate.Status == entity.LogBackupTruncateStatusRunning
	}).Each(func(truncate entity.LogBackupTruncate) {
		runningLogBackupTruncatingSet[truncate.LogBackupID] = true
	})

	for _, truncate := range pendingTruncates {
		if runningLogBackupTruncatingSet[truncate.LogBackupID] {
			logger.Info("there is another truncate running, waiting for it finished",
				zap.Uint64("truncate id", uint64(truncate.ID)))
			continue
		}
		canTruncate, err := s.checkWhetherCanTruncate(ctx, &truncate)
		if err != nil {
			logger.Error("check whether can truncate error",
				zap.Uint64("truncate id", uint64(truncate.ID)),
				zap.Error(err))
			continue
		}
		if !canTruncate {
			continue
		}

		if err = s.ExecTruncate(ctx, &truncate); err != nil {
			logger.Error("execute truncate error",
				zap.Uint64("truncate id", uint64(truncate.ID)),
				zap.Error(err))
			continue
		}
		runningLogBackupTruncatingSet[truncate.LogBackupID] = true
	}
	return nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) listLogBackupTruncate(
	ctx context.Context, logger *zap.Logger,
) ([]entity.LogBackupTruncate, bizerr.BizError) {
	truncates, err := s.logBackupTruncateRepo.List(ctx,
		func(qs entity.LogBackupTruncateQuerySet) entity.LogBackupTruncateQuerySet {
			// try to execute the latest truncate firstly, so that we can skip earlier pending truncate
			// and reduce truncate execution times
			return qs.DeletedAtIsNull().
				StatusIn(entity.LogBackupTruncateStatusPending, entity.LogBackupTruncateStatusRunning).
				OrderDescByCreatedAt()
		})
	if err != nil {
		logger.Error("list log backup truncate", zap.Error(err))
		return nil, err
	}
	return truncates, nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) checkWhetherCanTruncate(
	ctx context.Context, truncate *entity.LogBackupTruncate,
) (bool, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	restores, err := s.restoreRepo.List(ctx, func(qs entity.RestoreQuerySet) entity.RestoreQuerySet {
		return qs.LogBackupIDEq(truncate.LogBackupID).
			StatusIn(entity.RestoreStatusRUNNING, entity.RestoreStatusPENDING).
			PreloadBackup().
			OrderAscByCreatedAt()
	})
	if err != nil {
		logger.Error("list active restore by log backup error", zap.Error(err))
		return false, err
	}

	for _, restore := range restores {
		if restore.RestoredTS == 0 {
			continue
		}

		compareResult, err := util.CompareTSOStr(truncate.Until, restore.Backup.StartTimeStamp)
		if err != nil {
			logger.Error("compare truncate until and backup start timestamp error,"+
				"restore id: %d, backup id: %d, start ts: %s, truncate until: %s",
				zap.Any("restore id", restore.ID),
				zap.Any("backup id", restore.BackupID),
				zap.Any("start ts", restore.Backup.StartTimeStamp),
				zap.Any("truncate until", truncate.Until),
				zap.Error(err))
			return false, bizerr_wrapper.NewInternalErr("%s", err)
		}
		if compareResult > 0 {
			return false, nil
		}
	}
	return true, nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) ExecTruncate(
	ctx context.Context, truncate *entity.LogBackupTruncate,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	if truncate.Status != entity.LogBackupTruncateStatusPending {
		logger.Warn("truncate status isn't pending")
		return bizerr_wrapper.NewInternalErr("truncate status isn't pending")
	}

	logBackups, err := s.logBackupRepo.List(ctx, func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
		return qs.IDEq(truncate.LogBackupID).PreloadBackupSchedule()
	})
	if err != nil {
		logger.Error("load log backup error", zap.Error(err))
		return err
	}
	if len(logBackups) == 0 {
		logger.Warn("log backup not found")
		return bizerr_wrapper.NewInternalErr("log backup not found")
	}

	logBackup := logBackups[0]
	if logBackup.BackupSchedule == nil {
		logger.Debug("log backup has no backup schedule")
		return nil
	}

	cluster, err := s.getCluster(ctx, logBackup.ClusterID)
	if err != nil {
		logger.Error("can't get cluster region", zap.Error(err))
		return err
	}
	region := cluster.GetRegion()
	cloudProvider := cluster.GetCloudProvider()
	if err := s.transProvider.Transaction(ctx, func(ctx context.Context) error {
		truncate.Status = entity.LogBackupTruncateStatusRunning
		truncate.RunningAt = pointer.To(time.Now())
		if err := s.logBackupTruncateRepo.Save(ctx, truncate); err != nil {
			logger.Error("save log backup truncate", zap.Error(err))
			return err
		}

		if err := s.backupRegionalFacade.TruncateLogBackup(ctx, cloudProvider, region,
			uint64(logBackup.TenantID), uint64(logBackup.ProjectID),
			logBackup.ClusterID.Uint64(), logBackup.CRName(), truncate.Until); err != nil {
			logger.Error("truncate log backup in infra api", zap.Error(err))
			return err
		}

		return nil
	}); err != nil {
		return bizerr_wrapper.NewInternalErr("%s", err)
	}
	return nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) getCluster(
	ctx context.Context, clusterId entity.ID,
) (*mgmt.Cluster, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	cluster, err := s.mgmtCli.GetAnyDedicatedClusterBasicByID(ctx, uint64(clusterId))
	if err != nil {
		logger.Error("fail to get cluster", zap.Error(err))
		return nil, err
	}
	return cluster, nil
}

func (s *logBackupLongTermSynchronizationFromInfraImpl) getClusterRegion(
	ctx context.Context, clusterId entity.ID,
) (string, bizerr.BizError) {
	cluster, err := s.getCluster(ctx, clusterId)
	if err != nil {
		return "", err
	}
	return cluster.GetRegion(), nil
}