gitlink-cli/shortcuts/workflow/daemon.go

package workflow

import (
	"fmt"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"strconv"
	"syscall"
	"time"

	"github.com/gitlink-org/gitlink-cli/internal/config"
	"github.com/gitlink-org/gitlink-cli/shortcuts/common"
)

// StartDaemon launches a workflow as a background daemon process.
func StartDaemon(ctx *common.RuntimeContext, wf *WorkflowDef, interval time.Duration, aiMode string) error {
	bin, err := os.Executable()
	if err != nil {
		return fmt.Errorf("cannot find executable: %w", err)
	}

	args := []string{
		"workflow", "+run", "--name", wf.Name,
		"--owner", ctx.Owner, "--repo", ctx.Repo,
		"--format", "json", "--daemon-loop",
		"--interval", interval.String(),
	}
	if aiMode == "ai" {
		args = append(args, "--ai")
	} else if aiMode == "no-ai" {
		args = append(args, "--no-ai")
	}

	cmd := exec.Command(bin, args...)
	cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}

	// Redirect output to log file instead of discarding.
	logPath := daemonLogPath(wf.Name)
	logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
	if err != nil {
		return fmt.Errorf("create log file: %w", err)
	}
	cmd.Stdout = logFile
	cmd.Stderr = logFile
	cmd.Stdin = nil

	if err := cmd.Start(); err != nil {
		logFile.Close()
		return fmt.Errorf("start daemon: %w", err)
	}
	// logFile is owned by child process; it will be closed when child exits.

	pid := cmd.Process.Pid
	if err := savePID(wf.Name, pid); err != nil {
		return fmt.Errorf("save pid: %w", err)
	}

	fmt.Printf("Daemon started for %q (PID: %d)\n", wf.Name, pid)
	fmt.Printf("Log: %s\n", logPath)
	return nil
}

// StopDaemon stops a running workflow daemon by name.
func StopDaemon(name string) error {
	pid, err := readPID(name)
	if err != nil {
		return err
	}

	proc, err := os.FindProcess(pid)
	if err != nil {
		cleanPID(name)
		return fmt.Errorf("daemon %q not running (PID %d not found)", name, pid)
	}

	if err := proc.Signal(os.Interrupt); err != nil {
		// Process might already be dead; clean up pid file anyway.
		cleanPID(name)
		return fmt.Errorf("failed to stop daemon %q: %w", name, err)
	}

	cleanPID(name)
	fmt.Printf("Daemon %q stopped (PID %d)\n", name, pid)
	return nil
}

// StatusDaemon prints the current daemon status for a workflow.
func StatusDaemon(name string) error {
	state, err := LoadState(name)
	if err != nil {
		return fmt.Errorf("load state: %w", err)
	}

	pid, pidErr := readPID(name)
	running := pidErr == nil && processRunning(pid)

	fmt.Printf("工作流: %s\n", name)
	if running {
		fmt.Printf("状态: 运行中 (PID: %d)\n", pid)
	} else {
		fmt.Println("状态: 已停止")
	}
	if state.LastRun != "" {
		t, err := time.Parse(time.RFC3339, state.LastRun)
		if err == nil {
			fmt.Printf("上次运行: %s\n", t.Format("2006-01-02 15:04"))
		} else {
			fmt.Printf("上次运行: %s\n", state.LastRun)
		}
	}
	fmt.Printf("累计运行: %d 次\n", state.TotalRuns)
	fmt.Printf("快照步骤: %d 个\n", len(state.Snapshots))
	fmt.Printf("日志文件: %s\n", daemonLogPath(name))
	return nil
}

// DaemonLoop runs the workflow repeatedly in a loop (used by the daemon subprocess).
func DaemonLoop(ctx *common.RuntimeContext, wf *WorkflowDef, interval time.Duration) error {
	tick := time.NewTicker(interval)
	defer tick.Stop()

	// Run immediately on start (dry-run to establish baseline).
	doDaemonCycle(ctx, wf)

	for {
		select {
		case <-tick.C:
			doDaemonCycle(ctx, wf)
		}
	}
}

func doDaemonCycle(ctx *common.RuntimeContext, wf *WorkflowDef) {
	state, _ := LoadState(wf.Name)

	// Phase 1: cheap dry-run — collect data without AI.
	dryResult, err := Run(ctx, wf, true)
	if err != nil {
		fmt.Fprintf(os.Stderr, "[%s] error: %v\n", time.Now().Format(time.RFC3339), err)
		return
	}

	changed := state.Diff(dryResult.Steps)
	if len(changed) == 0 && state.TotalRuns > 0 {
		// No data changes — save state, skip expensive run.
		state.TotalRuns++
		state.Save()
		return
	}

	fmt.Fprintf(os.Stderr, "[%s] 🔔 检测到变更: %v\n", time.Now().Format(time.RFC3339), changed)

	// Phase 2: full run (AI or rules based on context.AIMode).
	fullResult, err := Run(ctx, wf, false)
	if err != nil {
		fmt.Fprintf(os.Stderr, "[%s] run error: %v\n", time.Now().Format(time.RFC3339), err)
		return
	}

	state.TotalRuns++
	state.Diff(fullResult.Steps)
	state.Save()

	ok, total := 0, len(fullResult.Steps)
	for _, sr := range fullResult.Steps {
		if sr.OK {
			ok++
		}
	}
	fmt.Fprintf(os.Stderr, "[%s] ✅ %d/%d steps OK\n", time.Now().Format(time.RFC3339), ok, total)
}

// daemonLogPath returns the log file path for a workflow daemon.
func daemonLogPath(name string) string {
	return filepath.Join(config.ConfigDir(), fmt.Sprintf("workflow-%s.log", name))
}

// tailDaemonLog reads and optionally follows a daemon log file.
func tailDaemonLog(name string, follow bool) error {
	path := daemonLogPath(name)
	data, err := os.ReadFile(path)
	if err != nil {
		return fmt.Errorf("读取日志文件 %s: %w (daemon 可能尚未启动)", path, err)
	}
	fmt.Print(string(data))

	if !follow {
		return nil
	}

	sig := make(chan os.Signal, 1)
	signal.Notify(sig, os.Interrupt)
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	offset := int64(len(data))
	for {
		select {
		case <-sig:
			return nil
		case <-ticker.C:
			fi, err := os.Stat(path)
			if err != nil {
				continue
			}
			if fi.Size() > offset {
				f, err := os.Open(path)
				if err != nil {
					continue
				}
				f.Seek(offset, 0)
				buf := make([]byte, fi.Size()-offset)
				n, _ := f.Read(buf)
				if n > 0 {
					fmt.Print(string(buf[:n]))
				}
				offset = fi.Size()
				f.Close()
			}
		}
	}
}

// installSystemdUnit generates a systemd service unit file for a workflow daemon.
func installSystemdUnit(ctx *common.RuntimeContext, wf *WorkflowDef, interval, aiMode string) error {
	if err := ctx.ResolveOwnerRepo(); err != nil {
		return fmt.Errorf("resolve repo: %w", err)
	}

	bin, _ := os.Executable()
	extraArgs := ""
	if aiMode == "ai" {
		extraArgs = " --ai"
	} else if aiMode == "no-ai" {
		extraArgs = " --no-ai"
	}

	unit := fmt.Sprintf(`[Unit]
Description=GitLink CLI Workflow: %s (%s/%s)
After=network.target

[Service]
Type=simple
ExecStart=%s workflow +run --name %s --owner %s --repo %s --format json --daemon-loop --interval %s%s
Restart=on-failure
RestartSec=30
StandardOutput=append:%s
StandardError=append:%s

[Install]
WantedBy=multi-user.target
`,
		wf.Name, ctx.Owner, ctx.Repo,
		bin, wf.Name, ctx.Owner, ctx.Repo, interval, extraArgs,
		daemonLogPath(wf.Name), daemonLogPath(wf.Name),
	)

	unitPath := filepath.Join(config.ConfigDir(), fmt.Sprintf("workflow-%s.service", wf.Name))
	if err := os.WriteFile(unitPath, []byte(unit), 0644); err != nil {
		return fmt.Errorf("写入 unit 文件: %w", err)
	}

	fmt.Printf("Systemd unit 已写入: %s\n\n", unitPath)
	fmt.Println("安装步骤:")
	fmt.Printf("  sudo cp %s /etc/systemd/system/\n", unitPath)
	fmt.Println("  sudo systemctl daemon-reload")
	fmt.Printf("  sudo systemctl enable workflow-%s\n", wf.Name)
	fmt.Printf("  sudo systemctl start workflow-%s\n", wf.Name)
	fmt.Println()
	fmt.Printf("查看日志: journalctl -u workflow-%s -f\n", wf.Name)
	return nil
}

func savePID(name string, pid int) error {
	dir := config.ConfigDir()
	if err := os.MkdirAll(dir, 0700); err != nil {
		return err
	}
	return os.WriteFile(pidPath(name), []byte(strconv.Itoa(pid)), 0600)
}

func readPID(name string) (int, error) {
	data, err := os.ReadFile(pidPath(name))
	if err != nil {
		if os.IsNotExist(err) {
			return 0, fmt.Errorf("daemon %q is not running (no PID file)", name)
		}
		return 0, err
	}
	return strconv.Atoi(string(data))
}

func cleanPID(name string) {
	os.Remove(pidPath(name))
}

func processRunning(pid int) bool {
	proc, err := os.FindProcess(pid)
	if err != nil {
		return false
	}
	return proc.Signal(syscall.Signal(0)) == nil
}

func pidPath(name string) string {
	return filepath.Join(config.ConfigDir(), fmt.Sprintf("workflow-%s.pid", name))
}