opsxsql/backend/internal/ghost/ghost.go

package ghost

import (
	"bytes"
	"context"
	"fmt"
	"os/exec"
	"strings"
	"sync"
	"time"
)

// GhOstTask defines a gh-ost execution task
type GhOstTask struct {
	Host           string `json:"host"`
	User           string `json:"user"`
	Password       string `json:"password"`
	Database       string `json:"database"`
	Table          string `json:"table"`
	Alter          string `json:"alter"`              // e.g. "ADD COLUMN status INT NOT NULL DEFAULT 0"
	ChunkSize      int    `json:"chunk_size,omitempty"`
	MaxLoad        string `json:"max_load,omitempty"`
	CriticalLoad   string `json:"critical_load,omitempty"`
	MaxLagMillis   int    `json:"max_lag_millis,omitempty"`
	AllowOnMaster  bool   `json:"allow_on_master"`
	DryRun         bool   `json:"dry_run"`
	Execute        bool   `json:"execute"`
	ExactRowCount  bool   `json:"exact_row_count"`
	PostponeCutOver bool  `json:"postpone_cut_over"`
}

// GhOstResult holds the execution result
type GhOstResult struct {
	Success  bool   `json:"success"`
	Output   string `json:"output"`
	Error    string `json:"error"`
	Command  string `json:"command"`
	Duration string `json:"duration"`
	TaskID   string `json:"task_id"`
}

// ProgressCallback is called with stdout lines for real-time progress
type ProgressCallback func(taskID string, line string)

// TaskManager manages async gh-ost tasks
type TaskManager struct {
	mu     sync.RWMutex
	tasks  map[string]*GhOstResult
	status map[string]string // running, completed, failed
}

var Manager = &TaskManager{
	tasks:  make(map[string]*GhOstResult),
	status: make(map[string]string),
}

// RunGhOst executes gh-ost synchronously
func RunGhOst(ctx context.Context, task GhOstTask, ghOstPath string) (*GhOstResult, error) {
	start := time.Now()

	args := []string{
		"--host=" + task.Host,
		"--user=" + task.User,
		"--password=" + task.Password,
		"--database=" + task.Database,
		"--table=" + task.Table,
		"--alter=" + task.Alter,
		"--verbose",
	}

	if task.ChunkSize > 0 {
		args = append(args, fmt.Sprintf("--chunk-size=%d", task.ChunkSize))
	}
	if task.MaxLoad != "" {
		args = append(args, "--max-load="+task.MaxLoad)
	}
	if task.CriticalLoad != "" {
		args = append(args, "--critical-load="+task.CriticalLoad)
	}
	if task.MaxLagMillis > 0 {
		args = append(args, fmt.Sprintf("--max-lag-millis=%d", task.MaxLagMillis))
	}
	if task.AllowOnMaster {
		args = append(args, "--allow-on-master")
	}
	if task.ExactRowCount {
		args = append(args, "--exact-rowcount")
	}
	if task.PostponeCutOver {
		args = append(args, "--postpone-cut-over-flag-file=/tmp/gh-ost-cutover.flag")
	}

	// Dry-Run vs Execute
	if task.DryRun || !task.Execute {
		args = append(args, "--dry-run")
	} else {
		args = append(args, "--execute")
	}

	cmd := exec.CommandContext(ctx, ghOstPath, args...)
	var stdout, stderr bytes.Buffer
	cmd.Stdout = &stdout
	cmd.Stderr = &stderr

	err := cmd.Run()

	result := &GhOstResult{
		Command:  "gh-ost " + strings.Join(args, " "),
		Duration: time.Since(start).String(),
		Output:   stdout.String(),
	}

	if err != nil {
		result.Success = false
		result.Error = stderr.String() + "\n" + err.Error()
		return result, err
	}

	result.Success = true
	return result, nil
}

// RunGhOstAsync runs gh-ost in background and tracks status
func RunGhOstAsync(task GhOstTask, ghOstPath string, cb ProgressCallback) string {
	taskID := fmt.Sprintf("ghost-%d", time.Now().UnixNano())

	Manager.mu.Lock()
	Manager.status[taskID] = "running"
	Manager.tasks[taskID] = &GhOstResult{TaskID: taskID}
	Manager.mu.Unlock()

	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour)
		defer cancel()

		start := time.Now()

		args := buildArgs(task, ghOstPath)

		cmd := exec.CommandContext(ctx, ghOstPath, args...)
		var stdout, stderr bytes.Buffer
		cmd.Stdout = &stdout
		cmd.Stderr = &stderr

		err := cmd.Run()

		Manager.mu.Lock()
		defer Manager.mu.Unlock()

		result := &GhOstResult{
			TaskID:   taskID,
			Command:  "gh-ost " + strings.Join(args, " "),
			Duration: time.Since(start).String(),
			Output:   stdout.String(),
			Success:  err == nil,
		}

		if err != nil {
			result.Error = stderr.String() + "\n" + err.Error()
			Manager.status[taskID] = "failed"
		} else {
			Manager.status[taskID] = "completed"
		}

		Manager.tasks[taskID] = result

		if cb != nil {
			cb(taskID, result.Output)
		}
	}()

	return taskID
}

// GetTaskStatus returns the status of an async task
func (m *TaskManager) GetTaskStatus(taskID string) (string, *GhOstResult) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.status[taskID], m.tasks[taskID]
}

// ListTasks returns all tracked tasks
func (m *TaskManager) ListTasks() map[string]string {
	m.mu.RLock()
	defer m.mu.RUnlock()
	result := make(map[string]string)
	for k, v := range m.status {
		result[k] = v
	}
	return result
}

func buildArgs(task GhOstTask, ghOstPath string) []string {
	args := []string{
		"--host=" + task.Host,
		"--user=" + task.User,
		"--password=" + task.Password,
		"--database=" + task.Database,
		"--table=" + task.Table,
		"--alter=" + task.Alter,
		"--verbose",
	}

	if task.ChunkSize > 0 {
		args = append(args, fmt.Sprintf("--chunk-size=%d", task.ChunkSize))
	}
	if task.MaxLoad != "" {
		args = append(args, "--max-load="+task.MaxLoad)
	}
	if task.CriticalLoad != "" {
		args = append(args, "--critical-load="+task.CriticalLoad)
	}
	if task.MaxLagMillis > 0 {
		args = append(args, fmt.Sprintf("--max-lag-millis=%d", task.MaxLagMillis))
	}
	if task.AllowOnMaster {
		args = append(args, "--allow-on-master")
	}
	if task.ExactRowCount {
		args = append(args, "--exact-rowcount")
	}
	if task.PostponeCutOver {
		args = append(args, "--postpone-cut-over-flag-file=/tmp/gh-ost-cutover.flag")
	}

	if task.DryRun || !task.Execute {
		args = append(args, "--dry-run")
	} else {
		args = append(args, "--execute")
	}

	return args
}