package backup

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	vmbackup "github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/actions"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/backupnames"
	"github.com/avast/retry-go/v4"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/robfig/cron/v3"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/vmbackupmanager/pkg/config"
	"github.com/tidbcloud/observability/vmbackupmanager/pkg/storage"
	"go.uber.org/zap"
)

// BackupType defines the type of backup
type BackupType string

const (
	// IncrementalBackup represents an incremental backup
	IncrementalBackup BackupType = "incremental"
	// FullBackup represents a full backup
	FullBackup BackupType = "full"
	// MigratedBackup represents a migrated backup
	MigratedBackup BackupType = "migrated"
	// IncrementalBackupPath is the path of the incremental backup
	IncrementalBackupPath = "latest"
	// FullBackupPath is the path of the full backup
	FullBackupPath = "full"
	// BackupMgrMetadataPath is the path of the backup metadata file
	BackupMgrMetadataPath = "backupmgr_metadata"
	// MaxRetryAttempts is the maximum number of retry attempts for failed backups
	MaxRetryAttempts = 2
	// RetryDelay is the delay between retry attempts
	RetryDelay = 30 * time.Second
	// backupRetentionCheckCron is the cron expression for checking backup retention.
	backupRetentionCheckCron = "0 0 * * *" // every day at midnight
)

var (
	// ErrQueueFull is returned when the backup queue is full
	ErrQueueFull = fmt.Errorf("queue is full")

	backupStatus = prometheus.NewDesc(
		"vmbackupmanager_backup_status",
		"The status of the backup task scheduled by vmbackupmanager",
		[]string{"id", "destination", "type", "created_at", "completed_at", "retention", "status"},
		nil,
	)
)

// BackupStatus defines the status of a backup operation
type BackupStatus string

const (
	// StatusPending indicates a backup is pending
	StatusPending BackupStatus = "pending"
	// StatusRunning indicates a backup is in progress
	StatusRunning BackupStatus = "running"
	// StatusCompleted indicates a backup completed successfully
	StatusCompleted BackupStatus = "completed"
	// StatusFailed indicates a backup failed
	StatusFailed BackupStatus = "failed"
	// StatusDeleting indicates a backup is being deleted
	StatusDeleting BackupStatus = "deleting"
)

// BackupMetadata stores information about a single backup
type BackupMetadata struct {
	// Type is the type of backup.
	Type BackupType `json:"type"`
	// Id is the id of the backup.
	Id string `json:"id"`
	// DestPath is the path of the backup in the storage.
	DestPath string `json:"dest_path"`
	// CreatedAt is the time when the backup was created.
	CreatedAt string `json:"created_at,omitempty"`
	// ScheduledAt is the time when the backup was scheduled.
	ScheduledAt string `json:"scheduled_at,omitempty"`
	// StartedAt is the time when the backup started.
	// It might be modified to the actual start time after vmbackup is completed.
	StartedAt string `json:"started_at,omitempty"`
	// CompletedAt is the time when the backup completed.
	// It might be modified to the actual completion time after vmbackup is completed.
	CompletedAt string `json:"completed_at,omitempty"`
	// DeletedAt is the time when the backup was deleted.
	DeletedAt string `json:"deleted_at,omitempty"`
	// Status is the status of the backup.
	Status BackupStatus `json:"status"`
	// Error is the error message if the backup failed.
	Error string `json:"error,omitempty"`
	// DataRetention indicates the data retention period of the backup.
	// It is set to the retention period of the VM.
	DataRetention string `json:"data_retention"`
}

// Manager handles backup operations and scheduling
type Manager struct {
	ctx           context.Context
	cancel        context.CancelFunc
	config        *config.Config
	scheduler     *cron.Cron
	backups       sync.Map                       // contains entries of backupName -> *BackupMetadata
	wg            sync.WaitGroup                 // WaitGroup for running backups, used for graceful shutdown
	runningBackup atomic.Pointer[BackupMetadata] // Metadata of the currently running backup
	taskQueue     chan *BackupMetadata           // Channel for queuing backup tasks
	store         storage.Interface              // Unified storage client for all operations
}

var _ prometheus.Collector = &Manager{}

// NewManager creates a new backup manager
func NewManager(cfg *config.Config) (*Manager, error) {
	s, err := storage.NewStorage(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create storage client: %w", err)
	}
	ctx, cancel := context.WithCancel(context.Background())

	m := &Manager{
		ctx:       ctx,
		cancel:    cancel,
		config:    cfg,
		scheduler: cron.New(cron.WithLogger(&cronLogger{logger: log.L()})),
		backups:   sync.Map{},
		taskQueue: make(chan *BackupMetadata, 1),
		store:     s,
	}

	return m, nil
}

type cronLogger struct {
	logger *zap.Logger
}

func (l *cronLogger) Info(msg string, keysAndValues ...interface{}) {
	fields := make([]zap.Field, 0, len(keysAndValues)/2)
	for i := 0; i < len(keysAndValues); i += 2 {
		fields = append(fields, zap.Any(keysAndValues[i].(string), keysAndValues[i+1]))
	}
	l.logger.Info(msg, fields...)
}

func (l *cronLogger) Error(err error, msg string, keysAndValues ...interface{}) {
	fields := []zap.Field{zap.Error(err)}
	for i := 0; i < len(keysAndValues); i += 2 {
		fields = append(fields, zap.Any(keysAndValues[i].(string), keysAndValues[i+1]))
	}
	l.logger.Error(msg, fields...)
}

// loadMetadata loads backup metadata by scanning the directory structure in object storage
func (m *Manager) loadMetadata() error {
	// Reset backups map
	m.backups = sync.Map{}

	// List metadata files from storage
	files, err := m.store.ListFiles(context.Background(), BackupMgrMetadataPath)
	if err != nil {
		return fmt.Errorf("failed to list metadata directory: %w", err)
	}

	// Load each metadata file
	for _, file := range files {
		// Get metadata file content
		data, err := m.store.ReadFile(context.Background(), file)
		if err != nil {
			log.Error("Failed to read metadata file", zap.String("file", file), zap.Error(err))
			continue
		}

		// Parse metadata
		var metadata BackupMetadata
		if err := json.Unmarshal(data, &metadata); err != nil {
			log.Error("Failed to parse metadata file", zap.String("file", file), zap.Error(err))
			continue
		}

		// Add to backups map
		m.backups.Store(metadata.Id, &metadata)
		log.Info("Loaded backup metadata", zap.String("name", metadata.Id), zap.String("status", string(metadata.Status)))

		if metadata.Status != StatusCompleted && metadata.Type == FullBackup {
			// re-run the full backup if it's not completed, it's possible that the backup failed due to some reason.
			log.Info("Resuming failed full backup", zap.String("name", metadata.Id))
			m.ensureEnqueueBackup(metadata.Type, metadata.Id)
			continue
		}
	}

	return nil
}

// saveMetadata saves backup metadata to storage
func (m *Manager) saveMetadata(metadata *BackupMetadata) error {
	// Serialize metadata to JSON
	data, err := json.Marshal(metadata)
	if err != nil {
		return fmt.Errorf("failed to serialize metadata: %w", err)
	}

	// Save to storage using destPath as the file name
	metadataFilePath := fmt.Sprintf("%s/%s", BackupMgrMetadataPath, metadata.Id)
	err = m.store.CreateFile(context.Background(), metadataFilePath, data)
	if err != nil {
		return fmt.Errorf("failed to save metadata: %w", err)
	}

	// Update backups map
	m.backups.Store(metadata.Id, metadata)
	return nil
}

func (m *Manager) DeleteBackup(metadata *BackupMetadata) error {
	// delete the backup from the storage
	if err := m.store.DeleteDir(m.ctx, metadata.DestPath); err != nil {
		log.Error("Failed to delete backup", zap.String("name", metadata.Id), zap.Error(err))
		return err
	}
	// delete the backup metadata from the storage
	metadataFilePath := fmt.Sprintf("%s/%s", BackupMgrMetadataPath, metadata.Id)
	if err := m.store.DeleteFile(m.ctx, metadataFilePath); err != nil {
		log.Error("Failed to delete backup metadata", zap.String("name", metadata.Id), zap.String("file", metadataFilePath), zap.Error(err))
		return err
	}

	// Remove from backups map
	m.backups.Delete(metadata.Id)
	return nil
}

// Start starts the backup manager
func (m *Manager) Start() error {
	// Start the worker to process backup tasks
	go m.processBackupTasks()

	// migrate needed
	migrateNeeded, err := m.migrateNeeded()
	if err != nil {
		return fmt.Errorf("failed to check if migrate is needed: %w", err)
	}
	if migrateNeeded {
		log.Info("Migrating backup to the new path structure containing a lastest and full folder")
		m.ensureEnqueueBackup(MigratedBackup, fmt.Sprintf("%s-%s", "migrated", time.Now().Format(time.DateOnly)))
	}

	// Load existing metadata from the directory structure
	if err := m.loadMetadata(); err != nil {
		return fmt.Errorf("failed to load existing backup metadata: %w", err)
	}

	// start the scheduler
	// 1. Schedule incremental backups
	if _, err := m.scheduler.AddFunc(m.config.IncrSchedule, func() {
		log.Info("Scheduling incremental backup", zap.String("schedule", m.config.IncrSchedule))
		// skip scheduling if another incremental backup is already running.
		// In order to avoid concurrency issues with the metadata.
		if running := m.runningBackup.Load(); running != nil && running.Type == IncrementalBackup {
			log.Info("Skipping incremental backup scheduling", zap.String("reason", "another incremental backup is already running"))
			return
		}
		// use IncrementalBackupPath as name for incremental backup.
		if _, err := m.EnqueueBackup(IncrementalBackup, IncrementalBackupPath); err != nil {
			log.Info("Skipping incremental backup scheduling", zap.Error(err))
		}
	}); err != nil {
		return fmt.Errorf("failed to schedule incremental backups: %w", err)
	}

	// 2. Schedule full backups if enabled
	if !m.config.DisableFullBackup {
		if m.config.FullSchedule == "" {
			// default path: schedule retention-based full backups.
			schedule := &RetentionSchedule{
				RetentionPeriod:         m.config.VMRetentionPeriod,
				RetentionTimeZoneOffset: m.config.VMRetentionTimeZoneOffset,
			}
			log.Info("Enabling full backup schedule based on retention", zap.Any("schedule", schedule))
			m.scheduler.Schedule(schedule, cron.FuncJob(m.enqueueFullBackup))
		} else {
			log.Info("Enabling full backup schedule based on custom cron expression", zap.String("schedule", m.config.FullSchedule))
			if _, err := m.scheduler.AddFunc(m.config.FullSchedule, m.enqueueFullBackup); err != nil {
				return fmt.Errorf("failed to schedule full backups: %w", err)
			}
		}
	}

	// 3. Schedule backup GC task based on retention.
	if !m.config.DisableBackupRetention {
		_, err := m.scheduler.AddFunc(backupRetentionCheckCron, m.backupGC)
		if err != nil {
			return fmt.Errorf("failed to schedule backup retention check: %w", err)
		}
	}

	m.scheduler.Start()
	return nil
}

func (m *Manager) enqueueFullBackup() {
	backupID := fmt.Sprintf("full-%s", time.Now().Format(time.DateOnly))
	log.Info("Ensuring enqueue full backup", zap.String("name", backupID))
	m.ensureEnqueueBackup(FullBackup, backupID)
}

func (m *Manager) ensureEnqueueBackup(backupType BackupType, name string) {
	err := retry.Do(
		func() error {
			if _, err := m.EnqueueBackup(backupType, name); err != nil {
				return err
			}
			return nil
		},
		retry.Context(m.ctx),
		retry.Delay(RetryDelay),
	)
	if err != nil {
		log.Error("Failed to enqueue full backup", zap.Error(err))
	}
}

// EnqueueBackup adds a backup task to the queue
func (m *Manager) EnqueueBackup(backupType BackupType, name string) (*BackupMetadata, error) {
	if m.isQueueFull() {
		return nil, ErrQueueFull
	}

	var destPath string
	var backup *BackupMetadata
	if backupType == FullBackup {
		// For full backups, use a subdirectory
		destPath = fmt.Sprintf("%s/%s", FullBackupPath, name)
	} else {
		// For incremental backups, use the latest directory.
		destPath = IncrementalBackupPath
	}
	backup = &BackupMetadata{
		Id:        name,
		DestPath:  destPath,
		CreatedAt: time.Now().Format(time.RFC3339),
		Status:    StatusRunning,
		Type:      backupType,
	}

	err := m.saveMetadata(backup)
	if err != nil {
		return nil, fmt.Errorf("failed to save metadata: %w", err)
	}
	m.taskQueue <- backup
	log.Info("Enqueued backup task", zap.String("type", string(backupType)), zap.String("name", name))
	return backup, nil
}

func (m *Manager) isQueueFull() bool {
	return len(m.taskQueue) >= cap(m.taskQueue)
}

// ListBackups returns metadata for all backups
func (m *Manager) ListBackups() []*BackupMetadata {
	backups := make([]*BackupMetadata, 0)
	m.backups.Range(func(key, value interface{}) bool {
		backups = append(backups, value.(*BackupMetadata))
		return true
	})

	return backups
}

// processBackupTasks processes backup tasks from the queue
func (m *Manager) processBackupTasks() {
	for task := range m.taskQueue {
		// Skip if another backup is already running.
		// this is unlikely to happen since we run backups sequentially, but check it here to be safe.
		if m.IsBackupRunning() {
			log.Info("Skipping backup", zap.String("name", task.Id), zap.String("reason", "another backup is already running"))
			continue
		}

		err := m.executeBackup(task)
		if err != nil {
			log.Error("Failed to execute backup", zap.String("type", string(task.Type)), zap.Error(err))
		}
		log.Info("Backup completed", zap.String("name", task.Id))
	}
}

// executeBackup executes a backup operation of the specified type
func (m *Manager) executeBackup(backup *BackupMetadata) error {
	backup.ScheduledAt = time.Now().Format(time.RFC3339)
	backup.StartedAt = time.Now().Format(time.RFC3339)
	backup.Status = StatusRunning

	// Save initial metadata
	if err := m.saveMetadata(backup); err != nil {
		log.Warn("Failed to save initial metadata for backup", zap.String("name", backup.Id), zap.Error(err))
	}

	// Set as running backup
	m.runningBackup.Store(backup)
	m.wg.Add(1)

	// Make sure to reset flags when done
	defer func() {
		m.wg.Done()
		m.runningBackup.Store(nil)
	}()

	// Execute the backup task with retry
	var err error = retry.Do(
		func() error {

			return m.executeVMBackup(backup.Type, m.store.GetFullPath(backup.DestPath))
		},
		retry.Context(m.ctx),
		retry.Attempts(MaxRetryAttempts),
		retry.Delay(RetryDelay),
		retry.DelayType(retry.FixedDelay),
		retry.OnRetry(func(n uint, err error) {
			log.Error("Failed to execute backup", zap.String("name", backup.Id), zap.Uint("attempt", n+1), zap.Error(err))
		}),
	)

	completedAt := time.Now()
	backup.CompletedAt = completedAt.Format(time.RFC3339)
	if err != nil {
		backup.Status = StatusFailed
		backup.Error = err.Error()
		log.Error("Backup failed", zap.String("name", backup.Id), zap.Error(err))
	} else {
		backup.Status = StatusCompleted
		// update the metadata based on the vmbackup metadata file if it's present.
		vmMeta := &vmbackup.BackupMetadata{}
		if data, err := m.store.ReadFile(m.ctx, fmt.Sprintf("%s/%s", backup.DestPath, backupnames.BackupMetadataFilename)); err != nil {
			log.Warn("Failed to read vmbackup metadata", zap.String("name", backup.Id), zap.Error(err))
		} else {
			if err := json.Unmarshal(data, vmMeta); err != nil {
				log.Warn("Failed to unmarshal backup metadata", zap.String("name", backup.Id), zap.Error(err))
			} else {
				backup.CompletedAt = vmMeta.CompletedAt
				backup.StartedAt = vmMeta.CreatedAt
			}
		}
		log.Info("Backup completed successfully", zap.String("name", backup.Id))
	}

	// Save final metadata
	backup.DataRetention = m.config.VMRetentionPeriod.String()
	if err := m.saveMetadata(backup); err != nil {
		log.Warn("Failed to save final metadata for backup", zap.String("name", backup.Id), zap.Error(err))
	}

	return err
}

// executeVMBackup performs the actual backup operation by invoking vmbackup command.
func (m *Manager) executeVMBackup(backupType BackupType, destPath string) error {
	// clean possible leftover snapshots before running vmbackup.
	if err := m.cleanSnapshots(); err != nil {
		// ignore error, it's not critical.
		log.Warn("Failed to clean snapshots", zap.Error(err))
	}

	// Prepare vmbackup arguments
	concurrency := m.config.Concurrency
	args := []string{
		"-storageDataPath=" + m.config.StorageDataPath,
		"-dst=" + destPath,
		"-maxBytesPerSecond=" + m.config.MaxBytesPerSecond,
		"-loggerFormat=json",
		"-loggerJSONFields=ts:time,msg:message",
	}

	if m.config.Vendor == constants.AliCloud {
		args = append(args, "-customS3Endpoint="+m.config.CustomS3Endpoint)
		args = append(args, "-s3ForcePathStyle="+strconv.FormatBool(m.config.S3ForcePathStyle))
	}

	switch backupType {
	case FullBackup:
		args = append(args, "-origin="+m.store.GetFullPath(IncrementalBackupPath))
		args = append(args, "-snapshot.createURL="+m.config.SnapshotCreateURL)
		concurrency = m.config.FullBackupConcurrency
	case MigratedBackup:
		args = append(args, "-origin="+m.config.Destination)
		args = append(args, "-snapshot.createURL="+m.config.SnapshotCreateURL)
		concurrency = m.config.FullBackupConcurrency
	case IncrementalBackup:
		args = append(args, "-snapshot.createURL="+m.config.SnapshotCreateURL)
	default:
		return fmt.Errorf("invalid backup type: %s", backupType)
	}

	args = append(args, fmt.Sprintf("-concurrency=%d", concurrency))

	log.Info("Executing vmbackup command", zap.String("type", string(backupType)), zap.Strings("args", args))
	cmd := exec.CommandContext(m.ctx, "vmbackup", args...)
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr

	if err := cmd.Run(); err != nil {
		return fmt.Errorf("vmbackup failed: %w", err)
	}
	return nil
}

// cleanSnapshots deletes all snapshots.
func (m *Manager) cleanSnapshots() error {
	if m.config.SnapshotCreateURL == "" {
		return nil
	}
	deleteURL := strings.Replace(m.config.SnapshotCreateURL, "create", "delete_all", 1)
	//nolint:gosec
	resp, err := http.Post(deleteURL, "application/json", nil)
	if err != nil {
		return fmt.Errorf("failed to clean snapshot: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, err := io.ReadAll(resp.Body)
		if err != nil {
			return fmt.Errorf("failed to read response body: %w", err)
		}
		return fmt.Errorf("failed to clean snapshot, status: %d, body: %s", resp.StatusCode, string(body))
	}

	return nil
}

// GetBackup returns metadata for a specific backup
func (m *Manager) GetBackup(id string) (*BackupMetadata, bool) {
	value, exists := m.backups.Load(id)
	if !exists {
		return nil, false
	}
	return value.(*BackupMetadata), exists
}

// IsBackupRunning returns whether any backup is currently running
func (m *Manager) IsBackupRunning() bool {
	return m.runningBackup.Load() != nil
}

// GetRunningBackupIfExists atomically checks if a backup is running and returns it
// This eliminates race conditions between IsBackupRunning() and GetRunningBackup() calls
func (m *Manager) GetRunningBackupIfExists() (*BackupMetadata, bool) {
	running := m.runningBackup.Load()
	if running != nil {
		return running, true
	}
	return nil, false
}

func (m *Manager) RegisterMetrics(reg prometheus.Registerer) {
	reg.MustRegister(m)
}

// Shutdown stops the scheduler and waits for running backups to complete
func (m *Manager) Shutdown(ctx context.Context) {
	m.cancel()
	m.scheduler.Stop()
	done := make(chan struct{})
	go func() {
		m.wg.Wait()
		close(done)
	}()

	select {
	case <-done:
		log.Info("All backups completed")
	case <-ctx.Done():
		log.Warn("Timeout waiting for backups to complete")
	}
}

func (m *Manager) Describe(ch chan<- *prometheus.Desc) {
	ch <- backupStatus
}

func (m *Manager) Collect(ch chan<- prometheus.Metric) {
	m.backups.Range(func(key, value interface{}) bool {
		backup := value.(*BackupMetadata)
		ch <- prometheus.MustNewConstMetric(backupStatus, prometheus.GaugeValue, 1, backup.Id, backup.DestPath, string(backup.Type), backup.CreatedAt, backup.CompletedAt, backup.DataRetention, string(backup.Status))
		return true
	})
}

// backupGC checks if the backup retention is met.
// it will delete the backups that are older than the backup retention.
func (m *Manager) backupGC() {
	log.Info("Checking backup retention")

	if m.config.BackupRetention <= 0 {
		log.Info("Backup retention check skipped", zap.String("reason", "retention period not set"))
		return
	}

	cutoffTime := time.Now().Add(-m.config.BackupRetention)
	log.Info("Calculated retention cutoff time", zap.Time("cutoffTime", cutoffTime))

	// Find backups that are older than retention period
	var backupsToDelete []string
	m.backups.Range(func(key, value interface{}) bool {
		backup := value.(*BackupMetadata)
		if exists, err := m.store.Exists(m.ctx, backup.DestPath); err != nil {
			log.Error("Failed to check if backup exists", zap.String("name", key.(string)), zap.Error(err))
			return true
		} else if !exists {
			log.Info("Backup does not exist, removing it from the backups map", zap.String("name", key.(string)))
			m.backups.Delete(key)
			return true
		}

		// Skip non-completed backups and incremental backups
		if backup.Status != StatusCompleted || backup.Type != FullBackup {
			return true
		}

		completedTime, err := time.Parse(time.RFC3339, backup.CompletedAt)
		if err != nil {
			log.Error("Failed to parse backup completion time", zap.String("name", key.(string)), zap.String("time", backup.CompletedAt), zap.Error(err))
			return true
		}

		if completedTime.Before(cutoffTime) {
			backupsToDelete = append(backupsToDelete, key.(string))
			log.Info("Marking backup for deletion", zap.String("name", key.(string)), zap.Time("completedAt", completedTime))
		}
		return true
	})

	// Delete backups
	for _, id := range backupsToDelete {
		log.Info("Deleting expired backup", zap.String("name", id))
		value, exists := m.backups.Load(id)
		if !exists {
			continue
		}
		backup := value.(*BackupMetadata)
		if err := m.DeleteBackup(backup); err != nil {
			log.Error("Failed to delete backup", zap.String("name", id), zap.Error(err))
			continue
		}
		log.Info("Successfully deleted expired backup", zap.String("name", id))
	}

	log.Info("Backup retention check completed", zap.Int("deletedCount", len(backupsToDelete)))
}