darroyo
84b185be39
Phase A - Stability: - Engine.Start(): recovers orphaned jobs (running/queued/waking_up) after crash - Engine.Stop(): graceful shutdown - cancels all in-flight jobs and waits - Queue keyed by jobID (not syncPairID): cancel now targets exact job - Local rsync uses jobCtx (context.Background() replaced) - Migrations wrapped in transactions; checksums stored Phase B - Security: - admin/admin default removed: SYNCSERVER_ADMIN_PASSWORD required on first run - Path validation: rejects .., leading -, null bytes in sync pair paths - Rsync flags allowlist: dangerous flags blocked (--rsync-path, -e, --files-from) - Shell concat in RunRemote replaced with proper sh -c escaping - knownhosts: replaced custom parser with golang.org/x/crypto/ssh/knownhosts - RequireAdmin wired: machine CRUD, SSH key ops, settings require admin role - deploy-keys: uses authorized_keys only (no private key upload) - Hardcoded /var/lib/syncserver/ssh paths replaced with cfg.SSHDir() Phase C - Operational: - /readyz health check: DB query + SSH dir accessibility - /metrics endpoint: Prometheus text format (jobs, queue, machines) - Event struct JSON tags: job_id, machine_id, type (snake_case) - EventBus broadcast: fanned out to all subscribers - SQLite VACUUM INTO backup: scheduled before cleanup if BackupDir set - Filesystem job log cleanup: removes .log files for purged jobs - Backup retention: old backups auto-purged Phase D - Frontend: - Schedules page: REST API + full CRUD UI for cron schedules - Dashboard: cancel button for running/queued jobs - JobDetail: server-side log download via API - Settings: displays data_dir from server - 404 page: proper NotFound component Phase E - Tests: - auth_test.go: JWT, bcrypt, middleware, seed (18 tests) - models_test.go: Job, Machine, SyncPair, Schedule repos (18 tests) - go test -race: no data races found
107 lines
2.1 KiB
Go
107 lines
2.1 KiB
Go
package api
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"time"
|
|
|
|
"github.com/syncserver/internal/syncengine"
|
|
)
|
|
|
|
type SSEHandler struct {
|
|
engine *syncengine.Engine
|
|
}
|
|
|
|
func NewSSEHandler(engine *syncengine.Engine) *SSEHandler {
|
|
return &SSEHandler{engine: engine}
|
|
}
|
|
|
|
func (h *SSEHandler) StreamAll(w http.ResponseWriter, r *http.Request) {
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "SSE not supported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
flusher.Flush()
|
|
|
|
if h.engine == nil {
|
|
return
|
|
}
|
|
|
|
events, unsub := h.engine.SubscribeGlobal()
|
|
defer unsub()
|
|
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case evt := <-events:
|
|
data, _ := json.Marshal(evt)
|
|
fmt.Fprintf(w, "data: %s\n\n", data)
|
|
flusher.Flush()
|
|
case <-r.Context().Done():
|
|
return
|
|
case <-ticker.C:
|
|
fmt.Fprintf(w, ": keepalive\n\n")
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (h *SSEHandler) StreamJob(w http.ResponseWriter, r *http.Request) {
|
|
jobIDStr := r.URL.Query().Get("job_id")
|
|
if jobIDStr == "" {
|
|
http.Error(w, "job_id required", http.StatusBadRequest)
|
|
return
|
|
}
|
|
jobID, err := strconv.ParseInt(jobIDStr, 10, 64)
|
|
if err != nil {
|
|
http.Error(w, "invalid job_id", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "SSE not supported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
flusher.Flush()
|
|
|
|
if h.engine == nil {
|
|
return
|
|
}
|
|
|
|
events, unsub := h.engine.SubscribeJob(jobID)
|
|
defer unsub()
|
|
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case evt := <-events:
|
|
data, _ := json.Marshal(evt)
|
|
fmt.Fprintf(w, "data: %s\n\n", data)
|
|
flusher.Flush()
|
|
case <-r.Context().Done():
|
|
return
|
|
case <-ticker.C:
|
|
fmt.Fprintf(w, ": keepalive\n\n")
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|