dbbackup/internal/engine/clone.go

package engine

import (
	"archive/tar"
	"compress/gzip"
	"context"
	"database/sql"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"regexp"
	"strconv"
	"strings"
	"time"

	"dbbackup/internal/logger"
	"dbbackup/internal/metadata"
	"dbbackup/internal/security"
)

// CloneEngine implements BackupEngine using MySQL Clone Plugin (8.0.17+)
type CloneEngine struct {
	db     *sql.DB
	config *CloneConfig
	log    logger.Logger
}

// CloneConfig contains Clone Plugin configuration
type CloneConfig struct {
	// Connection
	Host     string
	Port     int
	User     string
	Password string

	// Clone mode
	Mode string // "local" or "remote"

	// Local clone options
	DataDirectory string // Target directory for clone

	// Remote clone options
	Remote *RemoteCloneConfig

	// Post-clone handling
	Compress       bool
	CompressFormat string // "gzip", "zstd", "lz4"
	CompressLevel  int

	// Performance
	MaxBandwidth string // e.g., "100M" for 100 MB/s
	Threads      int

	// Progress
	ProgressInterval time.Duration
}

// RemoteCloneConfig contains settings for remote clone
type RemoteCloneConfig struct {
	Host     string
	Port     int
	User     string
	Password string
}

// CloneProgress represents clone progress from performance_schema
type CloneProgress struct {
	Stage        string // "DROP DATA", "FILE COPY", "PAGE COPY", "REDO COPY", "FILE SYNC", "RESTART", "RECOVERY"
	State        string // "Not Started", "In Progress", "Completed"
	BeginTime    time.Time
	EndTime      time.Time
	Threads      int
	Estimate     int64 // Estimated bytes
	Data         int64 // Bytes transferred
	Network      int64 // Network bytes (remote clone)
	DataSpeed    int64 // Bytes/sec
	NetworkSpeed int64 // Network bytes/sec
}

// CloneStatus represents final clone status from performance_schema
type CloneStatus struct {
	ID           int64
	State        string
	BeginTime    time.Time
	EndTime      time.Time
	Source       string // Source host for remote clone
	Destination  string
	ErrorNo      int
	ErrorMessage string
	BinlogFile   string
	BinlogPos    int64
	GTIDExecuted string
}

// NewCloneEngine creates a new Clone Plugin engine
func NewCloneEngine(db *sql.DB, config *CloneConfig, log logger.Logger) *CloneEngine {
	if config == nil {
		config = &CloneConfig{
			Mode:             "local",
			Compress:         true,
			CompressFormat:   "gzip",
			CompressLevel:    6,
			ProgressInterval: time.Second,
		}
	}
	return &CloneEngine{
		db:     db,
		config: config,
		log:    log,
	}
}

// Name returns the engine name
func (e *CloneEngine) Name() string {
	return "clone"
}

// Description returns a human-readable description
func (e *CloneEngine) Description() string {
	return "MySQL Clone Plugin (physical backup, MySQL 8.0.17+)"
}

// CheckAvailability verifies Clone Plugin is available
func (e *CloneEngine) CheckAvailability(ctx context.Context) (*AvailabilityResult, error) {
	result := &AvailabilityResult{
		Info: make(map[string]string),
	}

	if e.db == nil {
		result.Available = false
		result.Reason = "database connection not established"
		return result, nil
	}

	// Check MySQL version
	var version string
	if err := e.db.QueryRowContext(ctx, "SELECT VERSION()").Scan(&version); err != nil {
		result.Available = false
		result.Reason = fmt.Sprintf("failed to get version: %v", err)
		return result, nil
	}
	result.Info["version"] = version

	// Extract numeric version
	re := regexp.MustCompile(`(\d+\.\d+\.\d+)`)
	matches := re.FindStringSubmatch(version)
	if len(matches) < 2 {
		result.Available = false
		result.Reason = "could not parse version"
		return result, nil
	}
	versionNum := matches[1]
	result.Info["version_number"] = versionNum

	// Check if version >= 8.0.17
	if !versionAtLeast(versionNum, "8.0.17") {
		result.Available = false
		result.Reason = fmt.Sprintf("MySQL Clone requires 8.0.17+, got %s", versionNum)
		return result, nil
	}

	// Check if clone plugin is installed
	var pluginName, pluginStatus string
	err := e.db.QueryRowContext(ctx, `
		SELECT PLUGIN_NAME, PLUGIN_STATUS 
		FROM INFORMATION_SCHEMA.PLUGINS 
		WHERE PLUGIN_NAME = 'clone'
	`).Scan(&pluginName, &pluginStatus)

	if err == sql.ErrNoRows {
		// Try to install the plugin
		e.log.Info("Clone plugin not installed, attempting to install...")
		_, installErr := e.db.ExecContext(ctx, "INSTALL PLUGIN clone SONAME 'mysql_clone.so'")
		if installErr != nil {
			result.Available = false
			result.Reason = fmt.Sprintf("clone plugin not installed and failed to install: %v", installErr)
			return result, nil
		}
		result.Warnings = append(result.Warnings, "Clone plugin was installed automatically")
		pluginStatus = "ACTIVE"
	} else if err != nil {
		result.Available = false
		result.Reason = fmt.Sprintf("failed to check clone plugin: %v", err)
		return result, nil
	}

	result.Info["plugin_status"] = pluginStatus

	if pluginStatus != "ACTIVE" {
		result.Available = false
		result.Reason = fmt.Sprintf("clone plugin is %s (needs ACTIVE)", pluginStatus)
		return result, nil
	}

	// Check required privileges
	var hasBackupAdmin bool
	rows, err := e.db.QueryContext(ctx, "SHOW GRANTS")
	if err == nil {
		defer rows.Close()
		for rows.Next() {
			var grant string
			rows.Scan(&grant)
			if strings.Contains(strings.ToUpper(grant), "BACKUP_ADMIN") ||
				strings.Contains(strings.ToUpper(grant), "ALL PRIVILEGES") {
				hasBackupAdmin = true
				break
			}
		}
	}

	if !hasBackupAdmin {
		result.Warnings = append(result.Warnings, "BACKUP_ADMIN privilege recommended for clone operations")
	}

	result.Available = true
	result.Info["mode"] = e.config.Mode
	return result, nil
}

// Backup performs a clone backup
func (e *CloneEngine) Backup(ctx context.Context, opts *BackupOptions) (*BackupResult, error) {
	startTime := time.Now()

	e.log.Info("Starting Clone Plugin backup",
		"database", opts.Database,
		"mode", e.config.Mode)

	// Validate prerequisites
	warnings, err := e.validatePrerequisites(ctx)
	if err != nil {
		return nil, fmt.Errorf("prerequisites validation failed: %w", err)
	}
	for _, w := range warnings {
		e.log.Warn(w)
	}

	// Determine output directory
	cloneDir := e.config.DataDirectory
	if cloneDir == "" {
		timestamp := time.Now().Format("20060102_150405")
		cloneDir = filepath.Join(opts.OutputDir, fmt.Sprintf("clone_%s_%s", opts.Database, timestamp))
	}

	// Ensure parent directory exists
	if err := os.MkdirAll(filepath.Dir(cloneDir), 0755); err != nil {
		return nil, fmt.Errorf("failed to create parent directory: %w", err)
	}

	// Ensure clone directory doesn't exist
	if _, err := os.Stat(cloneDir); err == nil {
		return nil, fmt.Errorf("clone directory already exists: %s", cloneDir)
	}

	// Start progress monitoring in background
	progressCtx, cancelProgress := context.WithCancel(ctx)
	progressCh := make(chan CloneProgress, 10)
	go e.monitorProgress(progressCtx, progressCh, opts.ProgressFunc)

	// Perform clone
	var cloneErr error
	if e.config.Mode == "remote" && e.config.Remote != nil {
		cloneErr = e.remoteClone(ctx, cloneDir)
	} else {
		cloneErr = e.localClone(ctx, cloneDir)
	}

	// Stop progress monitoring
	cancelProgress()
	close(progressCh)

	if cloneErr != nil {
		// Cleanup on failure
		os.RemoveAll(cloneDir)
		return nil, fmt.Errorf("clone failed: %w", cloneErr)
	}

	// Get clone status for binlog position
	status, err := e.getCloneStatus(ctx)
	if err != nil {
		e.log.Warn("Failed to get clone status", "error", err)
	}

	// Calculate clone size
	var cloneSize int64
	filepath.Walk(cloneDir, func(path string, info os.FileInfo, err error) error {
		if err == nil && !info.IsDir() {
			cloneSize += info.Size()
		}
		return nil
	})

	// Output file path
	var finalOutput string
	var files []BackupFile

	// Optionally compress the clone
	if opts.Compress || e.config.Compress {
		e.log.Info("Compressing clone directory...")
		timestamp := time.Now().Format("20060102_150405")
		tarFile := filepath.Join(opts.OutputDir, fmt.Sprintf("clone_%s_%s.tar.gz", opts.Database, timestamp))

		if err := e.compressClone(ctx, cloneDir, tarFile, opts.ProgressFunc); err != nil {
			return nil, fmt.Errorf("failed to compress clone: %w", err)
		}

		// Remove uncompressed clone
		os.RemoveAll(cloneDir)

		// Get compressed file info
		info, _ := os.Stat(tarFile)
		checksum, _ := security.ChecksumFile(tarFile)

		finalOutput = tarFile
		files = append(files, BackupFile{
			Path:     tarFile,
			Size:     info.Size(),
			Checksum: checksum,
		})

		e.log.Info("Clone compressed",
			"output", tarFile,
			"original_size", formatBytes(cloneSize),
			"compressed_size", formatBytes(info.Size()),
			"ratio", fmt.Sprintf("%.1f%%", float64(info.Size())/float64(cloneSize)*100))
	} else {
		finalOutput = cloneDir
		files = append(files, BackupFile{
			Path: cloneDir,
			Size: cloneSize,
		})
	}

	endTime := time.Now()
	lockDuration := time.Duration(0)
	if status != nil && !status.BeginTime.IsZero() && !status.EndTime.IsZero() {
		lockDuration = status.EndTime.Sub(status.BeginTime)
	}

	// Save metadata
	meta := &metadata.BackupMetadata{
		Version:      "3.40.0",
		Timestamp:    startTime,
		Database:     opts.Database,
		DatabaseType: "mysql",
		Host:         e.config.Host,
		Port:         e.config.Port,
		User:         e.config.User,
		BackupFile:   finalOutput,
		SizeBytes:    cloneSize,
		BackupType:   "full",
		ExtraInfo:    make(map[string]string),
	}
	meta.ExtraInfo["backup_engine"] = "clone"

	if status != nil {
		meta.ExtraInfo["binlog_file"] = status.BinlogFile
		meta.ExtraInfo["binlog_position"] = fmt.Sprintf("%d", status.BinlogPos)
		meta.ExtraInfo["gtid_set"] = status.GTIDExecuted
	}

	if opts.Compress || e.config.Compress {
		meta.Compression = "gzip"
	}

	if err := meta.Save(); err != nil {
		e.log.Warn("Failed to save metadata", "error", err)
	}

	result := &BackupResult{
		Engine:       "clone",
		Database:     opts.Database,
		StartTime:    startTime,
		EndTime:      endTime,
		Duration:     endTime.Sub(startTime),
		Files:        files,
		TotalSize:    cloneSize,
		LockDuration: lockDuration,
		Metadata: map[string]string{
			"clone_mode": e.config.Mode,
		},
	}

	if status != nil {
		result.BinlogFile = status.BinlogFile
		result.BinlogPos = status.BinlogPos
		result.GTIDExecuted = status.GTIDExecuted
	}

	e.log.Info("Clone backup completed",
		"database", opts.Database,
		"output", finalOutput,
		"size", formatBytes(cloneSize),
		"duration", result.Duration,
		"binlog", fmt.Sprintf("%s:%d", result.BinlogFile, result.BinlogPos))

	return result, nil
}

// localClone performs a local clone
func (e *CloneEngine) localClone(ctx context.Context, targetDir string) error {
	e.log.Info("Starting local clone", "target", targetDir)

	// Execute CLONE LOCAL DATA DIRECTORY
	query := fmt.Sprintf("CLONE LOCAL DATA DIRECTORY = '%s'", targetDir)
	_, err := e.db.ExecContext(ctx, query)
	if err != nil {
		return fmt.Errorf("CLONE LOCAL failed: %w", err)
	}

	return nil
}

// remoteClone performs a remote clone from another server
func (e *CloneEngine) remoteClone(ctx context.Context, targetDir string) error {
	if e.config.Remote == nil {
		return fmt.Errorf("remote clone config not provided")
	}

	e.log.Info("Starting remote clone",
		"source", fmt.Sprintf("%s:%d", e.config.Remote.Host, e.config.Remote.Port),
		"target", targetDir)

	// Execute CLONE INSTANCE FROM
	query := fmt.Sprintf(
		"CLONE INSTANCE FROM '%s'@'%s':%d IDENTIFIED BY '%s' DATA DIRECTORY = '%s'",
		e.config.Remote.User,
		e.config.Remote.Host,
		e.config.Remote.Port,
		e.config.Remote.Password,
		targetDir,
	)

	_, err := e.db.ExecContext(ctx, query)
	if err != nil {
		return fmt.Errorf("CLONE INSTANCE failed: %w", err)
	}

	return nil
}

// monitorProgress monitors clone progress via performance_schema
func (e *CloneEngine) monitorProgress(ctx context.Context, progressCh chan<- CloneProgress, progressFunc ProgressFunc) {
	ticker := time.NewTicker(e.config.ProgressInterval)
	if e.config.ProgressInterval == 0 {
		ticker = time.NewTicker(time.Second)
	}
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			progress, err := e.queryProgress(ctx)
			if err != nil {
				continue
			}

			// Send to channel
			select {
			case progressCh <- progress:
			default:
			}

			// Call progress function
			if progressFunc != nil {
				percent := float64(0)
				if progress.Estimate > 0 {
					percent = float64(progress.Data) / float64(progress.Estimate) * 100
				}
				progressFunc(&Progress{
					Stage:      progress.Stage,
					Percent:    percent,
					BytesDone:  progress.Data,
					BytesTotal: progress.Estimate,
					Speed:      float64(progress.DataSpeed),
					Message:    fmt.Sprintf("Clone %s: %s/%s", progress.Stage, formatBytes(progress.Data), formatBytes(progress.Estimate)),
				})
			}

			if progress.State == "Completed" {
				return
			}
		}
	}
}

// queryProgress queries clone progress from performance_schema
func (e *CloneEngine) queryProgress(ctx context.Context) (CloneProgress, error) {
	var progress CloneProgress

	query := `
		SELECT 
			COALESCE(STAGE, '') as stage,
			COALESCE(STATE, '') as state,
			COALESCE(BEGIN_TIME, NOW()) as begin_time,
			COALESCE(END_TIME, NOW()) as end_time,
			COALESCE(THREADS, 0) as threads,
			COALESCE(ESTIMATE, 0) as estimate,
			COALESCE(DATA, 0) as data,
			COALESCE(NETWORK, 0) as network,
			COALESCE(DATA_SPEED, 0) as data_speed,
			COALESCE(NETWORK_SPEED, 0) as network_speed
		FROM performance_schema.clone_progress
		ORDER BY ID DESC
		LIMIT 1
	`

	err := e.db.QueryRowContext(ctx, query).Scan(
		&progress.Stage,
		&progress.State,
		&progress.BeginTime,
		&progress.EndTime,
		&progress.Threads,
		&progress.Estimate,
		&progress.Data,
		&progress.Network,
		&progress.DataSpeed,
		&progress.NetworkSpeed,
	)

	if err != nil {
		return progress, err
	}

	return progress, nil
}

// getCloneStatus gets final clone status
func (e *CloneEngine) getCloneStatus(ctx context.Context) (*CloneStatus, error) {
	var status CloneStatus

	query := `
		SELECT 
			COALESCE(ID, 0) as id,
			COALESCE(STATE, '') as state,
			COALESCE(BEGIN_TIME, NOW()) as begin_time,
			COALESCE(END_TIME, NOW()) as end_time,
			COALESCE(SOURCE, '') as source,
			COALESCE(DESTINATION, '') as destination,
			COALESCE(ERROR_NO, 0) as error_no,
			COALESCE(ERROR_MESSAGE, '') as error_message,
			COALESCE(BINLOG_FILE, '') as binlog_file,
			COALESCE(BINLOG_POSITION, 0) as binlog_position,
			COALESCE(GTID_EXECUTED, '') as gtid_executed
		FROM performance_schema.clone_status
		ORDER BY ID DESC
		LIMIT 1
	`

	err := e.db.QueryRowContext(ctx, query).Scan(
		&status.ID,
		&status.State,
		&status.BeginTime,
		&status.EndTime,
		&status.Source,
		&status.Destination,
		&status.ErrorNo,
		&status.ErrorMessage,
		&status.BinlogFile,
		&status.BinlogPos,
		&status.GTIDExecuted,
	)

	if err != nil {
		return nil, err
	}

	return &status, nil
}

// validatePrerequisites checks clone prerequisites
func (e *CloneEngine) validatePrerequisites(ctx context.Context) ([]string, error) {
	var warnings []string

	// Check disk space
	// TODO: Implement disk space check

	// Check that we're not cloning to same directory as source
	var datadir string
	if err := e.db.QueryRowContext(ctx, "SELECT @@datadir").Scan(&datadir); err == nil {
		if e.config.DataDirectory != "" && strings.HasPrefix(e.config.DataDirectory, datadir) {
			return nil, fmt.Errorf("cannot clone to same directory as source data (%s)", datadir)
		}
	}

	return warnings, nil
}

// compressClone compresses clone directory to tar.gz
func (e *CloneEngine) compressClone(ctx context.Context, sourceDir, targetFile string, progressFunc ProgressFunc) error {
	// Create output file
	outFile, err := os.Create(targetFile)
	if err != nil {
		return err
	}
	defer outFile.Close()

	// Create gzip writer
	level := e.config.CompressLevel
	if level == 0 {
		level = gzip.DefaultCompression
	}
	gzWriter, err := gzip.NewWriterLevel(outFile, level)
	if err != nil {
		return err
	}
	defer gzWriter.Close()

	// Create tar writer
	tarWriter := tar.NewWriter(gzWriter)
	defer tarWriter.Close()

	// Walk directory and add files
	return filepath.Walk(sourceDir, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}

		// Check context
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		// Create header
		header, err := tar.FileInfoHeader(info, "")
		if err != nil {
			return err
		}

		// Use relative path
		relPath, err := filepath.Rel(sourceDir, path)
		if err != nil {
			return err
		}
		header.Name = relPath

		// Write header
		if err := tarWriter.WriteHeader(header); err != nil {
			return err
		}

		// Write file content
		if !info.IsDir() {
			file, err := os.Open(path)
			if err != nil {
				return err
			}
			defer file.Close()

			_, err = io.Copy(tarWriter, file)
			if err != nil {
				return err
			}
		}

		return nil
	})
}

// Restore restores from a clone backup
func (e *CloneEngine) Restore(ctx context.Context, opts *RestoreOptions) error {
	e.log.Info("Clone restore", "source", opts.SourcePath, "target", opts.TargetDir)

	// Check if source is compressed
	if strings.HasSuffix(opts.SourcePath, ".tar.gz") {
		// Extract tar.gz
		return e.extractClone(ctx, opts.SourcePath, opts.TargetDir)
	}

	// Source is already a directory - just copy
	return copyDir(opts.SourcePath, opts.TargetDir)
}

// extractClone extracts a compressed clone backup
func (e *CloneEngine) extractClone(ctx context.Context, sourceFile, targetDir string) error {
	// Open source file
	file, err := os.Open(sourceFile)
	if err != nil {
		return err
	}
	defer file.Close()

	// Create gzip reader
	gzReader, err := gzip.NewReader(file)
	if err != nil {
		return err
	}
	defer gzReader.Close()

	// Create tar reader
	tarReader := tar.NewReader(gzReader)

	// Extract files
	for {
		header, err := tarReader.Next()
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}

		// Check context
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		targetPath := filepath.Join(targetDir, header.Name)

		switch header.Typeflag {
		case tar.TypeDir:
			if err := os.MkdirAll(targetPath, 0755); err != nil {
				return err
			}
		case tar.TypeReg:
			if err := os.MkdirAll(filepath.Dir(targetPath), 0755); err != nil {
				return err
			}
			outFile, err := os.Create(targetPath)
			if err != nil {
				return err
			}
			if _, err := io.Copy(outFile, tarReader); err != nil {
				outFile.Close()
				return err
			}
			outFile.Close()
		}
	}

	return nil
}

// SupportsRestore returns true
func (e *CloneEngine) SupportsRestore() bool {
	return true
}

// SupportsIncremental returns false
func (e *CloneEngine) SupportsIncremental() bool {
	return false
}

// SupportsStreaming returns false (clone writes to disk)
func (e *CloneEngine) SupportsStreaming() bool {
	return false
}

// versionAtLeast checks if version is at least minVersion
func versionAtLeast(version, minVersion string) bool {
	vParts := strings.Split(version, ".")
	mParts := strings.Split(minVersion, ".")

	for i := 0; i < len(mParts) && i < len(vParts); i++ {
		v, _ := strconv.Atoi(vParts[i])
		m, _ := strconv.Atoi(mParts[i])
		if v > m {
			return true
		}
		if v < m {
			return false
		}
	}

	return len(vParts) >= len(mParts)
}

// copyDir recursively copies a directory
func copyDir(src, dst string) error {
	return filepath.Walk(src, func(path string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}

		relPath, err := filepath.Rel(src, path)
		if err != nil {
			return err
		}
		targetPath := filepath.Join(dst, relPath)

		if info.IsDir() {
			return os.MkdirAll(targetPath, info.Mode())
		}

		return copyFile(path, targetPath)
	})
}

// copyFile copies a single file
func copyFile(src, dst string) error {
	srcFile, err := os.Open(src)
	if err != nil {
		return err
	}
	defer srcFile.Close()

	dstFile, err := os.Create(dst)
	if err != nil {
		return err
	}
	defer dstFile.Close()

	_, err = io.Copy(dstFile, srcFile)
	return err
}