package entity

import (
	"context"
	"fmt"
	"net/url"
	"path"
	"strconv"
	"strings"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service/app/utils"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

type TSParser func(string) (int64, error)

const zeroStr = "0"

func (b *Backup) CRName() string {
	if b.OldBackupName != "" {
		return b.OldBackupName
	}
	if b.ExtraInfo != nil && b.ExtraInfo.BackupCRNameVersion == "v1" {
		return utils.UniqueIdentifierLte63(fmt.Sprintf("backup-%d", b.ID))
	}
	return fmt.Sprintf("%s-%d", strings.ToLower(b.BackupSchedule.Name), b.ID)
}

func (b *Backup) InFinalStatus() bool {
	for _, s := range FinalBackupStatuses {
		if b.Status == s {
			return true
		}
	}
	return false
}

func (b *Backup) BackupTS(ctx context.Context, tsParser TSParser) *int64 {
	logger := log.GetLogger(ctx)
	if b.StartTimeStamp == "" {
		return nil
	}
	ts, err := tsParser(b.StartTimeStamp)
	if err != nil {
		logger.Warn("parse backup ts failed", zap.String("ts", b.StartTimeStamp), zap.Error(err))
		return nil
	}
	return pointer.To(ts)
}

func (b *Backup) IsFullClusterBackup() bool {
	return len(b.TableFilter) == 0
}

func (b *Backup) Name() string {
	return b.BackupSchedule.BackupName(b.CreatedAt)
}

func (b *Backup) DeleteBackup(deletedAt time.Time) bizerr.BizError {
	if b.Status == BackupStatusRunning || b.Status == BackupStatusPending {
		b.Status = BackupStatusTerminated
	}
	b.DeletedAt = &deletedAt

	return nil
}

func (s *BackupSchedule) BackupName(backupCreatedAt time.Time) string {
	name := s.Name
	if s.Type == BackupTypeAuto {
		name = fmt.Sprintf("%s-%s", s.Name, backupCreatedAt.UTC().Format("2006-01-02t15-04-05"))
	}
	return name
}

func (m *BackupSetting) QueryMasterBucket() *BackupSettingBucket {
	if m.Bucket == nil {
		return nil
	}
	buckets := pie.Of(m.Bucket.Buckets).
		Filter(func(bucket BackupSettingBucket) bool {
			return bucket.Master
		}).Result
	if len(buckets) != 1 {
		return nil
	}
	return &buckets[0]
}

func (m *BackupSetting) MasterRegion() (string, bool) {
	masterBucket := m.QueryMasterBucket()
	if masterBucket == nil || len(masterBucket.Regions) == 0 {
		return "", false
	}
	masterRegion := masterBucket.Regions[0]
	return masterRegion, true
}

func (m *BackupSetting) IsEligibleForBackup() bool {
	return (m.Status == BackupSettingStatusNormal ||
		m.Status == BackupSettingStatusEditingPitr) &&
		m.QueryMasterBucket() != nil && m.QueryMasterBucket().BucketId != 0
}

func (m *BackupSetting) QuerySecondaryBucket() *BackupSettingBucket {
	if m.Bucket == nil {
		return nil
	}
	if !m.Bucket.DualRegionEnable {
		return nil
	}
	buckets := pie.Of(m.Bucket.Buckets).
		Filter(func(bucket BackupSettingBucket) bool {
			return !bucket.Master
		}).Result
	if len(buckets) != 1 {
		return nil
	}
	return &buckets[0]
}

func (m *BackupSetting) GenAWSBucketPrefix() *string {
	masterBucket := m.QueryMasterBucket()
	if masterBucket == nil {
		return nil
	}

	secondaryBucket := m.QuerySecondaryBucket()
	if secondaryBucket != nil {
		prefix := fmt.Sprintf("cluster-%d-sr(%s)-dr(%s)",
			m.ClusterID, masterBucket.Regions[0], secondaryBucket.Regions[0])
		return &prefix
	}
	prefix := fmt.Sprintf("cluster-%d-sr(%s)",
		m.ClusterID, masterBucket.Regions[0])
	return &prefix
}

func (m *BackupSetting) GenGCPBucketPrefix() *string {
	prefix := strconv.FormatUint(m.ClusterID.Uint64(), 10)
	return &prefix
}

func (m *BackupSetting) GenAzureBucketPrefix() *string {
	prefix := strconv.FormatUint(m.ClusterID.Uint64(), 10)
	return &prefix
}

func (m *BackupSetting) RefreshBucketPrefix() error {
	var prefix *string
	switch m.InfraProviderType {
	case InfraProviderTypeAWS, InfraProviderTypeEKS:
		prefix = m.GenAWSBucketPrefix()
	case InfraProviderTypeGCP, InfraProviderTypeGKE:
		prefix = m.GenGCPBucketPrefix()
	case InfraProviderTypeAzure, InfraProviderTypeAKS:
		prefix = m.GenAzureBucketPrefix()
	}

	if prefix == nil {
		return fmt.Errorf("gen bucket prefix failed")
	}
	m.Bucket.Prefix = prefix
	return nil
}

func (m *BackupSetting) IsPiTREnabled() bool {
	if m.PiTR != nil {
		return m.PiTR.Enable
	}
	return false
}

func (m *BackupSetting) SetPiTREnable(val bool) {
	if m.PiTR == nil {
		m.PiTR = &PiTRSetting{}
	}
	m.PiTR.Enable = val
}

func (m *BackupSetting) IsAutoBackupDisabled() bool {
	if m.AutoBackup != nil {
		return m.AutoBackup.Disable
	}
	return false
}

func (m *BackupSetting) SetAutoBackupDisable(val bool) {
	if m.AutoBackup == nil {
		m.AutoBackup = &AutoBackupSetting{}
	}
	m.AutoBackup.Disable = val
}

func (m *BackupSetting) GetBucketName(env string, prefix string, regions []string) string {
	// set default prefix if not provided
	if prefix == "" {
		if m.Bucket != nil && m.Bucket.DualRegionEnable {
			prefix = "dbaas-dr"
		} else {
			prefix = "dbaas"
		}
	}

	suffix := ""
	// Appends account string to avoid bucket name collisions only if Dual Region Backup is enabled
	if m.Bucket != nil && m.Bucket.DualRegionEnable && m.CloudAccountID != nil {
		suffix = *m.CloudAccountID
	}
	return utils.UniqueIdentifierLte63V2(m.formatProjectBucketName(env, regions, prefix, suffix))
}

func (m *BackupSetting) formatProjectBucketName(
	env string, regions []string, bucketPrefix, bucketSuffix string,
) string {
	var projectName string
	if env == "nightly" {
		projectName = fmt.Sprintf("nightly-ms-%d", m.ProjectID)
	} else {
		projectName = fmt.Sprintf("%d", m.ProjectID)
	}
	name := fmt.Sprintf("%s-%s-%s-%s", bucketPrefix, env, strings.Join(regions, "-"), projectName)
	if bucketSuffix != "" {
		name = fmt.Sprintf("%s-%s", name, bucketSuffix)
	}
	return name
}

// GetBucketByRegion returns the BucketId of the BackupSettingBucket that covers the
// requested region.
//
// Returns 0, false when:
//   - region is empty, or
//   - no configured bucket covers the requested region.
func (m *BackupSetting) GetBucketByRegion(region string) (uint64, bool) {
	if m.Bucket == nil || region == "" {
		return 0, false
	}
	for _, b := range m.Bucket.Buckets {
		for _, r := range b.Regions {
			if r == region {
				return b.BucketId, true
			}
		}
	}
	return 0, false
}

// GetBucketByRegion returns the bucket in BucketSettingSnapshot that covers the
// requested region.
//
// Returns BackupSettingBucket{}, false when:
//   - the backup has no BucketSettingSnapshot, or
//   - no bucket in the snapshot covers the requested region.
func (b *Backup) GetBucketByRegion(region string) (BackupSettingBucket, bool) {
	if b.BucketSettingSnapshot == nil {
		return BackupSettingBucket{}, false
	}
	for _, bucket := range b.BucketSettingSnapshot.Buckets {
		for _, r := range bucket.Regions {
			if r == region {
				return bucket, true
			}
		}
	}
	return BackupSettingBucket{}, false
}

func (b *LogBackup) CRName() string {
	if b.ExtraInfo != nil && b.ExtraInfo.BackupCRNameVersion == "v1" {
		return fmt.Sprintf("log-%d", b.ID)
	}
	return fmt.Sprintf("log-%s-%d", strings.ToLower(b.BackupSchedule.Name), b.ID)
}

// GenerateDownstreamID returns the deterministic downstream CR name for this compact backup.
// The ID is derived from the stable DB auto-increment ID, ensuring idempotency across retries.
func (cb *CompactBackup) GenerateDownstreamID() string {
	return fmt.Sprintf("compact-%d", cb.ID)
}

// MarkDeleting transitions the CompactBackup to the deleting state.
// Returns false if already in deleted or deleting state (idempotent no-op, caller should skip).
// Returns true if the transition was applied and the entity should be saved.
func (cb *CompactBackup) MarkDeleting() bool {
	if cb.Status == CompactBackupStatusDeleted || cb.Status == CompactBackupStatusDeleting {
		return false
	}
	cb.Status = CompactBackupStatusDeleting
	return true
}

func (cb *CompactBackup) DeletingOrDeleted() bool {
	return cb.Status == CompactBackupStatusDeleting || cb.Status == CompactBackupStatusDeleted
}

// FinishDelete transitions the CompactBackup to the deleted state.
func (cb *CompactBackup) FinishDelete(deletedAt time.Time) (changed bool, _ error) {
	switch cb.Status {
	case CompactBackupStatusDeleted:
		return false, nil
	case CompactBackupStatusDeleting:
		cb.Status = CompactBackupStatusDeleted
		cb.DeletedAt = &deletedAt
		return true, nil
	default:
		return false, fmt.Errorf("cannot finish delete compact backup that is not in deleting status, current status: %s", cb.Status)
	}
}

func (b *LogBackup) IsDeleted() bool {
	if b.DeletedAt != nil {
		return true
	}
	if b.Status == LogBackupStatusStopped {
		if b.CheckpointTS == "" {
			return true
		}
		if b.TruncateUntil >= b.CheckpointTS {
			return true
		}
	}
	return false
}

func (b *LogBackup) CheckpointTSEmpty() bool {
	return b.CheckpointTS == "" || b.CheckpointTS == zeroStr
}

type PiTRTimeRange struct {
	StartTS int64
	EndTS   int64

	Backup    Backup
	LogBackup LogBackup
}

// GenPiTRTimeRanges generates available PiTR time ranges per snapshot backup.
// -----------------|-----------------|-----------------|-----------------|-----------------|
// //	        log-startTs                                             log-endTs
// //backup1-TS	                  backup2-TS        backup3-TS                          backup4-TS
// the result is
// {startTs: backup3-TS+1, endTs: log-endTs, backup: backup3, logBackup: log}
// {startTs: backup2-TS+1, endTs: log-endTs, backup: backup2, logBackup: log}
func (b *LogBackup) GenPiTRTimeRanges(
	ctx context.Context, backups []Backup,
) []*PiTRTimeRange {
	logger := log.GetLogger(ctx)
	startTs, endTs, err := b.GenLogBackupValidRange()
	if err != nil || startTs == nil || endTs == nil {
		logger.Warn("failed to generate log backup valid range", zap.Error(err))
		return nil
	}

	var timeRanges []*PiTRTimeRange
	endTsTemp := *endTs
	for _, backup := range backups {
		if backup.Status != BackupStatusSuccess {
			continue
		}
		if !backup.IsFullClusterBackup() {
			continue
		}
		if backup.BucketID != b.BucketID {
			logger.Info("bucket id not match",
				zap.Uint64("backup_id", backup.ID.Uint64()),
				zap.Uint64("backup_bucket_id", backup.BucketID.Uint64()),
				zap.Uint64("log_backup_id", b.ID.Uint64()),
				zap.Uint64("log_backup_bucket_id", b.BucketID.Uint64()),
			)
			continue
		}
		backupTS, err := utils.ParseTSFromTSString(backup.StartTimeStamp)
		if err != nil {
			logger.Warn("failed to parse backup start ts",
				zap.Uint64("backup_id", backup.ID.Uint64()),
				zap.String("start_ts", backup.StartTimeStamp),
				zap.Error(err),
			)
			continue
		}
		if backupTS >= *endTs {
			continue
		}
		// Too old snapshot backup, skip remaining backups because backups are ordered by StartTimeStamp in descending order.
		// When enabling PiTR, the TS (snapshot backup start TSO - 1 second) will be used as the startTS of logbackup. So, the PiT trigger
		// snapshot backup's start TSO should always greater than the log backup's checkpoint TSO.
		if backupTS < *startTs {
			break
		}
		timeRanges = append(timeRanges, &PiTRTimeRange{
			// the real PiTR start time should start with a full backup ts
			// when we convert tso to second level timestamp, we lose some degree of precision
			// for example, the tso is 2022-10-24 12:23:43.986, we get the start ts is 2022-10-24 12:23:43
			// but the restore ts should greater than the backup ts, so we should change the start ts to 2022-10-24 12:23:44
			// for the end ts, lose some degree of precision is right, because restore ts should less than the checkpoint ts
			StartTS:   backupTS + 1,
			EndTS:     endTsTemp,
			Backup:    backup,
			LogBackup: *b,
		})
	}
	logger.Info("generate time range success",
		zap.Uint64("log_backup_id", b.ID.Uint64()),
		zap.Int("range_count", len(timeRanges)),
	)
	return timeRanges
}

// GenPiTRTimeRange generates the available PiTR time ranges based on the log backup and full backup information.
// backups should be ordered by StartTimeStamp in descending order
func (b *LogBackup) GenPiTRTimeRange(
	ctx context.Context, backups []Backup,
) (_, _ int64, found bool) {
	if b == nil {
		return 0, 0, false
	}
	timeRanges := b.GenPiTRTimeRanges(ctx, backups)
	if len(timeRanges) == 0 {
		return 0, 0, false
	}
	latest := timeRanges[0]
	oldest := timeRanges[len(timeRanges)-1]
	return oldest.StartTS, latest.EndTS, true
}

// GenLogBackupValidRange generate available time range for restore, return startTS and checkpointTS
func (b *LogBackup) GenLogBackupValidRange() (*int64, *int64, error) {
	if b == nil || b.StartTS == nil || b.CheckpointTSEmpty() {
		return nil, nil, nil
	}
	var (
		startTS, checkpointTS, truncateUntil int64
		err                                  error
	)
	startTS, checkpointTS, truncateUntil, err = b.ParseStartAndCheckPointAndTruncateUntilTS()
	if err != nil {
		return nil, nil, err
	}

	logBackupStartTS := startTS
	if truncateUntil != 0 && truncateUntil > logBackupStartTS {
		logBackupStartTS = truncateUntil
	}
	if logBackupStartTS >= checkpointTS {
		return nil, nil, nil
	}
	return &logBackupStartTS, &checkpointTS, nil
}

func (b *LogBackup) ParseStartAndCheckPointAndTruncateUntilTS() (int64, int64, int64, error) {
	if b == nil || b.StartTS == nil || b.CheckpointTSEmpty() {
		return 0, 0, 0, fmt.Errorf("log backup is nil or missing required fields")
	}
	startTS, err := utils.ParseTSFromTSString(*b.StartTS)
	if err != nil {
		return 0, 0, 0, fmt.Errorf("parse start ts %s: %w", *b.StartTS, err)
	}
	checkpointTS, err := utils.ParseTSFromTSString(b.CheckpointTS)
	if err != nil {
		return 0, 0, 0, fmt.Errorf("parse checkpoint ts %s: %w", b.CheckpointTS, err)
	}
	if b.TruncateUntil != "" {
		truncateUntilTS, err := utils.ParseTSFromTSString(b.TruncateUntil)
		if err != nil {
			return 0, 0, 0, fmt.Errorf("parse truncate until %s: %w", b.TruncateUntil, err)
		}
		return startTS, checkpointTS, truncateUntilTS, nil
	}
	return startTS, checkpointTS, 0, nil
}

// GetAvailableRetentionPeriod returns the available retention period in days
func (b *BackupSchedule) GetAvailableRetentionPeriod() uint64 {
	return b.RetentionPeriod
}

func (b *BackupSchedule) ExpiredTSOByTSO(latestTSO uint64) uint64 {
	retentionPeriod := b.GetAvailableRetentionPeriod()
	latestTS := utils.TransToTS(latestTSO)
	expiredTS := time.Unix(latestTS, 0).Add(-time.Hour * 24 * time.Duration(retentionPeriod)).Unix()
	return utils.TransToTSO(expiredTS)
}

func (b *BackupSchedule) ExpiredTSO(latestTime time.Time) uint64 {
	retentionPeriod := b.GetAvailableRetentionPeriod()
	expiredTime := latestTime.Add(-time.Hour * 24 * time.Duration(retentionPeriod))
	return utils.GoTimeToTSO(expiredTime)
}

func (e *BackupExport) IsCancellable() bool {
	return e.Status == BackupExportStatusPending || e.Status == BackupExportStatusRunning
}

// InFinalStatus reports whether the export has reached a terminal status
// (failed/success/canceled). Mirrors Backup.InFinalStatus so callers don't
// re-derive finality as "not pending/running", which would wrongly treat an
// unexpected/zero status as terminal.
func (e *BackupExport) InFinalStatus() bool {
	for _, s := range FinalBackupExportStatuses {
		if e.Status == s {
			return true
		}
	}
	return false
}

func (e *BackupExport) Run(reason string) error {
	if e.Status != BackupExportStatusPending {
		return fmt.Errorf("backup export is not pending, current status: %s", e.Status)
	}

	e.Status = BackupExportStatusRunning
	e.StatusMsg = reason
	return nil
}

func (e *BackupExport) RunExportingJob(jobID string) error {
	if e.Status != BackupExportStatusRunning {
		return fmt.Errorf("backup export is not running, current status: %s", e.Status)
	}
	// Update backup export with job ID and status
	e.JobID = jobID
	e.StatusMsg = "Transfer job created and running"
	return nil
}

func (e *BackupExport) Fail(now time.Time, reason string) error {
	if e.Status != BackupExportStatusRunning {
		return fmt.Errorf("backup export is not running, current status: %s", e.Status)
	}

	e.Status = BackupExportStatusFailed
	e.StatusMsg = reason
	e.FinishedAt = &now
	return nil
}

func (e *BackupExport) Success(now time.Time, reason string) error {
	if e.Status != BackupExportStatusRunning {
		return fmt.Errorf("backup export is not running, current status: %s", e.Status)
	}

	e.Status = BackupExportStatusSuccess
	e.StatusMsg = reason
	e.FinishedAt = &now
	return nil
}

func (e *BackupExport) Cancel(now time.Time, reason string) error {
	if !e.IsCancellable() {
		return fmt.Errorf("backup export is not cancellable, current status: %s", e.Status)
	}

	e.Status = BackupExportStatusCanceled
	e.StatusMsg = reason
	e.FinishedAt = &now
	return nil
}

// FinalTargetURI returns the final target URI which with a suffix of cluster id and backup export name
// appended to the customer provided target path
func (e *BackupExport) FinalTargetURI() string {
	targetURI := strings.TrimSpace(e.Target.TargetURI)
	u, err := url.Parse(targetURI)
	if err != nil || u.Scheme == "" || u.Host == "" {
		return targetURI
	}

	basePath := strings.TrimPrefix(u.Path, "/")
	retPath := path.Join(
		basePath,
		fmt.Sprintf("%d", e.ClusterID),
		fmt.Sprintf("ExportBackup-%d", e.BackupID),
	)
	if !strings.HasSuffix(retPath, "/") {
		retPath += "/"
	}
	u.Path = "/" + retPath
	u.RawPath = ""
	return u.String()
}

// JobName returns the name of the job for this backup export
func (e *BackupExport) JobName() string {
	return fmt.Sprintf("transferJobs/backup-export-%d", e.ID)
}

// JobDescription returns a description for the transfer job
func (e *BackupExport) JobDescription(sourceURI string) string {
	return fmt.Sprintf("Transfer from %s to %s", sourceURI, e.FinalTargetURI())
}

func GetCron(backupCycle BackupCycle, backupTime string, weekdays []int32) (string, error) {
	timeFormat := strings.Split(backupTime, ":")
	_, err := utils.ParseHour(timeFormat[0])
	if err != nil {
		return "", err
	}
	_, err = utils.ParseMinute(timeFormat[1])
	if err != nil {
		return "", err
	}

	if BackupCycleWeekly == backupCycle {
		return fmt.Sprintf("0 %s %s ? * %s *", timeFormat[1], timeFormat[0], utils.FormatWeekDays(weekdays)), nil
	} else {
		return fmt.Sprintf("0 %s %s 1/1 * ? *", timeFormat[1], timeFormat[0]), nil
	}
}

func (t *LogBackupTruncate) Finish(
	ctx context.Context, finishedAt time.Time, status LogBackupTruncateStatus, msg string,
) bizerr.BizError {
	if status != LogBackupTruncateStatusSuccess &&
		status != LogBackupTruncateStatusFailed {
		return bizerr_wrapper.NewInternalErr("invalid log backup truncate status to finish, status=%s", status)
	}

	logger := log.GetLogger(ctx)
	if t.DeletedAt != nil {
		return bizerr_wrapper.NewInternalErr("log backup truncate already deleted, deletedAt=%v", t.DeletedAt)
	}
	if t.Status == LogBackupTruncateStatusSuccess ||
		t.Status == LogBackupTruncateStatusFailed {
		return bizerr_wrapper.NewInternalErr("log backup truncate already finished, status=%s", t.Status)
	}

	logger.Info("finish log backup truncate",
		zap.Uint64("log backup truncate id", uint64(t.ID)),
		zap.Uint64("log backup id", uint64(t.LogBackupID)),
		zap.String("status", string(status)),
	)
	t.Status = status
	t.ErrMsg = msg
	// t.FinishedAt = &finishedAt // TODO(ideascf): fix it in next PR. The previous implementation doesn't update it.
	return nil
}