Add SSH key management, job history persistence, and live streaming

- SSH key management: generate ed25519 keypairs or import public keys from UI (/ssh-keys), per-machine key selection in Machines form, one-time private key download with hash verification - Fix engine to use machine-specific SSH key (was hardcoded to server key) - Job log persistence: write to job_logs table (DB) with batched inserts, buffer of 50 lines; GetAllFiltered with status/pair/date range filters - EventBus refactor: per-job subscriber channels, global channel, non-blocking - SSE endpoints: /jobs/stream (all), /jobs/:id/log/stream (per-job live) - JobDetail page: live log streaming, auto-scroll, cancel, duration - JobHistory: filters (pair, status, date range), pagination, link to detail - Cleanup scheduler: daily purge of job_logs and finished jobs older than SYNCSERVER_RETENTION_DAYS (default 30) - Migration 0002: indexes on job_logs(job_id), jobs(status,created_at), jobs(sync_pair_id)
2026-07-07 20:36:11 -04:00
parent 5374ab81cc
commit bfa006f4ab
22 changed files with 1424 additions and 136 deletions
@@ -19,17 +19,18 @@ type Engine struct {
 	db       *sql.DB
 	cfg      *config.Config
 	queue    *Queue
-	eventBus chan Event
+	eventBus *EventBus
 	mu       sync.RWMutex
 	stopped  bool
 }

 type Event struct {
-	Type    string
-	JobID   int64
-	Status  string
-	Line    string
-	Stream  string
+	Type  string
+	JobID int64
+	Key   string
+	Value string
+	Line  string
+	Stream string
 }

 func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine {
@@ -37,7 +38,7 @@ func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine {
 		db:       database.SQLDB(),
 		cfg:      cfg,
 		queue:    NewQueue(),
-		eventBus: make(chan Event, 100),
+		eventBus: NewEventBus(200),
 	}
 	return e
 }
@@ -45,8 +46,12 @@ func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine {
 func (e *Engine) Start() {}
 func (e *Engine) Stop()  {}

-func (e *Engine) Events() <-chan Event {
-	return e.eventBus
+func (e *Engine) SubscribeJob(jobID int64) (chan Event, func()) {
+	return e.eventBus.Subscribe(jobID)
+}
+
+func (e *Engine) SubscribeGlobal() (chan Event, func()) {
+	return e.eventBus.SubscribeGlobal()
 }

 func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
@@ -106,7 +111,7 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
 	}

 	e.setJobStatus(jobID, "waking_up")
-	e.emit(Event{Type: "status", JobID: jobID, Status: "waking_up"})
+	e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "waking_up"})

 	var targetMachine *models.Machine
 	var remotePort int
@@ -134,52 +139,103 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error {
 			timeout := time.Duration(targetMachine.WakeTimeoutSeconds) * time.Second
 			interval := time.Duration(targetMachine.WakeCheckIntervalSeconds) * time.Second
 			if err := wol.WaitUntilReady(jobCtx, targetMachine.Host, remotePort, timeout, interval, false); err != nil {
-				e.setJobStatus(jobID, "failed")
-				e.emit(Event{Type: "status", JobID: jobID, Status: "failed", Line: err.Error()})
-				return fmt.Errorf("machine not ready: %w", err)
+		e.setJobStatus(jobID, "failed")
+		e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: err.Error()})
+		return fmt.Errorf("machine not ready: %w", err)
 			}
 		}
 	}

 	e.setJobStatus(jobID, "running")
 	e.setJobLogFile(jobID, logPath)
-	e.emit(Event{Type: "status", JobID: jobID, Status: "running"})
+	e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "running"})
 	slog.Info("job started", "job_id", jobID, "pair", pair.Name)

-	var privKey string
-	if targetMachine != nil && targetMachine.SSHKeyID != nil {
+	privKey, err := e.resolveSSHKey(targetMachine)
+	if err != nil {
+		slog.Warn("failed to resolve SSH key, using server key", "error", err)
 		privKey = filepath.Join(e.cfg.SSHDir(), "id_ed25519")
 	}

 	runner := NewRsyncRunner(e.cfg.SSHDir(), privKey)
+	logRepo := models.NewJobLogRepository(e.db)
+	var outBuf, errBuf []string
+	flush := func() {
+		if len(outBuf) > 0 {
+			logRepo.InsertBatch(jobID, "stdout", outBuf)
+			for _, l := range outBuf {
+				e.emit(Event{Type: "log", JobID: jobID, Stream: "stdout", Line: l})
+			}
+			outBuf = nil
+		}
+		if len(errBuf) > 0 {
+			logRepo.InsertBatch(jobID, "stderr", errBuf)
+			for _, l := range errBuf {
+				e.emit(Event{Type: "log", JobID: jobID, Stream: "stderr", Line: l})
+			}
+			errBuf = nil
+		}
+	}
+
 	onLine := func(stream, line string) {
-		e.emit(Event{Type: "log", JobID: jobID, Stream: stream, Line: line})
+		f, _ := os.OpenFile(logPath, os.O_APPEND|os.O_WRONLY, 0644)
+		if f != nil {
+			fmt.Fprintln(f, line)
+			f.Close()
+		}
+		if stream == "stdout" {
+			outBuf = append(outBuf, line)
+		} else {
+			errBuf = append(errBuf, line)
+		}
+		if len(outBuf) >= 50 || len(errBuf) >= 50 {
+			flush()
+		}
 	}

 	result, err := runner.Run(jobCtx, cfg, onLine)
+	flush()
+
 	if err != nil {
 		if jobCtx.Err() != nil {
-			e.setJobStatus(jobID, "cancelled")
-			e.emit(Event{Type: "status", JobID: jobID, Status: "cancelled"})
+		e.setJobStatus(jobID, "cancelled")
+			e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "cancelled"})
 			return jobCtx.Err()
 		}
 		e.setJobStatus(jobID, "failed")
-		e.emit(Event{Type: "status", JobID: jobID, Status: "failed", Line: err.Error()})
+		e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: err.Error()})
 		return fmt.Errorf("rsync error: %w", err)
 	}

 	if result.ExitCode != 0 {
-		e.setJobStatus(jobID, "failed")
-		e.emit(Event{Type: "status", JobID: jobID, Status: "failed", Line: result.Stderr})
+	e.setJobStatus(jobID, "failed")
+		e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "failed", Line: result.Stderr})
 		return fmt.Errorf("rsync exited with code %d: %s", result.ExitCode, result.Stderr)
 	}

 	e.setJobStatus(jobID, "success")
-	e.emit(Event{Type: "status", JobID: jobID, Status: "success"})
+	e.emit(Event{Type: "status", JobID: jobID, Key: "status", Value: "success"})
 	slog.Info("job completed", "job_id", jobID, "pair", pair.Name)
+	e.persistAndClose(jobID)
 	return nil
 }

+func (e *Engine) persistAndClose(jobID int64) {
+	e.eventBus.CloseJobChannels(jobID)
+}
+
+func (e *Engine) resolveSSHKey(machine *models.Machine) (string, error) {
+	if machine == nil || machine.SSHKeyID == nil {
+		return filepath.Join(e.cfg.SSHDir(), "id_ed25519"), nil
+	}
+	sshKeyRepo := models.NewSSHKeyRepository(e.db)
+	sshKey, err := sshKeyRepo.GetByID(*machine.SSHKeyID)
+	if err != nil {
+		return "", fmt.Errorf("fetching ssh key: %w", err)
+	}
+	return sshKey.PrivateKeyPath, nil
+}
+
 func (e *Engine) Cancel(jobID int64, syncPairID int64) bool {
 	if e.queue.IsRunning(syncPairID) {
 		e.queue.Cancel(syncPairID)
@@ -199,11 +255,7 @@ func (e *Engine) setJobLogFile(jobID int64, path string) {
 }

 func (e *Engine) emit(evt Event) {
-	select {
-	case e.eventBus <- evt:
-	default:
-		slog.Warn("event bus full, dropping event", "type", evt.Type)
-	}
+	e.eventBus.Publish(evt)
 }

 func buildPath(path string, machine *models.Machine) string {
@@ -0,0 +1,92 @@
+package syncengine
+
+import (
+	"log/slog"
+	"sync"
+)
+
+type EventBus struct {
+	subscribers map[int64]map[chan Event]struct{}
+	mu          sync.RWMutex
+	global      chan Event
+	bufferSize  int
+}
+
+func NewEventBus(bufferSize int) *EventBus {
+	return &EventBus{
+		subscribers: make(map[int64]map[chan Event]struct{}),
+		global:      make(chan Event, bufferSize),
+		bufferSize:  bufferSize,
+	}
+}
+
+func (eb *EventBus) Subscribe(jobID int64) (chan Event, func()) {
+	eb.mu.Lock()
+	defer eb.mu.Unlock()
+	if eb.subscribers[jobID] == nil {
+		eb.subscribers[jobID] = make(map[chan Event]struct{})
+	}
+	ch := make(chan Event, eb.bufferSize)
+	eb.subscribers[jobID][ch] = struct{}{}
+	unsubscribe := func() {
+		eb.mu.Lock()
+		defer eb.mu.Unlock()
+		if subs, ok := eb.subscribers[jobID]; ok {
+			delete(subs, ch)
+			if len(subs) == 0 {
+				delete(eb.subscribers, jobID)
+			}
+		}
+		close(ch)
+	}
+	return ch, unsubscribe
+}
+
+func (eb *EventBus) SubscribeGlobal() (chan Event, func()) {
+	eb.mu.RLock()
+	ch := make(chan Event, eb.bufferSize)
+	eb.mu.RUnlock()
+	go func() {
+		for evt := range eb.global {
+			select {
+			case ch <- evt:
+			default:
+				slog.Warn("global event subscriber buffer full, dropping event", "type", evt.Type)
+			}
+		}
+		close(ch)
+	}()
+	return ch, func() { close(ch) }
+}
+
+func (eb *EventBus) Publish(evt Event) {
+	eb.mu.RLock()
+	defer eb.mu.RUnlock()
+
+	if subs, ok := eb.subscribers[evt.JobID]; ok {
+		for ch := range subs {
+			select {
+			case ch <- evt:
+			default:
+				slog.Warn("job event subscriber buffer full, dropping event", "job_id", evt.JobID)
+			}
+		}
+	}
+
+	select {
+	case eb.global <- evt:
+	default:
+		slog.Warn("global event bus full, dropping event", "type", evt.Type)
+	}
+}
+
+func (eb *EventBus) CloseJobChannels(jobID int64) {
+	eb.mu.Lock()
+	defer eb.mu.Unlock()
+	if subs, ok := eb.subscribers[jobID]; ok {
+		for ch := range subs {
+			close(ch)
+		}
+		delete(eb.subscribers, jobID)
+	}
+}