package selfupdate import ( "bytes" "context" "encoding/json" "fmt" "log" "net/http" "os/exec" "strings" "sync" "time" "gitea.dooplex.hu/admin/felhom-controller/internal/agentapi" "gitea.dooplex.hu/admin/felhom-controller/internal/config" ) // CheckResult holds the result of a version check. type CheckResult struct { CurrentVersion string `json:"current_version"` LatestVersion string `json:"latest_version"` UpdateAvailable bool `json:"update_available"` Error string `json:"error,omitempty"` CheckedAt string `json:"checked_at"` } // UpdateStatus is the complete status returned by the API. type UpdateStatus struct { Running bool `json:"running"` LastCheck *CheckResult `json:"last_check,omitempty"` LastState *UpdateState `json:"last_state,omitempty"` } // AgentSwapper is the agent local-API capability the updater delegates the actual container swap to // (agentic controller update, Phase 1). The agent — external to this controller container — rewrites // /etc/felhom-controller-image, restarts the bootstrap unit, verifies health, and rolls back. Satisfied // by *agentapi.Client; faked in tests. nil when this is not a provisioned guest (no agent) → updates // are unavailable (the old in-container compose flow is GONE). type AgentSwapper interface { SwapController(ctx context.Context, image string) (agentapi.SwapResult, error) } // Updater manages controller self-updates. type Updater struct { cfg *config.SelfUpdateConfig gitCfg *config.GitConfig currentVer string dataDir string agent AgentSwapper // Phase 1: the host agent performs the swap (nil → updates unavailable) logger *log.Logger debug bool mu sync.Mutex latestVersion string lastCheck *CheckResult updateRunning bool backupRunning func() bool // Phase 2 managed updates: the operator-enforced minimum version (FLOOR), learned from the hub // report ACK, and the last floor we already auto-attempted (no flapping within this process; the // persisted UpdateState guards across restarts). Auto-target is ALWAYS the floor, never latest. floor string lastAutoFloorAttempt string // Seams (default to the real implementations; overridden in tests to avoid network/docker). queryFn func() (string, error) // resolve latest registry tag (default u.queryRegistry) pullFn func(targetImage) error // pull the image in-guest (default u.pullImage) } // targetImage is a tiny named type so the pullFn seam reads clearly. type targetImage = string // NewUpdater creates a new Updater instance. agent may be nil (un-provisioned guest → no self-update). func NewUpdater(cfg *config.SelfUpdateConfig, gitCfg *config.GitConfig, currentVersion, dataDir string, agent AgentSwapper, logger *log.Logger, debug bool) *Updater { u := &Updater{ cfg: cfg, gitCfg: gitCfg, currentVer: currentVersion, dataDir: dataDir, agent: agent, logger: logger, debug: debug, } u.queryFn = u.queryRegistry u.pullFn = u.pullImage return u } func (u *Updater) dbg(format string, args ...interface{}) { if u.debug { u.logger.Printf("[DEBUG] [selfupdate] "+format, args...) } } // SetBackupRunningCheck sets the callback to check if a backup is in progress. func (u *Updater) SetBackupRunningCheck(fn func() bool) { u.mu.Lock() defer u.mu.Unlock() u.backupRunning = fn } // IsUpdateRunning returns true if an update is currently in progress. func (u *Updater) IsUpdateRunning() bool { u.mu.Lock() defer u.mu.Unlock() return u.updateRunning } // GetStatus returns the current update status for API/UI. func (u *Updater) GetStatus() UpdateStatus { u.mu.Lock() lastCheck := u.lastCheck running := u.updateRunning u.mu.Unlock() state, err := LoadState(u.dataDir) if err != nil { u.logger.Printf("[WARN] [selfupdate] Failed to load update state: %v", err) } return UpdateStatus{ Running: running, LastCheck: lastCheck, LastState: state, } } // CheckForUpdate queries the Gitea registry for the latest version tag. // Caches the result. Thread-safe. func (u *Updater) CheckForUpdate() CheckResult { result := CheckResult{ CurrentVersion: u.currentVer, CheckedAt: time.Now().UTC().Format(time.RFC3339), } // Dev version can't check for updates currentVer, err := ParseVersion(u.currentVer) if err != nil { result.Error = "Dev verzió nem ellenőrizhető" u.mu.Lock() u.lastCheck = &result u.mu.Unlock() return result } // Query registry latestStr, err := u.queryFn() if err != nil { result.Error = fmt.Sprintf("Registry lekérdezés sikertelen: %v", err) u.logger.Printf("[WARN] [selfupdate] Registry check failed: %v", err) u.mu.Lock() u.lastCheck = &result u.mu.Unlock() return result } result.LatestVersion = latestStr latestVer, err := ParseVersion(latestStr) if err != nil { result.Error = fmt.Sprintf("Érvénytelen verzió a registry-ben: %s", latestStr) u.mu.Lock() u.lastCheck = &result u.mu.Unlock() return result } cmp := latestVer.Compare(currentVer) if cmp > 0 { result.UpdateAvailable = true u.logger.Printf("[INFO] [selfupdate] Update available: %s → %s", u.currentVer, latestStr) } else { u.logger.Printf("[INFO] [selfupdate] Current version %s is up to date", u.currentVer) } u.dbg("version comparison: current=%s (%d.%d.%d), latest=%s (%d.%d.%d), cmp=%d, updateAvailable=%v", u.currentVer, currentVer.Major, currentVer.Minor, currentVer.Patch, latestStr, latestVer.Major, latestVer.Minor, latestVer.Patch, cmp, result.UpdateAvailable) u.mu.Lock() u.latestVersion = latestStr u.lastCheck = &result u.mu.Unlock() return result } // queryRegistry queries the Gitea Docker Registry V2 API for available tags. // Returns the highest valid semver tag found. func (u *Updater) queryRegistry() (string, error) { if u.gitCfg.Username == "" || u.gitCfg.Token == "" { return "", fmt.Errorf("registry hitelesítő adatok hiányoznak") } // Gitea registry V2: GET /v2///tags/list url := fmt.Sprintf("https://gitea.dooplex.hu/v2/%s/tags/list", registryImagePath(u.cfg.Image)) u.dbg("queryRegistry: url=%s user=%s", url, u.gitCfg.Username) req, err := http.NewRequest("GET", url, nil) if err != nil { return "", fmt.Errorf("creating request: %w", err) } req.SetBasicAuth(u.gitCfg.Username, u.gitCfg.Token) client := &http.Client{Timeout: 15 * time.Second} resp, err := client.Do(req) if err != nil { u.dbg("queryRegistry: HTTP request failed: %v", err) return "", fmt.Errorf("HTTP request failed: %w", err) } defer resp.Body.Close() u.dbg("queryRegistry: HTTP %d", resp.StatusCode) if resp.StatusCode == 401 { return "", fmt.Errorf("authentication failed (401)") } if resp.StatusCode != 200 { return "", fmt.Errorf("unexpected status: %d", resp.StatusCode) } var tagsResp struct { Name string `json:"name"` Tags []string `json:"tags"` } if err := json.NewDecoder(resp.Body).Decode(&tagsResp); err != nil { return "", fmt.Errorf("decoding response: %w", err) } u.dbg("queryRegistry: %d tags returned: %v", len(tagsResp.Tags), tagsResp.Tags) // Find highest semver tag var highest *Version for _, tag := range tagsResp.Tags { v, err := ParseVersion(tag) if err != nil { continue // skip non-semver tags ("latest", "dev", etc.) } if highest == nil || v.Compare(*highest) > 0 { highest = &v } } if highest == nil { return "", fmt.Errorf("no valid semver tags found") } return highest.String(), nil } // registryImagePath extracts the "owner/repo" from a full image reference. // e.g., "gitea.dooplex.hu/admin/felhom-controller" → "admin/felhom-controller" func registryImagePath(image string) string { // Remove registry host parts := strings.SplitN(image, "/", 2) if len(parts) == 2 { return parts[1] } return image } // DryRunResult holds the result of a self-update dry run. type DryRunResult struct { CurrentVersion string `json:"current_version"` LatestVersion string `json:"latest_version"` UpdateAvailable bool `json:"update_available"` AgentReachable bool `json:"agent_reachable"` // the host agent (which performs the swap) is wired PullCapable bool `json:"pull_capable"` // registry creds present for the in-guest pull TargetImage string `json:"target_image"` // what we would pull + swap to BackupRunning bool `json:"backup_running"` Error string `json:"error,omitempty"` } // DryRun checks for updates and reports what would happen without performing any changes. func (u *Updater) DryRun() *DryRunResult { result := &DryRunResult{ CurrentVersion: u.currentVer, } check := u.CheckForUpdate() result.LatestVersion = check.LatestVersion result.UpdateAvailable = check.UpdateAvailable if check.Error != "" { result.Error = check.Error return result } // The new flow: pull in-guest, then the agent swaps. Report those two capabilities. result.AgentReachable = u.agent != nil result.PullCapable = u.gitCfg.Username != "" && u.gitCfg.Token != "" if check.UpdateAvailable { result.TargetImage = fmt.Sprintf("%s:%s", u.cfg.Image, check.LatestVersion) } if u.backupRunning != nil { result.BackupRunning = u.backupRunning() } return result } // TriggerUpdate starts the self-update process. Returns error immediately if // preconditions fail. The actual update runs in a goroutine. func (u *Updater) TriggerUpdate(initiatedBy string) error { u.dbg("TriggerUpdate: initiatedBy=%s currentVer=%s", initiatedBy, u.currentVer) u.mu.Lock() if u.updateRunning { u.mu.Unlock() u.dbg("TriggerUpdate: rejected — update already running") return fmt.Errorf("Frissítés már folyamatban") } // Dev version check if _, err := ParseVersion(u.currentVer); err != nil { u.mu.Unlock() return fmt.Errorf("Dev verzió nem frissíthető") } // Backup running check if u.backupRunning != nil && u.backupRunning() { u.mu.Unlock() return fmt.Errorf("Mentés fut, próbálja később") } // Agent reachable check — the host agent performs the swap; without it there is no update path // (the old in-container docker-compose flow is removed). if u.agent == nil { u.mu.Unlock() return fmt.Errorf("A frissítés nem érhető el (nincs gazda-ügynök)") } u.updateRunning = true u.mu.Unlock() // Check for update (or use cached) result := u.CheckForUpdate() if !result.UpdateAvailable { u.mu.Lock() u.updateRunning = false u.mu.Unlock() return fmt.Errorf("Nincs elérhető frissítés") } targetVersion := result.LatestVersion targetImage := fmt.Sprintf("%s:%s", u.cfg.Image, targetVersion) previousImage := fmt.Sprintf("%s:%s", u.cfg.Image, u.currentVer) u.logger.Printf("[INFO] [selfupdate] Starting self-update: %s → %s (initiated by: %s)", u.currentVer, targetVersion, initiatedBy) u.dbg("TriggerUpdate: target=%s image=%s previousImage=%s", targetVersion, targetImage, previousImage) go u.performUpdate(targetVersion, targetImage, previousImage, initiatedBy) return nil } // SetFloor records the operator-enforced minimum controller version (the managed-update FLOOR), // learned from the hub's report ACK. Empty clears it (Phase 2 inert for this box). Cheap + safe to // call every report; pair it with MaybeAutoUpdate to reconcile. func (u *Updater) SetFloor(version string) { u.mu.Lock() defer u.mu.Unlock() if version != u.floor { u.dbg("SetFloor: floor %q → %q", u.floor, version) } u.floor = version } // GetFloor returns the current floor (for the UI / status). func (u *Updater) GetFloor() string { u.mu.Lock() defer u.mu.Unlock() return u.floor } // MaybeAutoUpdate auto-updates the controller to the FLOOR when the box is below it — the managed // default (no customer click). It REUSES the Phase 1 performUpdate flow (pull in-guest → agent swap // → rollback on failure); it never adds a second swap path and never touches the agent binary. // // It is a strict no-op unless ALL hold: // - a floor is set AND the current version parses (a dev build can't compare), // - current < floor (Scenario B: at/above floor does NOTHING — we must NOT chase latest here), // - an agent is wired (it performs the swap) and no backup is running, // - no swap is in flight, we haven't already auto-attempted this exact floor in-process, and the // persisted state doesn't already record this floor as attempted (no flapping/storm across the // report cycle or a restart), // - the floor is a real, pullable tag: floor <= latest available in the registry. If the floor // EXCEEDS the latest available (operator misconfig) we log a warning and do nothing. // // Auto-target is ALWAYS the floor, never latest. Runs right after the floor is set (post-report) — // no new timer/endpoint. func (u *Updater) MaybeAutoUpdate() { u.mu.Lock() floor := u.floor if floor == "" { u.mu.Unlock() return } if u.agent == nil { u.mu.Unlock() u.dbg("maybeAutoUpdate: no agent — auto-update unavailable") return } if u.updateRunning { u.mu.Unlock() u.dbg("maybeAutoUpdate: an update is already running — skip") return } curVer, err := ParseVersion(u.currentVer) if err != nil { u.mu.Unlock() u.dbg("maybeAutoUpdate: current %q not parseable (dev?) — skip", u.currentVer) return } floorVer, err := ParseVersion(floor) if err != nil { u.mu.Unlock() u.logger.Printf("[WARN] [selfupdate] Floor %q is not a valid version — ignoring", floor) return } // Scenario B — at/above floor: NOTHING. Must NOT update just because latest > current (that's the // customer's opt-in button, not the floor's job). if curVer.Compare(floorVer) >= 0 { u.mu.Unlock() u.dbg("maybeAutoUpdate: current %s >= floor %s — no action", u.currentVer, floor) return } // No flapping (in-process): one auto-update per below-floor condition. if u.lastAutoFloorAttempt == floor { u.mu.Unlock() u.dbg("maybeAutoUpdate: already auto-attempted floor %s this process — skip", floor) return } u.mu.Unlock() // No flapping (across restart): if the persisted state already records an attempt at THIS floor, // don't re-trigger. A failed+rolled-back auto-update restarts this process (losing the in-memory // flag), so without this a persistent failure would retry every report. A 'success' would already // be caught by the at/above check; we still guard it for completeness. if st, _ := LoadState(u.dataDir); st != nil && st.TargetVersion == floor && (st.Status == "failed" || st.Status == "success") { u.dbg("maybeAutoUpdate: persisted state already records floor %s (status=%s) — skip", floor, st.Status) return } // Validate the floor is PULLABLE: it must not exceed the latest available registry tag. We do not // chase a non-existent image. queryFn is the same registry lookup Phase 1 uses. latestStr, err := u.queryFn() if err != nil { u.logger.Printf("[WARN] [selfupdate] Auto-update: registry check failed (%v) — deferring floor %s", err, floor) return } latestVer, err := ParseVersion(latestStr) if err != nil { u.logger.Printf("[WARN] [selfupdate] Auto-update: registry returned invalid latest %q — deferring floor %s", latestStr, floor) return } if floorVer.Compare(latestVer) > 0 { u.logger.Printf("[WARN] [selfupdate] Auto-update: floor %s exceeds latest available %s (operator misconfig?) — doing nothing", floor, latestStr) return } // Commit: re-check under lock (a concurrent report may have started one) and claim the run. u.mu.Lock() if u.updateRunning || u.lastAutoFloorAttempt == floor || u.floor != floor { u.mu.Unlock() return } if u.backupRunning != nil && u.backupRunning() { u.mu.Unlock() u.dbg("maybeAutoUpdate: backup running — defer floor %s (will retry next report)", floor) return } u.updateRunning = true u.lastAutoFloorAttempt = floor u.mu.Unlock() targetImage := fmt.Sprintf("%s:%s", u.cfg.Image, floor) previousImage := fmt.Sprintf("%s:%s", u.cfg.Image, u.currentVer) u.logger.Printf("[INFO] [selfupdate] Auto-update to FLOOR: %s → %s (managed, no customer action)", u.currentVer, floor) go u.performUpdate(floor, targetImage, previousImage, "auto-floor") } // performUpdate runs the actual update in a goroutine: pull the target image IN-GUEST (shared docker // socket, our registry token), then delegate the container SWAP to the host agent (which owns the // restart + verify + rollback). This controller process is expected to be killed when the agent swaps; // success/failure is detected on the NEXT boot by VerifyStartup (current version vs target). func (u *Updater) performUpdate(targetVersion, targetImage, previousImage, initiatedBy string) { defer func() { u.mu.Lock() u.updateRunning = false u.mu.Unlock() }() u.dbg("performUpdate: starting — target=%s image=%s", targetVersion, targetImage) // 1. Write pending state (VerifyStartup reads this on the next boot to mark success/failure). state := &UpdateState{ Status: "pending", PreviousVersion: u.currentVer, PreviousImage: previousImage, TargetVersion: targetVersion, TargetImage: targetImage, InitiatedAt: time.Now().UTC().Format(time.RFC3339), InitiatedBy: initiatedBy, } if err := SaveState(u.dataDir, state); err != nil { u.logger.Printf("[ERROR] [selfupdate] Failed to save update state: %v", err) return } // 2. Pull the target image into the guest's docker storage (via the shared socket). Auth with the // existing registry token (login → pull → logout). On failure the AGENT IS NEVER CALLED and the // current controller keeps running untouched. if err := u.pullFn(targetImage); err != nil { state.Status = "failed" state.Error = err.Error() state.CompletedAt = time.Now().UTC().Format(time.RFC3339) SaveState(u.dataDir, state) u.logger.Printf("[ERROR] [selfupdate] %v", err) return } u.logger.Printf("[INFO] [selfupdate] Image pulled into guest: %s", targetImage) // 3. Delegate the swap to the host agent (it restarts the bootstrap unit + verifies + rolls back). u.logger.Printf("[INFO] [selfupdate] Requesting agent controller swap → %s (this controller will restart)", targetImage) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() res, err := u.agent.SwapController(ctx, targetImage) if err != nil { state.Status = "failed" state.Error = fmt.Sprintf("agent swap request failed: %v", err) state.CompletedAt = time.Now().UTC().Format(time.RFC3339) SaveState(u.dataDir, state) u.logger.Printf("[ERROR] [selfupdate] Agent swap request failed: %v", err) return } u.logger.Printf("[INFO] [selfupdate] Agent accepted swap (status=%s, previous=%s) — awaiting restart", res.Status, res.PreviousImage) // We now expect to be killed by the agent's `docker rm -f` + `docker run`. If we survive (e.g. the // agent's verify is still polling), just wait — the new container replaces us shortly. } // pullImage authenticates to the registry and pulls targetImage into the guest's docker storage. The // token is passed via stdin (never argv) and the session is logged out afterwards. func (u *Updater) pullImage(targetImage string) error { if u.gitCfg.Username == "" || u.gitCfg.Token == "" { return fmt.Errorf("docker pull: registry hitelesítő adatok hiányoznak") } host := registryHost(u.cfg.Image) u.dbg("pullImage: docker login %s as %s", host, u.gitCfg.Username) if out, err := runCommandStdin(u.gitCfg.Token, "docker", "login", host, "-u", u.gitCfg.Username, "--password-stdin"); err != nil { return fmt.Errorf("docker login failed: %v — %s", err, out) } defer func() { if out, err := runCommand("docker", "logout", host); err != nil { u.logger.Printf("[WARN] [selfupdate] docker logout failed: %v — %s", err, out) } }() u.logger.Printf("[INFO] [selfupdate] Pulling image: %s", targetImage) pullStart := time.Now() if out, err := runCommand("docker", "pull", targetImage); err != nil { return fmt.Errorf("docker pull failed: %v — %s", err, out) } u.dbg("pullImage: docker pull completed in %s", time.Since(pullStart).Round(time.Millisecond)) return nil } // registryHost extracts the registry host from a full image reference // ("gitea.dooplex.hu/admin/felhom-controller" → "gitea.dooplex.hu"). func registryHost(image string) string { if i := strings.IndexByte(image, '/'); i > 0 { return image[:i] } return image } // VerifyStartup checks the update state file on startup. // Called once from main.go before the scheduler starts. // Returns the state if a pending update was detected, nil otherwise. func (u *Updater) VerifyStartup() *UpdateState { u.dbg("VerifyStartup: checking update state in %s", u.dataDir) state, err := LoadState(u.dataDir) if err != nil { u.logger.Printf("[WARN] [selfupdate] Failed to load update state on startup: %v — clearing", err) ClearState(u.dataDir, u.logger) return nil } if state == nil || state.Status != "pending" { u.dbg("VerifyStartup: no pending update (state=%v)", state) return nil } u.dbg("VerifyStartup: pending update found — target=%s previous=%s", state.TargetVersion, state.PreviousVersion) // Compare current version with target currentVer, curErr := ParseVersion(u.currentVer) targetVer, tgtErr := ParseVersion(state.TargetVersion) if curErr != nil || tgtErr != nil { state.Status = "failed" state.Error = "Version parse error on startup verification" state.CompletedAt = time.Now().UTC().Format(time.RFC3339) SaveState(u.dataDir, state) u.logger.Printf("[WARN] [selfupdate] Post-update startup: version parse error (current=%s, target=%s)", u.currentVer, state.TargetVersion) return state } if currentVer.Compare(targetVer) == 0 { // Success — we're running the target version state.Status = "success" state.CompletedAt = time.Now().UTC().Format(time.RFC3339) SaveState(u.dataDir, state) u.logger.Printf("[INFO] [selfupdate] Post-update startup: update successful (%s → %s)", state.PreviousVersion, state.TargetVersion) } else { // Version mismatch — update may have failed state.Status = "failed" state.Error = fmt.Sprintf("Version mismatch: expected %s, running %s", state.TargetVersion, u.currentVer) state.CompletedAt = time.Now().UTC().Format(time.RFC3339) SaveState(u.dataDir, state) u.logger.Printf("[WARN] [selfupdate] Post-update startup: version mismatch (expected %s, running %s)", state.TargetVersion, u.currentVer) } return state } // runCommand executes a command and returns combined stdout+stderr and error. func runCommand(name string, args ...string) (string, error) { cmd := exec.Command(name, args...) var out bytes.Buffer cmd.Stdout = &out cmd.Stderr = &out err := cmd.Run() return out.String(), err } // runCommandStdin executes a command, feeding `stdin` on its standard input (used for // `docker login --password-stdin` so the token is never in argv/ps). Returns combined output. func runCommandStdin(stdin, name string, args ...string) (string, error) { cmd := exec.Command(name, args...) cmd.Stdin = strings.NewReader(stdin) var out bytes.Buffer cmd.Stdout = &out cmd.Stderr = &out err := cmd.Run() return out.String(), err }