403 lines
14 KiB
Go
403 lines
14 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/spf13/cobra"
|
|
"github.com/steveyegge/beads"
|
|
"github.com/steveyegge/beads/internal/daemon"
|
|
"github.com/steveyegge/beads/internal/rpc"
|
|
"github.com/steveyegge/beads/internal/storage/sqlite"
|
|
)
|
|
|
|
var daemonCmd = &cobra.Command{
|
|
Use: "daemon",
|
|
Short: "Run background sync daemon",
|
|
Long: `Run a background daemon that automatically syncs issues with git remote.
|
|
|
|
The daemon will:
|
|
- Poll for changes at configurable intervals (default: 5 seconds)
|
|
- Export pending database changes to JSONL
|
|
- Auto-commit changes if --auto-commit flag set
|
|
- Auto-push commits if --auto-push flag set
|
|
- Pull remote changes periodically
|
|
- Auto-import when remote changes detected
|
|
|
|
Use --stop to stop a running daemon.
|
|
Use --status to check if daemon is running.
|
|
Use --health to check daemon health and metrics.`,
|
|
Run: func(cmd *cobra.Command, args []string) {
|
|
stop, _ := cmd.Flags().GetBool("stop")
|
|
status, _ := cmd.Flags().GetBool("status")
|
|
health, _ := cmd.Flags().GetBool("health")
|
|
metrics, _ := cmd.Flags().GetBool("metrics")
|
|
migrateToGlobal, _ := cmd.Flags().GetBool("migrate-to-global")
|
|
interval, _ := cmd.Flags().GetDuration("interval")
|
|
autoCommit, _ := cmd.Flags().GetBool("auto-commit")
|
|
autoPush, _ := cmd.Flags().GetBool("auto-push")
|
|
logFile, _ := cmd.Flags().GetString("log")
|
|
global, _ := cmd.Flags().GetBool("global")
|
|
|
|
if interval <= 0 {
|
|
fmt.Fprintf(os.Stderr, "Error: interval must be positive (got %v)\n", interval)
|
|
os.Exit(1)
|
|
}
|
|
|
|
pidFile, err := getPIDFilePath(global)
|
|
if err != nil {
|
|
fmt.Fprintf(os.Stderr, "Error: %v\n", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
if status {
|
|
showDaemonStatus(pidFile, global)
|
|
return
|
|
}
|
|
|
|
if health {
|
|
showDaemonHealth(global)
|
|
return
|
|
}
|
|
|
|
if metrics {
|
|
showDaemonMetrics(global)
|
|
return
|
|
}
|
|
|
|
if migrateToGlobal {
|
|
migrateToGlobalDaemon()
|
|
return
|
|
}
|
|
|
|
if stop {
|
|
stopDaemon(pidFile)
|
|
return
|
|
}
|
|
|
|
// Skip daemon-running check if we're the forked child (BD_DAEMON_FOREGROUND=1)
|
|
// because the check happens in the parent process before forking
|
|
if os.Getenv("BD_DAEMON_FOREGROUND") != "1" {
|
|
// Check if daemon is already running
|
|
if isRunning, pid := isDaemonRunning(pidFile); isRunning {
|
|
// Check if running daemon has compatible version
|
|
socketPath := getSocketPathForPID(pidFile, global)
|
|
if client, err := rpc.TryConnectWithTimeout(socketPath, 1*time.Second); err == nil && client != nil {
|
|
health, healthErr := client.Health()
|
|
_ = client.Close()
|
|
|
|
// If we can check version and it's compatible, exit
|
|
if healthErr == nil && health.Compatible {
|
|
fmt.Fprintf(os.Stderr, "Error: daemon already running (PID %d, version %s)\n", pid, health.Version)
|
|
fmt.Fprintf(os.Stderr, "Use 'bd daemon --stop%s' to stop it first\n", boolToFlag(global, " --global"))
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Version mismatch - auto-stop old daemon
|
|
if healthErr == nil && !health.Compatible {
|
|
fmt.Fprintf(os.Stderr, "Warning: daemon version mismatch (daemon: %s, client: %s)\n", health.Version, Version)
|
|
fmt.Fprintf(os.Stderr, "Stopping old daemon and starting new one...\n")
|
|
stopDaemon(pidFile)
|
|
// Continue with daemon startup
|
|
}
|
|
} else {
|
|
// Can't check version - assume incompatible
|
|
fmt.Fprintf(os.Stderr, "Error: daemon already running (PID %d)\n", pid)
|
|
fmt.Fprintf(os.Stderr, "Use 'bd daemon --stop%s' to stop it first\n", boolToFlag(global, " --global"))
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Global daemon doesn't support auto-commit/auto-push (no sync loop)
|
|
if global && (autoCommit || autoPush) {
|
|
fmt.Fprintf(os.Stderr, "Error: --auto-commit and --auto-push are not supported with --global\n")
|
|
fmt.Fprintf(os.Stderr, "Hint: global daemon runs in routing mode and doesn't perform background sync\n")
|
|
fmt.Fprintf(os.Stderr, " Use local daemon (without --global) for auto-commit/auto-push features\n")
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Validate we're in a git repo (skip for global daemon)
|
|
if !global && !isGitRepo() {
|
|
fmt.Fprintf(os.Stderr, "Error: not in a git repository\n")
|
|
fmt.Fprintf(os.Stderr, "Hint: run 'git init' to initialize a repository\n")
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Check for upstream if auto-push enabled
|
|
if autoPush && !gitHasUpstream() {
|
|
fmt.Fprintf(os.Stderr, "Error: no upstream configured (required for --auto-push)\n")
|
|
fmt.Fprintf(os.Stderr, "Hint: git push -u origin <branch-name>\n")
|
|
os.Exit(1)
|
|
}
|
|
|
|
// Warn if starting daemon in a git worktree
|
|
if !global {
|
|
// Ensure dbPath is set for warning
|
|
if dbPath == "" {
|
|
if foundDB := beads.FindDatabasePath(); foundDB != "" {
|
|
dbPath = foundDB
|
|
}
|
|
}
|
|
if dbPath != "" {
|
|
warnWorktreeDaemon(dbPath)
|
|
}
|
|
}
|
|
|
|
// Start daemon
|
|
scope := "local"
|
|
if global {
|
|
scope = "global"
|
|
}
|
|
fmt.Printf("Starting bd daemon (%s, interval: %v, auto-commit: %v, auto-push: %v)\n",
|
|
scope, interval, autoCommit, autoPush)
|
|
if logFile != "" {
|
|
fmt.Printf("Logging to: %s\n", logFile)
|
|
}
|
|
|
|
startDaemon(interval, autoCommit, autoPush, logFile, pidFile, global)
|
|
},
|
|
}
|
|
|
|
func init() {
|
|
daemonCmd.Flags().Duration("interval", 5*time.Second, "Sync check interval")
|
|
daemonCmd.Flags().Bool("auto-commit", false, "Automatically commit changes")
|
|
daemonCmd.Flags().Bool("auto-push", false, "Automatically push commits")
|
|
daemonCmd.Flags().Bool("stop", false, "Stop running daemon")
|
|
daemonCmd.Flags().Bool("status", false, "Show daemon status")
|
|
daemonCmd.Flags().Bool("health", false, "Check daemon health and metrics")
|
|
daemonCmd.Flags().Bool("metrics", false, "Show detailed daemon metrics")
|
|
daemonCmd.Flags().Bool("migrate-to-global", false, "Migrate from local to global daemon")
|
|
daemonCmd.Flags().String("log", "", "Log file path (default: .beads/daemon.log)")
|
|
daemonCmd.Flags().Bool("global", false, "Run as global daemon (socket at ~/.beads/bd.sock)")
|
|
daemonCmd.Flags().BoolVar(&jsonOutput, "json", false, "Output JSON format")
|
|
rootCmd.AddCommand(daemonCmd)
|
|
}
|
|
func runDaemonLoop(interval time.Duration, autoCommit, autoPush bool, logPath, pidFile string, global bool) {
|
|
logF, log := setupDaemonLogger(logPath)
|
|
defer func() { _ = logF.Close() }()
|
|
|
|
// Determine database path first (needed for lock file metadata)
|
|
daemonDBPath := ""
|
|
if !global {
|
|
daemonDBPath = dbPath
|
|
if daemonDBPath == "" {
|
|
if foundDB := beads.FindDatabasePath(); foundDB != "" {
|
|
daemonDBPath = foundDB
|
|
} else {
|
|
log.log("Error: no beads database found")
|
|
log.log("Hint: run 'bd init' to create a database or set BEADS_DB environment variable")
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
}
|
|
|
|
lock, err := setupDaemonLock(pidFile, daemonDBPath, log)
|
|
if err != nil {
|
|
os.Exit(1)
|
|
}
|
|
defer func() { _ = lock.Close() }()
|
|
defer func() { _ = os.Remove(pidFile) }()
|
|
|
|
log.log("Daemon started (interval: %v, auto-commit: %v, auto-push: %v)", interval, autoCommit, autoPush)
|
|
|
|
if global {
|
|
runGlobalDaemon(log)
|
|
return
|
|
}
|
|
|
|
// Check for multiple .db files (ambiguity error)
|
|
beadsDir := filepath.Dir(daemonDBPath)
|
|
matches, err := filepath.Glob(filepath.Join(beadsDir, "*.db"))
|
|
if err == nil && len(matches) > 1 {
|
|
// Filter out backup files (*.backup-*.db, *.backup.db)
|
|
var validDBs []string
|
|
for _, match := range matches {
|
|
baseName := filepath.Base(match)
|
|
// Skip if it's a backup file (contains ".backup" in name)
|
|
if !strings.Contains(baseName, ".backup") && baseName != "vc.db" {
|
|
validDBs = append(validDBs, match)
|
|
}
|
|
}
|
|
if len(validDBs) > 1 {
|
|
errMsg := fmt.Sprintf("Error: Multiple database files found in %s:\n", beadsDir)
|
|
for _, db := range validDBs {
|
|
errMsg += fmt.Sprintf(" - %s\n", filepath.Base(db))
|
|
}
|
|
errMsg += fmt.Sprintf("\nBeads requires a single canonical database: %s\n", beads.CanonicalDatabaseName)
|
|
errMsg += "Run 'bd init' to migrate legacy databases or manually remove old databases\n"
|
|
errMsg += "Or run 'bd doctor' for more diagnostics"
|
|
|
|
log.log(errMsg)
|
|
|
|
// Write error to file so user can see it without checking logs
|
|
errFile := filepath.Join(beadsDir, "daemon-error")
|
|
// nolint:gosec // G306: Error file needs to be readable for debugging
|
|
if err := os.WriteFile(errFile, []byte(errMsg), 0644); err != nil {
|
|
log.log("Warning: could not write daemon-error file: %v", err)
|
|
}
|
|
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// Validate using canonical name
|
|
dbBaseName := filepath.Base(daemonDBPath)
|
|
if dbBaseName != beads.CanonicalDatabaseName {
|
|
log.log("Error: Non-canonical database name: %s", dbBaseName)
|
|
log.log("Expected: %s", beads.CanonicalDatabaseName)
|
|
log.log("")
|
|
log.log("Run 'bd init' to migrate to canonical name")
|
|
os.Exit(1)
|
|
}
|
|
|
|
log.log("Using database: %s", daemonDBPath)
|
|
|
|
// Clear any previous daemon-error file on successful startup
|
|
errFile := filepath.Join(beadsDir, "daemon-error")
|
|
if err := os.Remove(errFile); err != nil && !os.IsNotExist(err) {
|
|
log.log("Warning: could not remove daemon-error file: %v", err)
|
|
}
|
|
|
|
store, err := sqlite.New(daemonDBPath)
|
|
if err != nil {
|
|
log.log("Error: cannot open database: %v", err)
|
|
os.Exit(1)
|
|
}
|
|
defer func() { _ = store.Close() }()
|
|
log.log("Database opened: %s", daemonDBPath)
|
|
|
|
// Hydrate from multi-repo if configured
|
|
hydrateCtx := context.Background()
|
|
if results, err := store.HydrateFromMultiRepo(hydrateCtx); err != nil {
|
|
log.log("Error: multi-repo hydration failed: %v", err)
|
|
os.Exit(1)
|
|
} else if results != nil {
|
|
log.log("Multi-repo hydration complete:")
|
|
for repo, count := range results {
|
|
log.log(" %s: %d issues", repo, count)
|
|
}
|
|
}
|
|
|
|
// Validate database fingerprint
|
|
if err := validateDatabaseFingerprint(store, &log); err != nil {
|
|
if os.Getenv("BEADS_IGNORE_REPO_MISMATCH") != "1" {
|
|
log.log("Error: %v", err)
|
|
os.Exit(1)
|
|
}
|
|
log.log("Warning: repository mismatch ignored (BEADS_IGNORE_REPO_MISMATCH=1)")
|
|
}
|
|
|
|
// Validate schema version matches daemon version
|
|
versionCtx := context.Background()
|
|
dbVersion, err := store.GetMetadata(versionCtx, "bd_version")
|
|
if err != nil && err.Error() != "metadata key not found: bd_version" {
|
|
log.log("Error: failed to read database version: %v", err)
|
|
os.Exit(1)
|
|
}
|
|
|
|
if dbVersion != "" && dbVersion != Version {
|
|
log.log("Warning: Database schema version mismatch")
|
|
log.log(" Database version: %s", dbVersion)
|
|
log.log(" Daemon version: %s", Version)
|
|
log.log(" Auto-upgrading database to daemon version...")
|
|
|
|
// Auto-upgrade database to daemon version
|
|
// The daemon operates on its own database, so it should always use its own version
|
|
if err := store.SetMetadata(versionCtx, "bd_version", Version); err != nil {
|
|
log.log("Error: failed to update database version: %v", err)
|
|
|
|
// Allow override via environment variable for emergencies
|
|
if os.Getenv("BEADS_IGNORE_VERSION_MISMATCH") != "1" {
|
|
os.Exit(1)
|
|
}
|
|
log.log("Warning: Proceeding despite version update failure (BEADS_IGNORE_VERSION_MISMATCH=1)")
|
|
} else {
|
|
log.log(" Database version updated to %s", Version)
|
|
}
|
|
} else if dbVersion == "" {
|
|
// Old database without version metadata - set it now
|
|
log.log("Warning: Database missing version metadata, setting to %s", Version)
|
|
if err := store.SetMetadata(versionCtx, "bd_version", Version); err != nil {
|
|
log.log("Error: failed to set database version: %v", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// Get workspace path (.beads directory) - beadsDir already defined above
|
|
// Get actual workspace root (parent of .beads)
|
|
workspacePath := filepath.Dir(beadsDir)
|
|
socketPath := filepath.Join(beadsDir, "bd.sock")
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
server, serverErrChan, err := startRPCServer(ctx, socketPath, store, workspacePath, daemonDBPath, log)
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
// Register daemon in global registry
|
|
registry, err := daemon.NewRegistry()
|
|
if err != nil {
|
|
log.log("Warning: failed to create registry: %v", err)
|
|
} else {
|
|
entry := daemon.RegistryEntry{
|
|
WorkspacePath: workspacePath,
|
|
SocketPath: socketPath,
|
|
DatabasePath: daemonDBPath,
|
|
PID: os.Getpid(),
|
|
Version: Version,
|
|
StartedAt: time.Now(),
|
|
}
|
|
if err := registry.Register(entry); err != nil {
|
|
log.log("Warning: failed to register daemon: %v", err)
|
|
} else {
|
|
log.log("Registered in global registry")
|
|
}
|
|
// Ensure we unregister on exit
|
|
defer func() {
|
|
if err := registry.Unregister(workspacePath, os.Getpid()); err != nil {
|
|
log.log("Warning: failed to unregister daemon: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
|
|
doSync := createSyncFunc(ctx, store, autoCommit, autoPush, log)
|
|
doSync()
|
|
|
|
// Choose event loop based on BEADS_DAEMON_MODE
|
|
daemonMode := os.Getenv("BEADS_DAEMON_MODE")
|
|
if daemonMode == "" {
|
|
daemonMode = "events" // Default to event-driven mode (production-ready as of v0.21.0)
|
|
}
|
|
|
|
switch daemonMode {
|
|
case "events":
|
|
log.log("Using event-driven mode")
|
|
jsonlPath := findJSONLPath()
|
|
if jsonlPath == "" {
|
|
log.log("Error: JSONL path not found, cannot use event-driven mode")
|
|
log.log("Falling back to polling mode")
|
|
runEventLoop(ctx, cancel, ticker, doSync, server, serverErrChan, log)
|
|
} else {
|
|
// Event-driven mode uses separate export-only and import-only functions
|
|
doExport := createExportFunc(ctx, store, autoCommit, autoPush, log)
|
|
doAutoImport := createAutoImportFunc(ctx, store, log)
|
|
runEventDrivenLoop(ctx, cancel, server, serverErrChan, store, jsonlPath, doExport, doAutoImport, log)
|
|
}
|
|
case "poll":
|
|
log.log("Using polling mode (interval: %v)", interval)
|
|
runEventLoop(ctx, cancel, ticker, doSync, server, serverErrChan, log)
|
|
default:
|
|
log.log("Unknown BEADS_DAEMON_MODE: %s (valid: poll, events), defaulting to poll", daemonMode)
|
|
runEventLoop(ctx, cancel, ticker, doSync, server, serverErrChan, log)
|
|
}
|
|
}
|