package updates import ( "archive/tar" "compress/gzip" "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" "net/http" "os" "os/exec" "path/filepath" "regexp" "runtime" "strconv" "strings" "sync" "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rs/zerolog/log" ) // UpdateStatus represents the current status of an update type UpdateStatus struct { Status string `json:"status"` Progress int `json:"progress"` Message string `json:"message"` Error string `json:"error,omitempty"` UpdatedAt string `json:"updatedAt"` } // ReleaseInfo represents a GitHub release type ReleaseInfo struct { TagName string `json:"tag_name"` Name string `json:"name"` Body string `json:"body"` Prerelease bool `json:"prerelease"` Draft bool `json:"draft"` PublishedAt time.Time `json:"published_at"` Assets []struct { Name string `json:"name"` BrowserDownloadURL string `json:"browser_download_url"` } `json:"assets"` } // UpdateInfo represents available update information type UpdateInfo struct { Available bool `json:"available"` CurrentVersion string `json:"currentVersion"` LatestVersion string `json:"latestVersion"` ReleaseNotes string `json:"releaseNotes"` ReleaseDate time.Time `json:"releaseDate"` DownloadURL string `json:"downloadUrl"` IsPrerelease bool `json:"isPrerelease"` Warning string `json:"warning,omitempty"` } var ( errGitHubRateLimited = errors.New("GitHub API rate limit exceeded") stageDelayOnce sync.Once stageDelayValue time.Duration ) // Manager handles update operations type Manager struct { config *config.Config history *UpdateHistory status UpdateStatus statusMu sync.RWMutex checkCache map[string]*UpdateInfo // keyed by channel cacheTime map[string]time.Time // keyed by channel cacheDuration time.Duration progressChan chan UpdateStatus queue *UpdateQueue sseBroadcast *SSEBroadcaster } // ApplyUpdateRequest describes an update request initiated via the API/UI. type ApplyUpdateRequest struct { DownloadURL string Channel string InitiatedBy InitiatedBy InitiatedVia InitiatedVia Notes string } // NewManager creates a new update manager func NewManager(cfg *config.Config) *Manager { m := &Manager{ config: cfg, checkCache: make(map[string]*UpdateInfo), cacheTime: make(map[string]time.Time), cacheDuration: 5 * time.Minute, // Cache update checks for 5 minutes progressChan: make(chan UpdateStatus, 100), queue: NewUpdateQueue(), sseBroadcast: NewSSEBroadcaster(), status: UpdateStatus{ Status: "idle", UpdatedAt: time.Now().Format(time.RFC3339), }, } // Clean up old temp directories from previous failed/killed updates go m.cleanupOldTempDirs() // Start heartbeat for SSE connections (every 30 seconds) go m.sseHeartbeatLoop() return m } // SetHistory wires an update history sink for recording update progress. func (m *Manager) SetHistory(history *UpdateHistory) { m.history = history } // GetProgressChannel returns the channel for update progress func (m *Manager) GetProgressChannel() <-chan UpdateStatus { return m.progressChan } // Close closes the progress channel and cleans up resources func (m *Manager) Close() { close(m.progressChan) if m.sseBroadcast != nil { m.sseBroadcast.Close() } } // GetSSEBroadcaster returns the SSE broadcaster func (m *Manager) GetSSEBroadcaster() *SSEBroadcaster { return m.sseBroadcast } // GetQueue returns the update queue func (m *Manager) GetQueue() *UpdateQueue { return m.queue } // CheckForUpdates checks GitHub for available updates using saved config channel func (m *Manager) CheckForUpdates(ctx context.Context) (*UpdateInfo, error) { return m.CheckForUpdatesWithChannel(ctx, "") } // CheckForUpdatesWithChannel checks GitHub for available updates with optional channel override func (m *Manager) CheckForUpdatesWithChannel(ctx context.Context, channel string) (*UpdateInfo, error) { // Get current version first to auto-detect channel if needed currentInfo, err := GetCurrentVersion() if err != nil { m.updateStatus("error", 0, "Failed to get current version") return nil, fmt.Errorf("failed to get current version: %w", err) } // Track whether an explicit channel override was provided explicitChannelProvided := channel != "" // Use provided channel, or fall back to config, or auto-detect from current version if channel == "" { channel = m.config.UpdateChannel } if channel == "" && currentInfo.Channel != "" { // Auto-detect channel from current version (RC users get RC updates) channel = currentInfo.Channel } if channel == "" { channel = "stable" } // Don't use cache when channel is explicitly provided (UI might have changed it) // But DO use cache for auto-detected or default channels useCache := !explicitChannelProvided // Check cache first (only if using saved channel) if useCache { m.statusMu.RLock() cachedInfo, hasCached := m.checkCache[channel] cachedTime, hasTime := m.cacheTime[channel] if hasCached && hasTime && time.Since(cachedTime) < m.cacheDuration { m.statusMu.RUnlock() return cachedInfo, nil } m.statusMu.RUnlock() } m.updateStatus("checking", 0, "Checking for updates...") // Skip update check for Docker if currentInfo.IsDocker { info := &UpdateInfo{ Available: false, CurrentVersion: currentInfo.Version, LatestVersion: currentInfo.Version, } if useCache { m.statusMu.Lock() m.checkCache[channel] = info m.cacheTime[channel] = time.Now() m.statusMu.Unlock() } m.updateStatus("idle", 0, "Updates not available in Docker") return info, nil } // Skip update check for source builds if currentInfo.IsSourceBuild { info := &UpdateInfo{ Available: false, CurrentVersion: currentInfo.Version, LatestVersion: currentInfo.Version, } if useCache { m.statusMu.Lock() m.checkCache[channel] = info m.cacheTime[channel] = time.Now() m.statusMu.Unlock() } m.updateStatus("idle", 0, "Updates not available for source builds") return info, nil } // Parse current version first currentVer, err := ParseVersion(currentInfo.Version) if err != nil { m.updateStatus("error", 0, "Invalid current version") return nil, fmt.Errorf("failed to parse current version: %w", err) } // Get latest release from GitHub with specified channel and current version release, err := m.getLatestReleaseForChannel(ctx, channel, currentVer) if err != nil { if errors.Is(err, errGitHubRateLimited) { log.Warn().Err(err).Str("channel", channel).Msg("GitHub rate limit encountered while checking for updates") if useCache { m.statusMu.RLock() cachedInfo, hasCached := m.checkCache[channel] m.statusMu.RUnlock() if hasCached && cachedInfo != nil { m.updateStatus("idle", 0, "Using cached update info (GitHub rate limit)") return cachedInfo, nil } } info := &UpdateInfo{ Available: false, CurrentVersion: currentInfo.Version, LatestVersion: currentInfo.Version, DownloadURL: "", IsPrerelease: currentVer.IsPrerelease(), Warning: "Update check temporarily unavailable because GitHub rate limit was reached. Try again in a few minutes.", } m.updateStatus("idle", 0, "GitHub rate limit reached during update check") return info, nil } // Check if this is a "no releases found" error - handle gracefully if strings.Contains(err.Error(), "no releases found") { // No releases available for this channel - return "no update available" info := &UpdateInfo{ Available: false, CurrentVersion: currentInfo.Version, LatestVersion: currentInfo.Version, } if useCache { m.statusMu.Lock() m.checkCache[channel] = info m.cacheTime[channel] = time.Now() m.statusMu.Unlock() } m.updateStatus("idle", 0, fmt.Sprintf("No releases available for %s channel", channel)) return info, nil } // For other errors, return the error m.updateStatus("error", 0, "Failed to check for updates", err) return nil, err } latestVer, err := ParseVersion(release.TagName) if err != nil { parseErr := fmt.Errorf("failed to parse latest version: %w", err) m.updateStatus("error", 0, "Invalid latest version", parseErr) return nil, parseErr } // Find download URL for current architecture downloadURL := "" arch := runtime.GOARCH // Map Go architecture names to release asset names archMap := map[string]string{ "amd64": "amd64", "arm64": "arm64", "arm": "armv7", "386": "386", } targetArch, ok := archMap[arch] if !ok { targetArch = arch // Use as-is if not in map } // Look for architecture-specific binary targetName := fmt.Sprintf("pulse-%s-linux-%s.tar.gz", release.TagName, targetArch) for _, asset := range release.Assets { if asset.Name == targetName { downloadURL = asset.BrowserDownloadURL break } } // Fallback to any pulse tarball if exact match not found if downloadURL == "" { for _, asset := range release.Assets { if strings.HasPrefix(asset.Name, "pulse-") && strings.Contains(asset.Name, "linux") && strings.HasSuffix(asset.Name, ".tar.gz") { downloadURL = asset.BrowserDownloadURL break } } } info := &UpdateInfo{ Available: latestVer.IsNewerThan(currentVer), CurrentVersion: currentInfo.Version, LatestVersion: release.TagName, ReleaseNotes: release.Body, ReleaseDate: release.PublishedAt, DownloadURL: downloadURL, IsPrerelease: release.Prerelease, } // Cache the result (only if using saved channel) if useCache { m.statusMu.Lock() m.checkCache[channel] = info m.cacheTime[channel] = time.Now() m.statusMu.Unlock() } status := "idle" message := "No updates available" if info.Available { status = "available" message = fmt.Sprintf("Update available: %s", info.LatestVersion) } m.updateStatus(status, 100, message) return info, nil } // ApplyUpdate downloads and applies an update func (m *Manager) ApplyUpdate(ctx context.Context, req ApplyUpdateRequest) error { // Validate download URL (allow test server URLs when PULSE_UPDATE_SERVER is set) if req.DownloadURL == "" { return fmt.Errorf("download URL is required") } if os.Getenv("PULSE_UPDATE_SERVER") == "" { if !strings.HasPrefix(req.DownloadURL, "https://github.com/rcourtman/Pulse/releases/download/") { return fmt.Errorf("invalid download URL") } } // Check if Docker currentInfo, _ := GetCurrentVersion() if currentInfo.IsDocker { return fmt.Errorf("updates cannot be applied in Docker environment") } // Check for pre-v4 installation if isPreV4Installation() { return fmt.Errorf("manual migration required: Pulse v4 is a complete rewrite. Please create a fresh installation. See https://github.com/rcourtman/Pulse/releases/v4.0.0") } // Enqueue the update job job, accepted := m.queue.Enqueue(req.DownloadURL) if !accepted { return fmt.Errorf("update already in progress") } // Mark job as running m.queue.MarkRunning(job.ID) // Use job context instead of passed context ctx = job.Context m.updateStatus("downloading", 10, "Downloading update...") channel := m.resolveChannel(req.Channel, currentInfo) targetVersion := inferVersionFromDownloadURL(req.DownloadURL) initiatedBy := req.InitiatedBy if initiatedBy == "" { initiatedBy = InitiatedByUser } initiatedVia := req.InitiatedVia if initiatedVia == "" { initiatedVia = InitiatedViaAPI } start := time.Now() eventID := m.createHistoryEntry(ctx, UpdateHistoryEntry{ Action: ActionUpdate, Channel: channel, VersionFrom: currentInfo.Version, VersionTo: targetVersion, DeploymentType: currentInfo.DeploymentType, InitiatedBy: initiatedBy, InitiatedVia: initiatedVia, Status: StatusInProgress, Notes: req.Notes, }) var runErr error defer func() { if eventID == "" { return } status := StatusSuccess if runErr != nil { status = StatusFailed } m.completeHistoryEntry(ctx, eventID, status, start, runErr) }() // Create temp directory in a location we can write to // Try multiple locations in order of preference var tempDir string var err error // Try data directory first dataDir := os.Getenv("PULSE_DATA_DIR") if dataDir == "" { dataDir = "/etc/pulse" } // Try to create temp dir in data directory tempDir, err = os.MkdirTemp(dataDir, "pulse-update-*") if err != nil { // Fallback to /tmp tempDir, err = os.MkdirTemp("/tmp", "pulse-update-*") if err != nil { // Last resort: current directory tempDir, err = os.MkdirTemp(".", "pulse-update-*") if err != nil { tempErr := fmt.Errorf("failed to create temp directory in any location: %w", err) m.updateStatus("error", 10, "Failed to create temp directory", tempErr) runErr = tempErr m.queue.MarkCompleted(job.ID, tempErr) return tempErr } } } defer os.RemoveAll(tempDir) // Download update tarballPath := filepath.Join(tempDir, "update.tar.gz") downloadBytes, err := m.downloadFile(ctx, req.DownloadURL, tarballPath) if err != nil { downloadErr := fmt.Errorf("failed to download update: %w", err) m.updateStatus("error", 20, "Failed to download update", downloadErr) m.queue.MarkCompleted(job.ID, downloadErr) runErr = downloadErr return runErr } if downloadBytes > 0 { m.updateHistoryEntry(ctx, eventID, func(entry *UpdateHistoryEntry) { entry.DownloadBytes = downloadBytes }) } // Verify checksum if available m.updateStatus("verifying", 30, "Verifying download...") if err := m.verifyChecksum(ctx, req.DownloadURL, tarballPath); err != nil { checksumErr := fmt.Errorf("checksum verification failed: %w", err) m.updateStatus("error", 30, "Failed to verify update checksum", checksumErr) m.queue.MarkCompleted(job.ID, checksumErr) runErr = checksumErr return runErr } log.Info().Msg("Checksum verification passed") m.updateStatus("extracting", 40, "Extracting update...") // Extract tarball extractDir := filepath.Join(tempDir, "extracted") if err := m.extractTarball(tarballPath, extractDir); err != nil { extractErr := fmt.Errorf("failed to extract update: %w", err) m.updateStatus("error", 40, "Failed to extract update", extractErr) m.queue.MarkCompleted(job.ID, extractErr) runErr = extractErr return runErr } m.updateStatus("backing-up", 60, "Creating backup...") // Create backup backupPath, err := m.createBackup() if err != nil { backupErr := fmt.Errorf("failed to create backup: %w", err) m.updateStatus("error", 60, "Failed to create backup", backupErr) m.queue.MarkCompleted(job.ID, backupErr) runErr = backupErr return runErr } log.Info().Str("backup", backupPath).Msg("Created backup") m.updateHistoryEntry(ctx, eventID, func(entry *UpdateHistoryEntry) { entry.BackupPath = backupPath }) m.updateStatus("applying", 80, "Applying update...") // Apply the update files // With the new directory structure (/opt/pulse/bin/), the pulse user has write access log.Info().Msg("Applying update files") if err := m.applyUpdateFiles(extractDir); err != nil { applyErr := fmt.Errorf("failed to apply update: %w", err) m.updateStatus("error", 80, "Failed to apply update", applyErr) m.queue.MarkCompleted(job.ID, applyErr) runErr = applyErr // Attempt to restore backup if restoreErr := m.restoreBackup(backupPath); restoreErr != nil { log.Error().Err(restoreErr).Msg("Failed to restore backup") } return runErr } m.updateStatus("restarting", 95, "Restarting service...") // Mark job as completed m.queue.MarkCompleted(job.ID, nil) // Schedule a clean exit after a short delay - systemd will restart us if !dockerUpdatesAllowed() { go func() { time.Sleep(2 * time.Second) log.Info().Msg("Exiting for restart after update") os.Exit(0) }() } else { log.Info().Msg("Skipping process exit after update (mock/CI mode)") } m.updateStatus("completed", 100, "Update completed, restarting...") return nil } // GetStatus returns the current update status func (m *Manager) GetStatus() UpdateStatus { m.statusMu.RLock() defer m.statusMu.RUnlock() return m.status } // GetCachedUpdateInfo returns the cached update info without making a network request // Returns nil if no cached info is available // Uses the configured or auto-detected channel func (m *Manager) GetCachedUpdateInfo() *UpdateInfo { // Determine which channel to use (same logic as CheckForUpdates) channel := m.config.UpdateChannel if channel == "" { // Try to auto-detect from current version currentInfo, err := GetCurrentVersion() if err == nil && currentInfo.Channel != "" { channel = currentInfo.Channel } } if channel == "" { channel = "stable" } m.statusMu.RLock() defer m.statusMu.RUnlock() return m.checkCache[channel] } // getLatestReleaseForChannel fetches the latest release from GitHub for a specific channel func (m *Manager) getLatestReleaseForChannel(ctx context.Context, channel string, currentVer *Version) (*ReleaseInfo, error) { if channel == "" { channel = "stable" } log.Info(). Str("channel", channel). Str("currentVersion", currentVer.String()). Bool("isPrerelease", currentVer.IsPrerelease()). Msg("Checking for updates") // GitHub API URL (can be overridden for testing) // Always fetch all releases so we can do version-aware filtering baseURL := os.Getenv("PULSE_UPDATE_SERVER") if baseURL == "" { baseURL = "https://api.github.com" } url := baseURL + "/repos/rcourtman/Pulse/releases" req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { return nil, err } // Add headers req.Header.Set("Accept", "application/vnd.github.v3+json") req.Header.Set("User-Agent", "Pulse-Update-Checker") client := &http.Client{Timeout: 30 * time.Second} resp, err := client.Do(req) if err != nil { return nil, fmt.Errorf("failed to fetch releases: %w", err) } defer resp.Body.Close() if resp.StatusCode == http.StatusForbidden { body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) log.Warn(). Str("channel", channel). Str("rateLimitRemaining", resp.Header.Get("X-RateLimit-Remaining")). Str("rateLimitReset", resp.Header.Get("X-RateLimit-Reset")). Msg("GitHub API rate limit encountered, trying RSS fallback") // Try RSS/Atom feed as fallback - doesn't count against rate limits if feedRelease, err := m.getLatestReleaseFromFeed(ctx, channel); err == nil { log.Info().Str("version", feedRelease.TagName).Msg("Got release info from RSS feed fallback") return feedRelease, nil } detail := strings.TrimSpace(string(body)) if detail == "" { detail = resp.Status } return nil, fmt.Errorf("%w: %s", errGitHubRateLimited, detail) } if resp.StatusCode != http.StatusOK { body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) detail := strings.TrimSpace(string(body)) if detail == "" { detail = resp.Status } return nil, fmt.Errorf("GitHub API returned status %d: %s", resp.StatusCode, detail) } var releases []ReleaseInfo if err := json.NewDecoder(resp.Body).Decode(&releases); err != nil { return nil, fmt.Errorf("failed to decode releases: %w", err) } // Find latest release based on channel // RC channel: return newest release (RC or stable), even if not newer than current // Stable channel: return newest stable release, even if not newer than current // The caller will determine if it's actually an update by comparing versions if channel == "rc" { // For RC channel: find newest release (RC or stable) // RC users should see both RCs and stable releases var newestRC *ReleaseInfo var newestStable *ReleaseInfo for i := range releases { // Skip draft releases if releases[i].Draft { log.Debug().Str("tag", releases[i].TagName).Msg("Skipping draft release") continue } releaseVer, err := ParseVersion(releases[i].TagName) if err != nil { log.Debug().Str("tag", releases[i].TagName).Err(err).Msg("Failed to parse release version") continue } if releases[i].Prerelease { // Track newest RC if newestRC == nil { newestRC = &releases[i] } else { newestRCVer, _ := ParseVersion(newestRC.TagName) if releaseVer.IsNewerThan(newestRCVer) { newestRC = &releases[i] } } } else { // Track newest stable if newestStable == nil { newestStable = &releases[i] } else { newestStableVer, _ := ParseVersion(newestStable.TagName) if releaseVer.IsNewerThan(newestStableVer) { newestStable = &releases[i] } } } } // Return the highest version among candidates // Stable versions are considered higher than RCs (4.22.0 > 4.22.0-rc.3) if newestStable != nil && newestRC != nil { stableVer, _ := ParseVersion(newestStable.TagName) rcVer, _ := ParseVersion(newestRC.TagName) if stableVer.IsNewerThan(rcVer) { isUpdate := stableVer.IsNewerThan(currentVer) if isUpdate { log.Info().Str("version", newestStable.TagName).Msg("Found stable update for RC user") } else { log.Info().Str("version", newestStable.TagName).Msg("On latest stable version") } return newestStable, nil } isUpdate := rcVer.IsNewerThan(currentVer) if isUpdate { log.Info().Str("version", newestRC.TagName).Msg("Found RC update") } else { log.Info().Str("version", newestRC.TagName).Msg("On latest RC version") } return newestRC, nil } else if newestStable != nil { isUpdate := newestStable.TagName != currentVer.String() if isUpdate { log.Info().Str("version", newestStable.TagName).Msg("Found stable update for RC user") } else { log.Info().Str("version", newestStable.TagName).Msg("On latest stable version") } return newestStable, nil } else if newestRC != nil { isUpdate := newestRC.TagName != currentVer.String() if isUpdate { log.Info().Str("version", newestRC.TagName).Msg("Found RC update") } else { log.Info().Str("version", newestRC.TagName).Msg("On latest RC version") } return newestRC, nil } } else { // For stable channel: find latest non-prerelease for i := range releases { // Skip draft releases if releases[i].Draft { log.Debug().Str("tag", releases[i].TagName).Msg("Skipping draft release") continue } if releases[i].Prerelease { continue } releaseVer, err := ParseVersion(releases[i].TagName) if err != nil { log.Debug().Str("tag", releases[i].TagName).Err(err).Msg("Failed to parse release version") continue } // Found the latest stable release isUpdate := releaseVer.IsNewerThan(currentVer) if isUpdate { log.Info().Str("version", releases[i].TagName).Msg("Found stable update") } else { log.Info().Str("version", releases[i].TagName).Msg("On latest stable version") } return &releases[i], nil } } // No releases found at all for this channel log.Warn().Str("channel", channel).Msg("No releases found for channel") return nil, fmt.Errorf("no releases found for channel %s", channel) } func (m *Manager) resolveChannel(requested string, currentInfo *VersionInfo) string { if requested != "" { return requested } if m.config != nil && m.config.UpdateChannel != "" { return m.config.UpdateChannel } if currentInfo != nil && currentInfo.Channel != "" { return currentInfo.Channel } return "stable" } // getLatestReleaseFromFeed fetches the latest release from GitHub's Atom feed // This is used as a fallback when the API is rate-limited, as the Atom feed // doesn't count against API rate limits. func (m *Manager) getLatestReleaseFromFeed(ctx context.Context, channel string) (*ReleaseInfo, error) { feedURL := "https://github.com/rcourtman/Pulse/releases.atom" req, err := http.NewRequestWithContext(ctx, "GET", feedURL, nil) if err != nil { return nil, fmt.Errorf("failed to create feed request: %w", err) } req.Header.Set("User-Agent", "Pulse-Update-Checker") client := &http.Client{Timeout: 30 * time.Second} resp, err := client.Do(req) if err != nil { return nil, fmt.Errorf("failed to fetch feed: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { return nil, fmt.Errorf("feed returned status %d", resp.StatusCode) } body, err := io.ReadAll(resp.Body) if err != nil { return nil, fmt.Errorf("failed to read feed: %w", err) } // Parse the Atom feed to extract version tags // The feed format includes entries like: