package updates
import (
"archive/tar"
"compress/gzip"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"os"
"os/exec"
"path/filepath"
"regexp"
"runtime"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
"github.com/rcourtman/pulse-go-rewrite/pkg/edition"
"github.com/rs/zerolog/log"
godisk "github.com/shirou/gopsutil/v4/disk"
)
// 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"`
IsMajorUpgrade bool `json:"isMajorUpgrade"`
Warning string `json:"warning,omitempty"`
// DockerUpdate carries digest-pinned image update commands from the
// license server download broker. Only set for Docker deployments of the
// compiled Pro binary, which cannot self-update in a container and must
// never be pointed at the community rcourtman/pulse image.
DockerUpdate *DockerUpdateCommands `json:"dockerUpdate,omitempty"`
}
var (
errGitHubRateLimited = errors.New("GitHub API rate limit exceeded")
stageDelayOnce sync.Once
stageDelayValue time.Duration
updateHTTPAttempts = 3
updateHTTPBackoff = 300 * time.Millisecond
updateHTTPMaxBackoff = 2 * time.Second
updateDiskUsage = godisk.UsageWithContext
)
const (
defaultUpdateReleaseRepo string = "rcourtman/Pulse"
defaultUpdateAPIBaseURL string = "https://api.github.com"
maxReleaseFeedBytes int64 = 1 << 20 // 1 MiB
maxChecksumFileBytes int64 = 1 << 20 // 1 MiB
maxUpdateDownloadBytes int64 = 512 << 20 // 512 MiB
minUpdateTempFreeBytes int64 = 128 << 20 // 128 MiB
updateBackupSafetyBytes int64 = 32 << 20 // 32 MiB
updateExtractSafetyBytes int64 = 32 << 20 // 32 MiB
maxRetainedUpdateBackups int = 3
)
func updateReleaseRepo() string {
repo := strings.TrimSpace(os.Getenv("PULSE_GITHUB_REPO"))
if repo == "" {
return defaultUpdateReleaseRepo
}
return repo
}
func updateReleaseDownloadPrefix() string {
return fmt.Sprintf("https://github.com/%s/releases/download/", updateReleaseRepo())
}
func updateReleaseAPIPath() string {
return fmt.Sprintf("/repos/%s/releases", updateReleaseRepo())
}
func updateReleaseAPIBaseURL() string {
baseURL := strings.TrimSpace(os.Getenv("PULSE_UPDATE_SERVER"))
if baseURL == "" {
return defaultUpdateAPIBaseURL
}
return baseURL
}
func resolveUpdateReleaseAPIURL() (*url.URL, error) {
baseURL, err := securityutil.NormalizeHTTPBaseURL(updateReleaseAPIBaseURL(), "https")
if err != nil {
return nil, fmt.Errorf("invalid update server base URL: %w", err)
}
target, err := securityutil.ResolveRelativeURL(baseURL, updateReleaseAPIPath())
if err != nil {
return nil, fmt.Errorf("build update release URL: %w", err)
}
return target, nil
}
func updateReleaseFeedURL() string {
return fmt.Sprintf("https://github.com/%s/releases.atom", updateReleaseRepo())
}
func resolveUpdateReleaseFeedURL() (*url.URL, error) {
target, err := securityutil.NormalizeAbsoluteHTTPURL(updateReleaseFeedURL())
if err != nil {
return nil, fmt.Errorf("invalid update release feed URL: %w", err)
}
return target, nil
}
func validateApplyDownloadURL(rawURL string) (*url.URL, error) {
downloadURL, err := securityutil.NormalizeAbsoluteHTTPURL(rawURL)
if err != nil {
return nil, fmt.Errorf("invalid download URL: %w", err)
}
if strings.TrimSpace(os.Getenv("PULSE_UPDATE_SERVER")) != "" {
return downloadURL, nil
}
allowedPrefix, err := securityutil.NormalizeHTTPBaseURL(updateReleaseDownloadPrefix(), "")
if err != nil {
return nil, fmt.Errorf("invalid download URL: %w", err)
}
if downloadURL.Scheme != allowedPrefix.Scheme || !strings.EqualFold(downloadURL.Host, allowedPrefix.Host) {
return nil, fmt.Errorf("invalid download URL")
}
allowedPathPrefix := allowedPrefix.Path
if downloadURL.Path != allowedPathPrefix && !strings.HasPrefix(downloadURL.Path, allowedPathPrefix+"/") {
return nil, fmt.Errorf("invalid download URL")
}
return downloadURL, nil
}
func updateReleaseMigrationURL() string {
return fmt.Sprintf("https://github.com/%s/releases/v4.0.0", updateReleaseRepo())
}
// Manager handles update operations
type Manager struct {
config *config.Config
history *UpdateHistory
status UpdateStatus
statusMu sync.RWMutex
progressMu sync.RWMutex
updateMu sync.Mutex
updateInFlight bool
checkCache map[string]*UpdateInfo // keyed by channel
cacheTime map[string]time.Time // keyed by channel
cacheDuration time.Duration
notesCache *ReleaseNotesInfo // release notes for notesCacheTag (nil when notesCacheMiss)
notesCacheTag string
notesCacheMiss bool
notesCacheTime time.Time
progressChan chan UpdateStatus
sseBroadcast *SSEBroadcaster
lifecycleMu sync.RWMutex
shutdownCh chan struct{}
closeOnce sync.Once
heartbeatWg sync.WaitGroup
closed bool
// proCredentialSource lazily supplies download-broker credentials for the
// compiled Pro binary (SetProUpdateCredentialSource, wired at startup).
// Nil on the community binary.
proCredentialSource func() (ProUpdateCredentials, bool)
}
// ApplyUpdateRequest describes an update request initiated via the API/UI.
type ApplyUpdateRequest struct {
DownloadURL string
Channel string
InitiatedBy InitiatedBy
InitiatedVia InitiatedVia
Notes string
// AllowDowngrade permits installing a target at or below the running
// version. The normal apply path rejects those so a valid-but-older
// release asset URL cannot silently downgrade; sanctioned rollbacks go
// through RollbackToBackup instead.
AllowDowngrade bool
}
// RollbackRequest describes a rollback of a recorded update, restoring the
// retained backup captured before that update was applied.
type RollbackRequest struct {
EventID string
InitiatedBy InitiatedBy
InitiatedVia InitiatedVia
}
// 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),
sseBroadcast: NewSSEBroadcaster(),
shutdownCh: make(chan struct{}),
status: UpdateStatus{
Status: "idle",
UpdatedAt: time.Now().Format(time.RFC3339),
},
}
// Clean up old temp directories and stale update backups from previous runs.
go m.cleanupOldUpdateArtifacts()
// Start heartbeat for SSE connections (every 30 seconds)
m.heartbeatWg.Add(1)
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() {
m.closeOnce.Do(func() {
m.progressMu.Lock()
m.closed = true
close(m.progressChan)
m.progressMu.Unlock()
close(m.shutdownCh)
if m.sseBroadcast != nil {
m.sseBroadcast.Close()
}
m.heartbeatWg.Wait()
})
}
// GetSSEBroadcaster returns the SSE broadcaster
func (m *Manager) GetSSEBroadcaster() *SSEBroadcaster {
m.lifecycleMu.RLock()
defer m.lifecycleMu.RUnlock()
return m.sseBroadcast
}
// AddSSEClient adds a new SSE client for update progress streaming
func (m *Manager) AddSSEClient(w http.ResponseWriter, clientID string) *SSEClient {
m.lifecycleMu.RLock()
defer m.lifecycleMu.RUnlock()
if m.sseBroadcast == nil {
return nil
}
return m.sseBroadcast.AddClient(w, clientID)
}
// RemoveSSEClient removes an SSE client
func (m *Manager) RemoveSSEClient(clientID string) {
m.lifecycleMu.RLock()
defer m.lifecycleMu.RUnlock()
if m.sseBroadcast != nil {
m.sseBroadcast.RemoveClient(clientID)
}
}
// GetCachedStatus returns the last broadcasted status
func (m *Manager) GetSSECachedStatus() (UpdateStatus, time.Time) {
m.lifecycleMu.RLock()
defer m.lifecycleMu.RUnlock()
if m.sseBroadcast == nil {
return UpdateStatus{}, time.Time{}
}
return m.sseBroadcast.GetCachedStatus()
}
// 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 := strings.TrimSpace(channel) != ""
channel = m.resolveChannel(channel, currentInfo)
// 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 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)
}
// The compiled Pro binary checks the license server download broker, never
// GitHub: the public release assets are community builds, and offering one
// here would set up the silent Pro→community downgrade.
if edition.IsPro() {
info, proErr := m.checkProUpdates(ctx, channel, currentInfo, currentVer)
if proErr != nil {
m.updateStatus("error", 0, "Failed to check for Pulse Pro updates", proErr)
return nil, proErr
}
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
}
// 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
}
}
}
isMajorUpgrade := latestVer.Major > currentVer.Major
// Derive prerelease from the parsed version tag (not GitHub metadata) so the
// warning is correct even if the release was published with prerelease=false.
isPrerelease := release.Prerelease || latestVer.IsPrerelease()
info := &UpdateInfo{
Available: latestVer.IsNewerThan(currentVer),
CurrentVersion: currentInfo.Version,
LatestVersion: strings.TrimPrefix(release.TagName, "v"),
ReleaseNotes: release.Body,
ReleaseDate: release.PublishedAt,
DownloadURL: downloadURL,
IsPrerelease: isPrerelease,
IsMajorUpgrade: isMajorUpgrade,
}
info.Warning = updateWarning(info.Available, isMajorUpgrade, isPrerelease, currentVer.Major, latestVer.Major)
// 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
}
// updateWarning derives the user-facing caution attached to an available
// update. Shared by the community (GitHub) and Pro (download broker) checks.
func updateWarning(available, isMajorUpgrade, isPrerelease bool, currentMajor, latestMajor int) string {
switch {
case available && isMajorUpgrade && isPrerelease:
return fmt.Sprintf(
"This is a major version upgrade (v%d → v%d) and a pre-release build. "+
"We strongly recommend installing this as a separate instance rather than upgrading your production installation. "+
"Pre-release builds may contain bugs and are intended for testing.",
currentMajor, latestMajor,
)
case available && isMajorUpgrade:
return fmt.Sprintf(
"This is a major version upgrade (v%d → v%d). Please review the release notes carefully before updating.",
currentMajor, latestMajor,
)
case available && isPrerelease:
return "This is a pre-release build. Pre-release builds are tested but may have rough edges."
default:
return ""
}
}
// ApplyUpdate downloads and applies an update
func (m *Manager) ApplyUpdate(ctx context.Context, req ApplyUpdateRequest) error {
if req.DownloadURL == "" {
return fmt.Errorf("download URL is required")
}
// The separately compiled Pulse Pro binary must never install the public
// community build (it would silently strip Audit, RBAC, Reporting, and
// SSO), so it updates through the license server download broker instead.
// This keys off the compiled edition, not license state: a community
// binary with an active license is still community and updates normally.
isPro := edition.IsPro()
var validatedDownloadURL *url.URL
if isPro {
creds, ok := m.proUpdateCredentials()
if !ok {
return errProUpdateNotActivated()
}
if err := validateProApplyRequestURL(req.DownloadURL, creds.LicenseServerURL); err != nil {
return err
}
} else {
var validationErr error
validatedDownloadURL, validationErr = validateApplyDownloadURL(req.DownloadURL)
if validationErr != nil {
return validationErr
}
req.DownloadURL = validatedDownloadURL.String()
}
// 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 %s", updateReleaseMigrationURL())
}
// Ensure only one update runs at a time.
m.updateMu.Lock()
if m.updateInFlight {
m.updateMu.Unlock()
return fmt.Errorf("update already in progress")
}
m.updateInFlight = true
m.updateMu.Unlock()
defer func() {
m.updateMu.Lock()
m.updateInFlight = false
m.updateMu.Unlock()
}()
m.updateStatus("downloading", 10, "Downloading update...")
channel := m.resolveChannel(req.Channel, currentInfo)
var artifact resolvedUpdateArtifact
if isPro {
// Resolve fresh signed URLs from the broker at apply time: the URLs it
// hands out expire in minutes, so the check-time response is never
// reused here.
var resolveErr error
artifact, resolveErr = m.resolveProUpdateArtifact(ctx, channel)
if resolveErr != nil {
m.updateStatus("error", 10, "Failed to resolve Pulse Pro update", resolveErr)
return resolveErr
}
} else {
targetVersion, validationErr := ValidateApplyTargetVersion(channel, req.DownloadURL)
if validationErr != nil {
m.updateStatus("error", 10, "Update rejected", validationErr)
return validationErr
}
artifact = resolvedUpdateArtifact{downloadURL: req.DownloadURL, version: targetVersion}
}
// A valid release asset URL can still point at an older release than the
// running binary; without this guard the "update" would silently install a
// downgrade. Sanctioned downgrades either restore a retained backup via
// RollbackToBackup or set AllowDowngrade explicitly on the request.
if !req.AllowDowngrade {
if err := ensureApplyTargetIsNewer(currentInfo.Version, artifact.version); err != nil {
m.updateStatus("error", 10, "Update rejected", err)
return err
}
}
initiatedBy := req.InitiatedBy
if initiatedBy == "" {
initiatedBy = InitiatedByUser
}
initiatedVia := req.InitiatedVia
if initiatedVia == "" {
initiatedVia = InitiatedViaAPI
}
start := time.Now()
eventID := m.createHistoryEntry(ctx, UpdateHistoryEntry{
Action: "update",
Channel: channel,
VersionFrom: currentInfo.Version,
VersionTo: artifact.version,
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)
}()
tempDir, err := m.createUpdateTempDir(ctx, minUpdateTempFreeBytes)
if err != nil {
tempErr := fmt.Errorf("failed to create temp directory: %w", err)
m.updateStatus("error", 10, "Failed to create temp directory", tempErr)
runErr = tempErr
return tempErr
}
defer os.RemoveAll(tempDir)
// Download update
tarballPath := filepath.Join(tempDir, "update.tar.gz")
downloadBytes, err := m.downloadFile(ctx, artifact.downloadURL, tarballPath)
if err != nil {
downloadErr := fmt.Errorf("failed to download update: %w", err)
m.updateStatus("error", 20, "Failed to download update", downloadErr)
runErr = downloadErr
return runErr
}
if downloadBytes > 0 {
m.updateHistoryEntry(ctx, eventID, func(entry *UpdateHistoryEntry) {
entry.DownloadBytes = downloadBytes
})
}
// Verify SSHSIG signature against the pinned pulse-installer key. This is
// the same trust root scripts/pulse-auto-update.sh and scripts/install.sh
// already enforce; the in-app updater must not run at a lower bar. The Pro
// broker hands out an explicit signed sidecar URL; the community path
// derives it from the asset URL.
m.updateStatus("verifying", 25, "Verifying signature...")
if artifact.sshsigURL != "" {
err = m.downloadAndVerifySignatureFromURL(ctx, artifact.sshsigURL, tarballPath)
} else {
err = m.downloadAndVerifyReleaseSignature(ctx, validatedDownloadURL, tarballPath)
}
if err != nil {
sigErr := fmt.Errorf("signature verification failed: %w", err)
m.updateStatus("error", 25, "Failed to verify update signature", sigErr)
runErr = sigErr
return runErr
}
log.Info().Msg("Signature verification passed")
// Verify checksum: the Pro broker manifest carries the expected sha256
// inline; the community path discovers a SHA256SUMS manifest next to the
// release asset.
m.updateStatus("verifying", 30, "Verifying download...")
if artifact.sha256 != "" {
err = verifyFileSHA256(tarballPath, artifact.sha256)
} else {
err = m.verifyChecksum(ctx, artifact.downloadURL, tarballPath)
}
if err != nil {
checksumErr := fmt.Errorf("checksum verification failed: %w", err)
m.updateStatus("error", 30, "Failed to verify update checksum", checksumErr)
runErr = checksumErr
return runErr
}
log.Info().Msg("Checksum verification passed")
m.updateStatus("extracting", 40, "Extracting update...")
extractBytes, err := estimateTarballExtractBytes(tarballPath)
if err != nil {
extractErr := fmt.Errorf("failed to size update archive: %w", err)
m.updateStatus("error", 40, "Failed to prepare update extraction", extractErr)
runErr = extractErr
return runErr
}
if err := ensureUpdatePathHasFreeSpace(ctx, tempDir, extractBytes+updateExtractSafetyBytes, "extract update archive"); err != nil {
extractErr := fmt.Errorf("insufficient disk space for update extraction: %w", err)
m.updateStatus("error", 40, "Not enough disk space to extract update", extractErr)
runErr = extractErr
return runErr
}
// 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)
runErr = extractErr
return runErr
}
m.updateStatus("verifying", 50, "Validating new binary...")
// Checksum and signature prove download integrity, not that the binary can
// run on this host or is the approved version. Probe it before anything is
// backed up or replaced, so a bad artifact fails with zero changes applied.
newBinary, err := locateExtractedPulseBinary(extractDir)
if err != nil {
m.updateStatus("error", 50, "Update package is missing the pulse binary", err)
runErr = err
return runErr
}
if err := selfTestNewBinary(ctx, newBinary, extractDir, artifact.version); err != nil {
selfTestErr := fmt.Errorf("new binary failed pre-install validation: %w", err)
m.updateStatus("error", 50, "New binary failed pre-install validation", selfTestErr)
runErr = selfTestErr
return runErr
}
log.Info().Str("version", artifact.version).Msg("New binary passed pre-install self-test")
m.updateStatus("backing-up", 60, "Creating backup...")
// Create backup
backupPath, err := m.createBackup(ctx)
if err != nil {
backupErr := fmt.Errorf("failed to create backup: %w", err)
m.updateStatus("error", 60, "Failed to create backup", 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)
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...")
// 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 {
currentInfo, _ := GetCurrentVersion()
channel := m.resolveChannel("", currentInfo)
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")
releasesURL, err := resolveUpdateReleaseAPIURL()
if err != nil {
return nil, fmt.Errorf("resolve releases request for channel %q: %w", channel, err)
}
client := &http.Client{Timeout: 30 * time.Second}
resp, err := m.getWithRetry(ctx, client, releasesURL, map[string]string{
"Accept": "application/vnd.github.v3+json",
"User-Agent": "Pulse-Update-Checker",
}, "fetch GitHub releases")
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, selecting by version rather than
// list position: GitHub returns releases sorted by created_at, and v5-line
// maintenance releases interleave with v6 releases in the same repo, so
// the most recently published stable is not necessarily the highest one.
// Prerelease channel: newest release (prerelease or stable), even if not newer than current
// Stable channel: newest stable release, even if not newer than current
// The caller will determine if it's actually an update by comparing versions
var newestRC, newestStable *ReleaseInfo
var newestRCVer, newestStableVer *Version
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
}
// A release counts as prerelease when either GitHub metadata or the
// version tag itself says so (guards against metadata set incorrectly)
if releases[i].Prerelease || releaseVer.IsPrerelease() {
if newestRC == nil || releaseVer.IsNewerThan(newestRCVer) {
newestRC = &releases[i]
newestRCVer = releaseVer
}
} else {
if newestStable == nil || releaseVer.IsNewerThan(newestStableVer) {
newestStable = &releases[i]
newestStableVer = releaseVer
}
}
}
if channel == "rc" {
// For the prerelease channel: return the highest version among both
// candidates, so prerelease users are moved forward onto stable once
// the rc line lands (6.0.5 > 6.0.0-rc.7).
if newestStable != nil && (newestRC == nil || newestStableVer.IsNewerThan(newestRCVer)) {
if newestStableVer.IsNewerThan(currentVer) {
log.Info().Str("version", newestStable.TagName).Msg("Found stable update for prerelease user")
} else {
log.Info().Str("version", newestStable.TagName).Msg("On latest stable version")
}
return newestStable, nil
}
if newestRC != nil {
if newestRCVer.IsNewerThan(currentVer) {
log.Info().Str("version", newestRC.TagName).Msg("Found prerelease update")
} else {
log.Info().Str("version", newestRC.TagName).Msg("On latest prerelease version")
}
return newestRC, nil
}
} else if newestStable != nil {
if newestStableVer.IsNewerThan(currentVer) {
log.Info().Str("version", newestStable.TagName).Msg("Found stable update")
} else {
log.Info().Str("version", newestStable.TagName).Msg("On latest stable version")
}
return newestStable, 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 canonical, ok := config.CanonicalUpdateChannel(requested); ok {
return canonical
}
if m.config != nil {
if canonical, ok := config.CanonicalUpdateChannel(m.config.UpdateChannel); ok {
return canonical
}
}
if currentInfo != nil {
if canonical, ok := config.CanonicalUpdateChannel(currentInfo.Channel); ok {
return canonical
}
}
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, err := resolveUpdateReleaseFeedURL()
if err != nil {
return nil, fmt.Errorf("resolve release feed URL: %w", err)
}
client := &http.Client{Timeout: 30 * time.Second}
resp, err := m.getWithRetry(ctx, client, feedURL, map[string]string{
"User-Agent": "Pulse-Update-Checker",
}, "fetch GitHub release feed")
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)
}
if resp.ContentLength > maxReleaseFeedBytes {
return nil, fmt.Errorf("feed response exceeds %d bytes", maxReleaseFeedBytes)
}
body, err := io.ReadAll(io.LimitReader(resp.Body, maxReleaseFeedBytes+1))
if err != nil {
return nil, fmt.Errorf("failed to read feed: %w", err)
}
if int64(len(body)) > maxReleaseFeedBytes {
return nil, fmt.Errorf("feed response exceeds %d bytes", maxReleaseFeedBytes)
}
// Parse the Atom feed to extract version tags
// The feed format includes entries like: Pulse v5.0.0
// We use simple string parsing rather than a full XML parser for minimal deps
content := string(body)
// Find all version tags in the feed (format: "Pulse vX.Y.Z" or "Pulse vX.Y.Z-rc.N")
versionRegex := regexp.MustCompile(`Pulse (v\d+\.\d+\.\d+(?:-[a-zA-Z0-9.]+)?)`)
matches := versionRegex.FindAllStringSubmatch(content, -1)
if len(matches) == 0 {
return nil, fmt.Errorf("no version tags found in feed")
}
// Pick the highest version matching the channel rather than the first
// entry: the feed is publication-ordered and v5-line maintenance releases
// interleave with v6 releases in the same repo.
var best *ReleaseInfo
var bestVer *Version
for _, match := range matches {
if len(match) < 2 {
continue
}
tagName := match[1]
// Parse the version to check if it's a prerelease
ver, err := ParseVersion(tagName)
if err != nil {
continue
}
isPrerelease := ver.IsPrerelease()
// For stable channel, skip prereleases
if channel == "stable" && isPrerelease {
continue
}
if best != nil && !ver.IsNewerThan(bestVer) {
continue
}
best = &ReleaseInfo{
TagName: tagName,
Name: "Pulse " + tagName,
Prerelease: isPrerelease,
// Note: Feed doesn't include full release notes or asset info
// This is just for version checking - actual download still uses known URL patterns
}
bestVer = ver
}
if best != nil {
log.Debug().
Str("tag", best.TagName).
Bool("prerelease", best.Prerelease).
Str("channel", channel).
Msg("Found release from feed")
return best, nil
}
return nil, fmt.Errorf("no suitable release found for channel %s", channel)
}
func (m *Manager) createHistoryEntry(ctx context.Context, entry UpdateHistoryEntry) string {
if m.history == nil {
return ""
}
eventID, err := m.history.CreateEntry(ctx, entry)
if err != nil {
log.Error().Err(err).Msg("Failed to create update history entry")
return ""
}
return eventID
}
func (m *Manager) updateHistoryEntry(ctx context.Context, eventID string, updateFn func(entry *UpdateHistoryEntry)) {
if m.history == nil || eventID == "" {
return
}
if err := m.history.UpdateEntry(ctx, eventID, func(e *UpdateHistoryEntry) error {
updateFn(e)
return nil
}); err != nil {
log.Error().Err(err).Str("event_id", eventID).Msg("Failed to update history entry")
}
}
func (m *Manager) completeHistoryEntry(ctx context.Context, eventID string, status UpdateStatusType, start time.Time, runErr error) {
if m.history == nil || eventID == "" {
return
}
if err := m.history.UpdateEntry(ctx, eventID, func(e *UpdateHistoryEntry) error {
e.Status = status
e.DurationMs = time.Since(start).Milliseconds()
if runErr != nil {
e.Error = &UpdateError{
Message: runErr.Error(),
Code: "update_failed",
}
} else {
e.Error = nil
}
return nil
}); err != nil {
log.Error().Err(err).Str("event_id", eventID).Msg("Failed to finalize history entry")
}
}
var versionInURLRegex = regexp.MustCompile(`v\d+\.\d+\.\d+(?:-[A-Za-z0-9\.]*\d[A-Za-z0-9\.]*)?`)
func inferVersionFromDownloadURL(downloadURL string) string {
if downloadURL == "" {
return ""
}
if parsed, err := securityutil.NormalizeAbsoluteHTTPURL(downloadURL); err == nil {
if match := versionInURLRegex.FindString(parsed.Path); match != "" {
return match
}
if match := versionInURLRegex.FindString(filepath.Base(parsed.Path)); match != "" {
return match
}
}
if match := versionInURLRegex.FindString(downloadURL); match != "" {
return match
}
if match := versionInURLRegex.FindString(filepath.Base(downloadURL)); match != "" {
return match
}
return ""
}
// ValidateApplyTargetVersion extracts and validates the target release from an
// update download URL using the same channel rules enforced by the updater.
func ValidateApplyTargetVersion(channel string, downloadURL string) (string, error) {
targetVersion := inferVersionFromDownloadURL(downloadURL)
if targetVersion == "" {
return "", fmt.Errorf("invalid download URL")
}
targetVer, err := ParseVersion(targetVersion)
if err != nil {
return "", fmt.Errorf("invalid download URL")
}
if channel == "stable" && targetVer.IsPrerelease() {
return "", fmt.Errorf("stable channel cannot install prerelease builds")
}
return targetVersion, nil
}
func isRetryableUpdateStatusCode(statusCode int) bool {
return statusCode == http.StatusRequestTimeout ||
statusCode == http.StatusTooManyRequests ||
statusCode >= http.StatusInternalServerError
}
func isRetryableUpdateRequestError(err error) bool {
if err == nil {
return false
}
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return false
}
var netErr net.Error
if errors.As(err, &netErr) {
return true
}
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
return true
}
msg := strings.ToLower(err.Error())
retryableFragments := []string{
"connection reset",
"connection refused",
"connection aborted",
"broken pipe",
"temporary failure",
"timeout",
"tls handshake timeout",
"http2: server sent goaway",
}
for _, fragment := range retryableFragments {
if strings.Contains(msg, fragment) {
return true
}
}
return false
}
func retryDelayForAttempt(attempt int) time.Duration {
if attempt <= 0 {
return updateHTTPBackoff
}
delay := updateHTTPBackoff
for i := 1; i < attempt; i++ {
delay *= 2
if delay >= updateHTTPMaxBackoff {
return updateHTTPMaxBackoff
}
}
if delay > updateHTTPMaxBackoff {
return updateHTTPMaxBackoff
}
return delay
}
func sleepWithContext(ctx context.Context, delay time.Duration) error {
timer := time.NewTimer(delay)
defer timer.Stop()
select {
case <-ctx.Done():
return ctx.Err()
case <-timer.C:
return nil
}
}
func (m *Manager) getWithRetry(ctx context.Context, client *http.Client, target *url.URL, headers map[string]string, operation string) (*http.Response, error) {
if client == nil {
client = &http.Client{Timeout: 30 * time.Second}
}
if target == nil {
return nil, fmt.Errorf("target URL is required")
}
if updateHTTPAttempts < 1 {
updateHTTPAttempts = 1
}
targetURL := target.String()
for attempt := 1; attempt <= updateHTTPAttempts; attempt++ {
req, err := securityutil.NewValidatedRequestWithContext(ctx, http.MethodGet, target, nil)
if err != nil {
return nil, err
}
for key, value := range headers {
req.Header.Set(key, value)
}
resp, err := client.Do(req)
if err != nil {
if !isRetryableUpdateRequestError(err) || attempt == updateHTTPAttempts {
return nil, err
}
delay := retryDelayForAttempt(attempt)
log.Warn().
Err(err).
Int("attempt", attempt).
Int("maxAttempts", updateHTTPAttempts).
Dur("retryIn", delay).
Str("operation", operation).
Str("url", targetURL).
Msg("Transient update request error; retrying")
if err := sleepWithContext(ctx, delay); err != nil {
return nil, err
}
continue
}
if isRetryableUpdateStatusCode(resp.StatusCode) && attempt < updateHTTPAttempts {
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
resp.Body.Close()
delay := retryDelayForAttempt(attempt)
log.Warn().
Int("statusCode", resp.StatusCode).
Int("attempt", attempt).
Int("maxAttempts", updateHTTPAttempts).
Dur("retryIn", delay).
Str("operation", operation).
Str("url", targetURL).
Msg("Transient update HTTP status; retrying")
if err := sleepWithContext(ctx, delay); err != nil {
return nil, err
}
continue
}
return resp, nil
}
return nil, fmt.Errorf("%s failed after retries", operation)
}
// downloadFile downloads a file from URL to dest
func (m *Manager) downloadFile(ctx context.Context, rawURL, dest string) (int64, error) {
downloadURL, err := securityutil.NormalizeAbsoluteHTTPURL(rawURL)
if err != nil {
return 0, fmt.Errorf("download %q: %w", rawURL, err)
}
client := &http.Client{Timeout: 5 * time.Minute}
resp, err := m.getWithRetry(ctx, client, downloadURL, nil, "download file")
if err != nil {
return 0, fmt.Errorf("download %q: %w", rawURL, err)
}
defer func() {
if closeErr := resp.Body.Close(); closeErr != nil {
log.Warn().Err(closeErr).Str("url", downloadURL.String()).Msg("Failed to close download response body")
}
}()
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("download %q: status %d", rawURL, resp.StatusCode)
}
if updateHTTPAttempts < 1 {
updateHTTPAttempts = 1
}
if resp.ContentLength > maxUpdateDownloadBytes {
return 0, fmt.Errorf("download exceeds maximum size of %d bytes", maxUpdateDownloadBytes)
}
out, err := os.Create(dest)
if err != nil {
return 0, fmt.Errorf("create download destination %q: %w", dest, err)
}
defer func() {
if closeErr := out.Close(); closeErr != nil {
log.Warn().Err(closeErr).Str("path", dest).Msg("Failed to close downloaded file")
}
}()
// Copy with progress updates
written, err := io.Copy(out, io.LimitReader(resp.Body, maxUpdateDownloadBytes+1))
if err != nil {
return 0, fmt.Errorf("copy download response to %q: %w", dest, err)
}
if written > maxUpdateDownloadBytes {
_ = out.Close()
_ = os.Remove(dest)
return 0, fmt.Errorf("download exceeds maximum size of %d bytes", maxUpdateDownloadBytes)
}
return written, nil
}
// verifyChecksum downloads and verifies the SHA256 checksum of a file
func (m *Manager) verifyChecksum(ctx context.Context, tarballURL, tarballPath string) error {
tarballTarget, err := securityutil.NormalizeAbsoluteHTTPURL(tarballURL)
if err != nil {
return fmt.Errorf("invalid tarball URL %q: %w", tarballURL, err)
}
// Common checksum file names used in GitHub releases
checksumNames := []string{"SHA256SUMS", "checksums.txt", "SHA256SUMS.txt"}
var checksumContent string
var checksumErr error
client := &http.Client{Timeout: 30 * time.Second}
// Try each checksum filename
for _, name := range checksumNames {
checksumURL := tarballTarget.ResolveReference(&url.URL{Path: name})
resp, err := m.getWithRetry(ctx, client, checksumURL, nil, "download checksum manifest")
if err != nil {
log.Debug().Err(err).Str("url", checksumURL.String()).Msg("Failed to create checksum request")
continue
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusOK {
if resp.ContentLength > maxChecksumFileBytes {
checksumErr = fmt.Errorf("checksum file %s exceeds %d bytes", name, maxChecksumFileBytes)
continue
}
body, err := io.ReadAll(io.LimitReader(resp.Body, maxChecksumFileBytes+1))
if err != nil {
checksumErr = fmt.Errorf("failed to read checksum file %s: %w", name, err)
continue
}
if int64(len(body)) > maxChecksumFileBytes {
checksumErr = fmt.Errorf("checksum file %s exceeds %d bytes", name, maxChecksumFileBytes)
continue
}
checksumContent = string(body)
log.Info().Str("file", name).Msg("Found checksum file")
break
}
log.Debug().Int("status", resp.StatusCode).Str("url", checksumURL.String()).Msg("Non-OK checksum response, trying next")
resp.Body.Close()
}
if checksumContent == "" {
if checksumErr != nil {
return checksumErr
}
return fmt.Errorf("no checksum file found")
}
// Parse checksum file to find the hash for our tarball
tarballName := filepath.Base(tarballURL)
expectedHash := ""
for _, line := range strings.Split(checksumContent, "\n") {
line = strings.TrimSpace(line)
if line == "" {
continue
}
// Format: "hash filename" or "hash *filename"
parts := strings.Fields(line)
if len(parts) >= 2 {
hash := parts[0]
filename := parts[1]
// Remove leading * if present (indicates binary mode)
filename = strings.TrimPrefix(filename, "*")
if filename == tarballName {
expectedHash = strings.ToLower(hash)
break
}
}
}
if expectedHash == "" {
return fmt.Errorf("checksum not found for %s in checksum file", tarballName)
}
// Compute SHA256 of downloaded file
file, err := os.Open(tarballPath)
if err != nil {
return fmt.Errorf("failed to open tarball for checksum: %w", err)
}
defer func() {
if closeErr := file.Close(); closeErr != nil {
log.Warn().Err(closeErr).Str("path", tarballPath).Msg("Failed to close tarball after checksum verification")
}
}()
hash := sha256.New()
if _, err := io.Copy(hash, file); err != nil {
return fmt.Errorf("failed to compute checksum: %w", err)
}
actualHash := hex.EncodeToString(hash.Sum(nil))
// Compare hashes
if actualHash != expectedHash {
checksumErr = fmt.Errorf("checksum mismatch: expected %s, got %s", expectedHash, actualHash)
log.Error().
Str("expected", expectedHash).
Str("actual", actualHash).
Msg("Checksum verification failed")
return checksumErr
}
log.Info().
Str("hash", actualHash).
Msg("Checksum verified successfully")
return nil
}
// extractTarball extracts a gzipped tarball
func (m *Manager) extractTarball(src, dest string) error {
file, err := os.Open(src)
if err != nil {
return fmt.Errorf("open tarball %q: %w", src, err)
}
defer func() {
if closeErr := file.Close(); closeErr != nil {
log.Warn().Err(closeErr).Str("path", src).Msg("Failed to close tarball file")
}
}()
gzr, err := gzip.NewReader(file)
if err != nil {
return fmt.Errorf("open gzip reader for %q: %w", src, err)
}
defer func() {
if closeErr := gzr.Close(); closeErr != nil {
log.Warn().Err(closeErr).Str("path", src).Msg("Failed to close gzip reader")
}
}()
tr := tar.NewReader(gzr)
for {
header, err := tr.Next()
if err == io.EOF {
break
}
if err != nil {
return fmt.Errorf("read tarball entry from %q: %w", src, err)
}
// Sanitize the path to prevent directory traversal attacks
cleanName := filepath.Clean(header.Name)
// Check for path traversal attempts
if strings.Contains(cleanName, "..") || filepath.IsAbs(cleanName) {
return fmt.Errorf("unsafe path in archive: %s", header.Name)
}
// Ensure the target path is within the destination directory
target := filepath.Join(dest, cleanName)
if !strings.HasPrefix(target, filepath.Clean(dest)+string(os.PathSeparator)) && target != filepath.Clean(dest) {
return fmt.Errorf("path escapes destination directory: %s", header.Name)
}
switch header.Typeflag {
case tar.TypeDir:
if err := os.MkdirAll(target, 0755); err != nil {
return fmt.Errorf("create archive directory %q: %w", target, err)
}
case tar.TypeReg:
if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
return fmt.Errorf("create parent directory for archive file %q: %w", target, err)
}
out, err := os.OpenFile(target, os.O_CREATE|os.O_RDWR, os.FileMode(header.Mode))
if err != nil {
return fmt.Errorf("open archive target file %q: %w", target, err)
}
if _, err := io.Copy(out, tr); err != nil {
if closeErr := out.Close(); closeErr != nil {
return errors.Join(
fmt.Errorf("write archive file %q: %w", target, err),
fmt.Errorf("close archive file %q after write failure: %w", target, closeErr),
)
}
return fmt.Errorf("write archive file %q: %w", target, err)
}
if closeErr := out.Close(); closeErr != nil {
return fmt.Errorf("close archive file %q: %w", target, closeErr)
}
}
}
return nil
}
// createBackup creates a backup of the current installation.
func (m *Manager) createBackup(ctx context.Context) (string, error) {
timestamp := time.Now().Format("20060102-150405")
if err := m.pruneRetainedUpdateBackups(ctx); err != nil {
log.Warn().Err(err).Msg("Failed to prune retained update backups before creating a new backup")
}
backupBytes, err := estimateUpdateBackupBytes()
if err != nil {
return "", fmt.Errorf("estimate backup size: %w", err)
}
requiredBytes := backupBytes + updateBackupSafetyBytes
backupRoot, err := selectUpdateRootWithSpace(ctx, managedUpdateBackupRoots(), requiredBytes)
if err != nil {
return "", fmt.Errorf("select backup directory: %w", err)
}
backupDir := managedUpdateBackupPath(backupRoot, timestamp)
if err := os.MkdirAll(backupDir, 0755); err != nil {
return "", fmt.Errorf("create backup directory %q: %w", backupDir, err)
}
// Backup important directories
dirsToBackup := []string{"data", "config"}
pulseDir := os.Getenv("PULSE_INSTALL_DIR")
if pulseDir == "" {
pulseDir = "/opt/pulse"
}
for _, dir := range dirsToBackup {
src := filepath.Join(pulseDir, dir)
dest := filepath.Join(backupDir, dir)
if _, err := os.Stat(src); err == nil {
if err := m.copyDirSafe(src, dest); err != nil {
log.Warn().Str("dir", dir).Err(err).Msg("Failed to backup directory")
}
}
}
// Backup .env file
envSrc := filepath.Join(pulseDir, ".env")
if _, err := os.Stat(envSrc); err == nil {
envDest := filepath.Join(backupDir, ".env")
if err := m.copyFileSafe(envSrc, envDest); err != nil {
log.Warn().Err(err).Msg("Failed to backup .env file")
}
}
// Backup the pulse binary itself
binaryPath, err := os.Executable()
if err == nil {
binaryDest := filepath.Join(backupDir, "pulse")
if err := m.copyFileSafe(binaryPath, binaryDest); err != nil {
log.Warn().Err(err).Msg("Failed to backup pulse binary")
} else {
log.Info().Str("binary", binaryPath).Msg("Backed up pulse binary")
}
}
// Backup VERSION file if it exists
versionSrc := filepath.Join(pulseDir, "VERSION")
if _, err := os.Stat(versionSrc); err == nil {
versionDest := filepath.Join(backupDir, "VERSION")
if err := m.copyFileSafe(versionSrc, versionDest); err != nil {
log.Warn().Err(err).Msg("Failed to backup VERSION file")
}
}
if err := m.pruneRetainedUpdateBackups(ctx); err != nil {
log.Warn().Err(err).Msg("Failed to prune retained update backups after creating a new backup")
}
return backupDir, nil
}
// ensureApplyTargetIsNewer rejects update targets at or below the running
// version so a valid-but-older release asset URL cannot silently downgrade.
// Versions that do not parse as semver (development builds) are left to the
// existing URL and channel validation.
func ensureApplyTargetIsNewer(currentVersion, targetVersion string) error {
current, err := ParseVersion(currentVersion)
if err != nil {
return nil
}
target, err := ParseVersion(targetVersion)
if err != nil {
return nil
}
if target.Compare(current) <= 0 {
return fmt.Errorf("target version %s is not newer than the running version %s; use the update history rollback to return to an earlier version", target.String(), current.String())
}
return nil
}
// validateRetainedBackupDir confirms a history-recorded backup path still
// points at an existing managed update backup directory before anything is
// restored from it. The history file is server-owned state, but the path is
// re-checked against the managed backup roots so a corrupted or hand-edited
// history entry cannot direct a restore from an arbitrary filesystem location.
func validateRetainedBackupDir(raw string) (string, error) {
cleaned := filepath.Clean(strings.TrimSpace(raw))
managed := false
for _, root := range managedUpdateBackupRoots() {
if filepath.Dir(cleaned) != root {
continue
}
if strings.HasPrefix(filepath.Base(cleaned), managedUpdateBackupPrefix(root)) {
managed = true
break
}
}
if !managed {
return "", fmt.Errorf("backup path is not a managed update backup: %s", cleaned)
}
info, err := os.Stat(cleaned)
if err != nil {
if os.IsNotExist(err) {
return "", fmt.Errorf("backup no longer exists on disk: %s", cleaned)
}
return "", fmt.Errorf("stat backup directory %q: %w", cleaned, err)
}
if !info.IsDir() {
return "", fmt.Errorf("backup path is not a directory: %s", cleaned)
}
return cleaned, nil
}
// RollbackToBackup restores the retained backup recorded on an update history
// entry, records the rollback as its own history entry, and restarts through
// the same exit-for-systemd path as ApplyUpdate. It is a purely local
// restore: no release download, broker call, or edition gate is involved, so
// it behaves identically on community and Pro binaries.
func (m *Manager) RollbackToBackup(ctx context.Context, req RollbackRequest) error {
if m.history == nil {
return fmt.Errorf("update history is not available")
}
eventID := strings.TrimSpace(req.EventID)
if eventID == "" {
return fmt.Errorf("event ID is required")
}
source, err := m.history.GetEntry(eventID)
if err != nil {
return fmt.Errorf("update history entry not found: %s", eventID)
}
if strings.TrimSpace(source.BackupPath) == "" {
return fmt.Errorf("no retained backup for this update; it may have been pruned by backup retention")
}
backupDir, err := validateRetainedBackupDir(source.BackupPath)
if err != nil {
return err
}
currentInfo, _ := GetCurrentVersion()
if currentInfo.IsDocker {
return fmt.Errorf("rollback cannot be applied in Docker environment")
}
// Rollback and update share the single in-flight slot: restoring a backup
// while an update is replacing the same files would corrupt both.
m.updateMu.Lock()
if m.updateInFlight {
m.updateMu.Unlock()
return fmt.Errorf("update already in progress")
}
m.updateInFlight = true
m.updateMu.Unlock()
defer func() {
m.updateMu.Lock()
m.updateInFlight = false
m.updateMu.Unlock()
}()
initiatedBy := req.InitiatedBy
if initiatedBy == "" {
initiatedBy = InitiatedByUser
}
initiatedVia := req.InitiatedVia
if initiatedVia == "" {
initiatedVia = InitiatedViaAPI
}
m.updateStatus("restoring", 20, fmt.Sprintf("Restoring Pulse %s from backup...", source.VersionFrom))
start := time.Now()
rollbackEventID := m.createHistoryEntry(ctx, UpdateHistoryEntry{
Action: "rollback",
Channel: source.Channel,
VersionFrom: currentInfo.Version,
VersionTo: source.VersionFrom,
DeploymentType: currentInfo.DeploymentType,
InitiatedBy: initiatedBy,
InitiatedVia: initiatedVia,
Status: StatusInProgress,
BackupPath: backupDir,
RelatedEventID: source.EventID,
})
var runErr error
defer func() {
if rollbackEventID == "" {
return
}
status := StatusSuccess
if runErr != nil {
status = StatusFailed
}
m.completeHistoryEntry(ctx, rollbackEventID, status, start, runErr)
}()
if err := m.restoreBackup(backupDir); err != nil {
restoreErr := fmt.Errorf("failed to restore backup: %w", err)
m.updateStatus("error", 40, "Failed to restore backup", restoreErr)
runErr = restoreErr
return restoreErr
}
// The update this backup predates is no longer the running install.
m.updateHistoryEntry(ctx, eventID, func(entry *UpdateHistoryEntry) {
entry.Status = StatusRolledBack
})
m.updateStatus("restarting", 95, "Restarting service...")
// 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 rollback")
os.Exit(0)
}()
} else {
log.Info().Msg("Skipping process exit after rollback (mock/CI mode)")
}
m.updateStatus("completed", 100, "Rollback completed, restarting...")
return nil
}
// restoreBackup restores from a backup
func (m *Manager) restoreBackup(backupDir string) error {
pulseDir := os.Getenv("PULSE_INSTALL_DIR")
if pulseDir == "" {
pulseDir = "/opt/pulse"
}
// Restore directories
dirsToRestore := []string{"data", "config"}
for _, dir := range dirsToRestore {
src := filepath.Join(backupDir, dir)
dest := filepath.Join(pulseDir, dir)
if _, err := os.Stat(src); err == nil {
// Remove existing directory first
if err := os.RemoveAll(dest); err != nil {
return fmt.Errorf("failed to remove existing %s: %w", dir, err)
}
if err := m.copyDirSafe(src, dest); err != nil {
return fmt.Errorf("failed to restore %s: %w", dir, err)
}
}
}
// Restore .env
envSrc := filepath.Join(backupDir, ".env")
if _, err := os.Stat(envSrc); err == nil {
envDest := filepath.Join(pulseDir, ".env")
if err := m.copyFileSafe(envSrc, envDest); err != nil {
return fmt.Errorf("failed to restore .env: %w", err)
}
}
// Restore the pulse binary if it exists in backup
binarySrc := filepath.Join(backupDir, "pulse")
if _, err := os.Stat(binarySrc); err == nil {
binaryPath, err := os.Executable()
if err == nil {
// Create temp copy first, then atomic rename
tempBinary := binaryPath + ".restored"
if err := m.copyFileSafe(binarySrc, tempBinary); err != nil {
return fmt.Errorf("failed to restore pulse binary: %w", err)
}
if err := os.Chmod(tempBinary, 0755); err != nil {
return fmt.Errorf("failed to set binary permissions: %w", err)
}
if err := os.Rename(tempBinary, binaryPath); err != nil {
return fmt.Errorf("failed to replace binary: %w", err)
}
log.Info().Str("binary", binaryPath).Msg("Restored pulse binary")
}
}
// Restore VERSION file if it exists in backup
versionSrc := filepath.Join(backupDir, "VERSION")
if _, err := os.Stat(versionSrc); err == nil {
versionDest := filepath.Join(pulseDir, "VERSION")
if err := m.copyFileSafe(versionSrc, versionDest); err != nil {
log.Warn().Err(err).Msg("Failed to restore VERSION file")
}
}
return nil
}
// applyUpdateFiles copies update files to the installation directory
// locateExtractedPulseBinary finds the pulse binary inside an extracted
// update, checking both the legacy root layout and the bin/ layout.
func locateExtractedPulseBinary(extractDir string) (string, error) {
pulseBinary := filepath.Join(extractDir, "pulse")
if _, err := os.Stat(pulseBinary); err == nil {
return pulseBinary, nil
}
pulseBinary = filepath.Join(extractDir, "bin", "pulse")
if _, err := os.Stat(pulseBinary); err != nil {
return "", fmt.Errorf("pulse binary not found in extract (checked both / and /bin/): %w", err)
}
return pulseBinary, nil
}
func (m *Manager) applyUpdateFiles(extractDir string) error {
pulseBinary, err := locateExtractedPulseBinary(extractDir)
if err != nil {
return err
}
// Detect where the current binary is running from
binaryPath, err := os.Executable()
if err != nil {
// Fallback to default location
binaryPath = "/usr/local/bin/pulse"
}
// Copy the pulse binary to a temporary location first, then move atomically
tempBinary := binaryPath + ".new"
cmd := exec.Command("cp", pulseBinary, tempBinary)
if err := cmd.Run(); err != nil {
return fmt.Errorf("failed to copy pulse binary: %w", err)
}
// Make it executable
if err := os.Chmod(tempBinary, 0755); err != nil {
return fmt.Errorf("failed to set permissions: %w", err)
}
// Atomically replace the old binary with the new one
if err := os.Rename(tempBinary, binaryPath); err != nil {
// If rename fails (cross-device), try mv command
cmd = exec.Command("mv", "-f", tempBinary, binaryPath)
if err := cmd.Run(); err != nil {
return fmt.Errorf("failed to replace pulse binary: %w", err)
}
}
// Frontend is now embedded in the binary as of v4.2.2+
// No need to copy frontend files separately
// The new binary contains everything needed
// Copy VERSION file if it exists (to both locations for compatibility)
versionSrc := filepath.Join(extractDir, "VERSION")
if _, err := os.Stat(versionSrc); err == nil {
// Copy to /opt/pulse
cmd = exec.Command("cp", versionSrc, "/opt/pulse/VERSION")
if err := cmd.Run(); err != nil {
log.Debug().Err(err).Msg("Failed to copy VERSION to /opt/pulse")
}
// Copy to binary directory
binaryDir := filepath.Dir(binaryPath)
cmd = exec.Command("cp", versionSrc, filepath.Join(binaryDir, "VERSION"))
if err := cmd.Run(); err != nil {
log.Warn().Err(err).Msg("Failed to copy VERSION file")
}
}
// Deploy agent installation scripts from tarball
scriptsDir := filepath.Join(extractDir, "scripts")
if _, err := os.Stat(scriptsDir); err == nil {
destScriptsDir := "/opt/pulse/scripts"
if err := os.MkdirAll(destScriptsDir, 0755); err != nil {
log.Warn().Err(err).Msg("Failed to create scripts directory")
} else {
// List of agent scripts to deploy
agentScripts := []string{
"install-container-agent.sh",
"install-docker.sh",
"install.sh",
"install.ps1",
}
deployed := 0
for _, script := range agentScripts {
srcPath := filepath.Join(scriptsDir, script)
if _, err := os.Stat(srcPath); err == nil {
destPath := filepath.Join(destScriptsDir, script)
cmd = exec.Command("cp", srcPath, destPath)
if err := cmd.Run(); err != nil {
log.Warn().Err(err).Str("script", script).Msg("Failed to copy agent script")
continue
}
if err := os.Chmod(destPath, 0755); err != nil {
log.Warn().Err(err).Str("script", script).Msg("Failed to set script permissions")
}
deployed++
}
}
if deployed > 0 {
log.Info().Int("count", deployed).Msg("Deployed agent installation scripts")
}
}
}
// Deploy agent binaries from tarball (for serving to remote hosts)
binDir := filepath.Join(extractDir, "bin")
if _, err := os.Stat(binDir); err == nil {
destBinDir := "/opt/pulse/bin"
if err := os.MkdirAll(destBinDir, 0755); err != nil {
log.Warn().Err(err).Msg("Failed to create bin directory")
} else {
// Copy agent binaries (pulse-agent-* artifacts)
entries, err := os.ReadDir(binDir)
if err == nil {
agentBinariesDeployed := 0
for _, entry := range entries {
name := entry.Name()
// Skip the main pulse binary (already handled above) and directories
if entry.IsDir() || name == "pulse" {
continue
}
// Copy agent binaries
if strings.HasPrefix(name, "pulse-agent-") {
srcPath, err := securityutil.JoinStorageLeaf(binDir, name)
if err != nil {
log.Warn().Err(err).Str("binary", name).Msg("Skipping invalid agent binary path")
continue
}
destPath, err := securityutil.JoinStorageLeaf(destBinDir, name)
if err != nil {
log.Warn().Err(err).Str("binary", name).Msg("Skipping invalid destination binary path")
continue
}
cmd = exec.Command("cp", "-a", srcPath, destPath)
if err := cmd.Run(); err != nil {
log.Warn().Err(err).Str("binary", name).Msg("Failed to copy agent binary")
continue
}
// Set executable permission (skip for symlinks)
if info, err := os.Lstat(destPath); err == nil && info.Mode()&os.ModeSymlink == 0 {
if err := os.Chmod(destPath, 0755); err != nil {
log.Warn().Err(err).Str("binary", name).Msg("Failed to set binary permissions")
}
}
agentBinariesDeployed++
}
}
if agentBinariesDeployed > 0 {
log.Info().Int("count", agentBinariesDeployed).Msg("Deployed agent binaries")
}
}
}
}
// Set ownership if /opt/pulse exists
if _, err := os.Stat("/opt/pulse"); err == nil {
cmd = exec.Command("chown", "-R", "pulse:pulse", "/opt/pulse")
if err := cmd.Run(); err != nil {
log.Warn().Err(err).Msg("Failed to set ownership")
}
}
return nil
}
// updateStatus updates the current status
func (m *Manager) updateStatus(status string, progress int, message string, err ...error) {
m.statusMu.Lock()
m.status = UpdateStatus{
Status: status,
Progress: progress,
Message: message,
UpdatedAt: time.Now().Format(time.RFC3339),
}
// If error provided, sanitize and add to status
if len(err) > 0 && err[0] != nil {
m.status.Error = sanitizeError(err[0])
}
statusCopy := m.status
m.statusMu.Unlock()
m.lifecycleMu.RLock()
if m.closed {
m.lifecycleMu.RUnlock()
if delay := statusDelayForStage(status); delay > 0 {
time.Sleep(delay)
}
return
}
// Send to progress channel (non-blocking) for WebSocket compatibility
m.progressMu.RLock()
if !m.closed {
select {
case m.progressChan <- statusCopy:
default:
}
}
m.progressMu.RUnlock()
// Broadcast to SSE clients
if m.sseBroadcast != nil {
m.sseBroadcast.Broadcast(statusCopy)
}
m.lifecycleMu.RUnlock()
if delay := statusDelayForStage(status); delay > 0 {
time.Sleep(delay)
}
}
// sseHeartbeatLoop sends periodic heartbeats to SSE clients
func (m *Manager) sseHeartbeatLoop() {
defer m.heartbeatWg.Done()
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case <-m.shutdownCh:
return
case <-ticker.C:
m.lifecycleMu.RLock()
broadcaster := m.sseBroadcast
m.lifecycleMu.RUnlock()
if broadcaster != nil {
broadcaster.SendHeartbeat()
}
}
}
}
// sanitizeError removes potentially sensitive information from error messages
func sanitizeError(err error) string {
if err == nil {
return ""
}
errMsg := err.Error()
// Cap length to prevent extremely long error messages
maxLen := 500
if len(errMsg) > maxLen {
errMsg = errMsg[:maxLen] + "..."
}
return errMsg
}
func statusDelayForStage(status string) time.Duration {
delay := configuredStageDelay()
if delay == 0 {
return 0
}
switch status {
case "downloading", "verifying", "extracting", "backing-up", "applying":
return delay
default:
return 0
}
}
func configuredStageDelay() time.Duration {
stageDelayOnce.Do(func() {
value := strings.TrimSpace(os.Getenv("PULSE_UPDATE_STAGE_DELAY_MS"))
if value == "" {
return
}
ms, err := strconv.Atoi(value)
if err != nil || ms <= 0 {
log.Warn().Str("value", value).Msg("Invalid PULSE_UPDATE_STAGE_DELAY_MS, ignoring")
return
}
stageDelayValue = time.Duration(ms) * time.Millisecond
})
return stageDelayValue
}
func resolveUpdateDataDir() string {
return config.ResolveRuntimeDataDir("")
}
type retainedUpdateBackup struct {
Path string
ModTime time.Time
}
func managedUpdateBackupRoots() []string {
seen := make(map[string]struct{}, 2)
roots := make([]string, 0, 2)
for _, root := range []string{resolveUpdateDataDir(), "/tmp"} {
cleaned := strings.TrimSpace(filepath.Clean(root))
if cleaned == "" {
continue
}
if _, exists := seen[cleaned]; exists {
continue
}
seen[cleaned] = struct{}{}
roots = append(roots, cleaned)
}
return roots
}
func managedUpdateBackupPath(root string, timestamp string) string {
cleaned := filepath.Clean(strings.TrimSpace(root))
if cleaned == "/tmp" {
return filepath.Join(cleaned, fmt.Sprintf("pulse-backup-%s", timestamp))
}
return filepath.Join(cleaned, fmt.Sprintf("backup-%s", timestamp))
}
func managedUpdateBackupPrefix(root string) string {
if filepath.Clean(strings.TrimSpace(root)) == "/tmp" {
return "pulse-backup-"
}
return "backup-"
}
func managedUpdateTempRoots() []string {
return []string{resolveUpdateDataDir(), "/tmp", "."}
}
func formatUpdateBytes(bytes int64) string {
if bytes <= 0 {
return "0 B"
}
units := []string{"B", "KiB", "MiB", "GiB", "TiB"}
value := float64(bytes)
unit := units[0]
for i := 1; i < len(units) && value >= 1024; i++ {
value /= 1024
unit = units[i]
}
if unit == "B" {
return fmt.Sprintf("%d %s", bytes, unit)
}
return fmt.Sprintf("%.1f %s", value, unit)
}
func resolveDiskUsagePath(path string) string {
current := filepath.Clean(strings.TrimSpace(path))
if current == "" {
return "."
}
for {
if _, err := os.Stat(current); err == nil {
return current
}
parent := filepath.Dir(current)
if parent == current {
return current
}
current = parent
}
}
func updatePathFreeBytes(ctx context.Context, path string) (int64, error) {
usagePath := resolveDiskUsagePath(path)
usage, err := updateDiskUsage(ctx, usagePath)
if err != nil {
return 0, fmt.Errorf("inspect disk usage for %q: %w", usagePath, err)
}
return int64(usage.Free), nil
}
func ensureUpdatePathHasFreeSpace(ctx context.Context, path string, requiredBytes int64, activity string) error {
if requiredBytes <= 0 {
return nil
}
freeBytes, err := updatePathFreeBytes(ctx, path)
if err != nil {
return err
}
if freeBytes >= requiredBytes {
return nil
}
return fmt.Errorf(
"%s needs %s free at %s, only %s available",
activity,
formatUpdateBytes(requiredBytes),
resolveDiskUsagePath(path),
formatUpdateBytes(freeBytes),
)
}
func selectUpdateRootWithSpace(ctx context.Context, candidates []string, requiredBytes int64) (string, error) {
type insufficientSpaceCandidate struct {
Path string
Free int64
Err error
}
seen := make(map[string]struct{}, len(candidates))
failures := make([]insufficientSpaceCandidate, 0, len(candidates))
for _, candidate := range candidates {
cleaned := strings.TrimSpace(filepath.Clean(candidate))
if cleaned == "" {
continue
}
if _, exists := seen[cleaned]; exists {
continue
}
seen[cleaned] = struct{}{}
if cleaned != "." {
if err := os.MkdirAll(cleaned, 0755); err != nil {
failures = append(failures, insufficientSpaceCandidate{Path: cleaned, Err: err})
continue
}
}
freeBytes, err := updatePathFreeBytes(ctx, cleaned)
if err != nil {
failures = append(failures, insufficientSpaceCandidate{Path: cleaned, Err: err})
continue
}
if freeBytes >= requiredBytes {
return cleaned, nil
}
failures = append(failures, insufficientSpaceCandidate{Path: cleaned, Free: freeBytes})
}
parts := make([]string, 0, len(failures))
for _, failure := range failures {
if failure.Err != nil {
parts = append(parts, fmt.Sprintf("%s unavailable (%v)", failure.Path, failure.Err))
continue
}
parts = append(parts, fmt.Sprintf("%s has %s free", failure.Path, formatUpdateBytes(failure.Free)))
}
if len(parts) == 0 {
return "", fmt.Errorf("no writable update directories available")
}
return "", fmt.Errorf("need %s free, candidates: %s", formatUpdateBytes(requiredBytes), strings.Join(parts, "; "))
}
func (m *Manager) createUpdateTempDir(ctx context.Context, requiredBytes int64) (string, error) {
tempRoot, err := selectUpdateRootWithSpace(ctx, managedUpdateTempRoots(), requiredBytes)
if err != nil {
return "", err
}
tempDir, err := os.MkdirTemp(tempRoot, "pulse-update-*")
if err != nil {
return "", fmt.Errorf("create temp directory in %q: %w", tempRoot, err)
}
return tempDir, nil
}
func estimatePathCopyBytes(path string) (int64, error) {
info, err := os.Lstat(path)
if err != nil {
if os.IsNotExist(err) {
return 0, nil
}
return 0, err
}
if info.Mode()&os.ModeSymlink != 0 {
return 0, nil
}
if !info.IsDir() {
return info.Size(), nil
}
var total int64
err = filepath.Walk(path, func(current string, currentInfo os.FileInfo, walkErr error) error {
if walkErr != nil {
return walkErr
}
if currentInfo.Mode()&os.ModeSymlink != 0 {
if currentInfo.IsDir() {
return filepath.SkipDir
}
return nil
}
if currentInfo.IsDir() {
return nil
}
total += currentInfo.Size()
return nil
})
if err != nil {
return 0, err
}
return total, nil
}
func estimateUpdateBackupBytes() (int64, error) {
pulseDir := strings.TrimSpace(os.Getenv("PULSE_INSTALL_DIR"))
if pulseDir == "" {
pulseDir = "/opt/pulse"
}
paths := []string{
filepath.Join(pulseDir, "data"),
filepath.Join(pulseDir, "config"),
filepath.Join(pulseDir, ".env"),
filepath.Join(pulseDir, "VERSION"),
}
if binaryPath, err := os.Executable(); err == nil && strings.TrimSpace(binaryPath) != "" {
paths = append(paths, binaryPath)
}
var total int64
for _, path := range paths {
size, err := estimatePathCopyBytes(path)
if err != nil {
return 0, fmt.Errorf("measure %q: %w", path, err)
}
total += size
}
return total, nil
}
func estimateTarballExtractBytes(path string) (int64, error) {
file, err := os.Open(path)
if err != nil {
return 0, fmt.Errorf("open tarball %q: %w", path, err)
}
defer file.Close()
gzReader, err := gzip.NewReader(file)
if err != nil {
return 0, fmt.Errorf("open gzip reader for %q: %w", path, err)
}
defer gzReader.Close()
tarReader := tar.NewReader(gzReader)
var total int64
for {
header, err := tarReader.Next()
if err == io.EOF {
break
}
if err != nil {
return 0, fmt.Errorf("walk tarball %q: %w", path, err)
}
switch header.Typeflag {
case tar.TypeReg, tar.TypeRegA, tar.TypeGNUSparse:
if header.Size > 0 {
total += header.Size
}
}
}
return total, nil
}
func (m *Manager) collectManagedUpdateBackups() ([]retainedUpdateBackup, error) {
backups := make([]retainedUpdateBackup, 0)
for _, root := range managedUpdateBackupRoots() {
entries, err := os.ReadDir(root)
if err != nil {
if os.IsNotExist(err) {
continue
}
return nil, fmt.Errorf("read backup root %q: %w", root, err)
}
prefix := managedUpdateBackupPrefix(root)
for _, entry := range entries {
if !entry.IsDir() || !strings.HasPrefix(entry.Name(), prefix) {
continue
}
fullPath, err := securityutil.JoinStorageLeaf(root, entry.Name())
if err != nil {
log.Debug().Err(err).Str("root", root).Str("entry", entry.Name()).Msg("Skipping invalid update backup leaf")
continue
}
info, err := entry.Info()
if err != nil {
continue
}
backups = append(backups, retainedUpdateBackup{
Path: fullPath,
ModTime: info.ModTime(),
})
}
}
sort.Slice(backups, func(i, j int) bool {
if backups[i].ModTime.Equal(backups[j].ModTime) {
return backups[i].Path > backups[j].Path
}
return backups[i].ModTime.After(backups[j].ModTime)
})
return backups, nil
}
func appendRetentionNote(notes string, prunedPath string) string {
retentionNote := fmt.Sprintf("Rollback backup pruned by retention: %s", prunedPath)
if strings.Contains(notes, retentionNote) {
return notes
}
trimmed := strings.TrimSpace(notes)
if trimmed == "" {
return retentionNote
}
return trimmed + "\n" + retentionNote
}
func (m *Manager) clearPrunedBackupHistoryReference(ctx context.Context, prunedPath string) {
if m.history == nil {
return
}
for _, entry := range m.history.ListEntries(HistoryFilter{}) {
if filepath.Clean(entry.BackupPath) != filepath.Clean(prunedPath) {
continue
}
if err := m.history.UpdateEntry(ctx, entry.EventID, func(existing *UpdateHistoryEntry) error {
existing.BackupPath = ""
existing.Notes = appendRetentionNote(existing.Notes, prunedPath)
return nil
}); err != nil {
log.Warn().
Err(err).
Str("event_id", entry.EventID).
Str("backup", prunedPath).
Msg("Failed to clear pruned backup reference from update history")
}
}
}
func (m *Manager) pruneRetainedUpdateBackups(ctx context.Context) error {
backups, err := m.collectManagedUpdateBackups()
if err != nil {
return err
}
if len(backups) <= maxRetainedUpdateBackups {
return nil
}
for _, backup := range backups[maxRetainedUpdateBackups:] {
if err := os.RemoveAll(backup.Path); err != nil {
return fmt.Errorf("remove retained backup %q: %w", backup.Path, err)
}
m.clearPrunedBackupHistoryReference(ctx, backup.Path)
log.Info().Str("path", backup.Path).Msg("Pruned retained update backup")
}
return nil
}
func (m *Manager) cleanupOldUpdateArtifacts() {
m.cleanupOldTempDirs()
if err := m.pruneRetainedUpdateBackups(context.Background()); err != nil {
log.Debug().Err(err).Msg("Failed to prune retained update backups during startup cleanup")
}
}
// cleanupOldTempDirs removes old pulse-update-* temp directories from previous runs
func (m *Manager) cleanupOldTempDirs() {
// Check multiple locations where temp dirs might exist
dirsToCheck := []string{"/tmp", resolveUpdateDataDir(), "."}
for _, dir := range dirsToCheck {
entries, err := os.ReadDir(dir)
if err != nil {
continue // Directory not accessible, skip
}
for _, entry := range entries {
if !entry.IsDir() {
continue
}
// Check if it matches pulse-update-* pattern
if !strings.HasPrefix(entry.Name(), "pulse-update-") {
continue
}
fullPath, err := securityutil.JoinStorageLeaf(dir, entry.Name())
if err != nil {
log.Debug().Err(err).Str("dir", dir).Str("entry", entry.Name()).Msg("Skipping invalid temp directory leaf")
continue
}
info, err := os.Stat(fullPath)
if err != nil {
continue
}
// Remove directories older than 24 hours
if time.Since(info.ModTime()) > 24*time.Hour {
if err := os.RemoveAll(fullPath); err != nil {
log.Debug().Err(err).Str("path", fullPath).Msg("Failed to cleanup old temp directory")
} else {
log.Info().Str("path", fullPath).Msg("Cleaned up old temp directory")
}
}
}
}
}
// copyFileSafe safely copies a file, skipping symlinks for security
func (m *Manager) copyFileSafe(src, dest string) error {
// Get file info and check if it's a symlink
info, err := os.Lstat(src)
if err != nil {
return fmt.Errorf("lstat source file %q: %w", src, err)
}
// Skip symlinks for security
if info.Mode()&os.ModeSymlink != 0 {
log.Warn().Str("file", src).Msg("Skipping symlink during backup/restore")
return nil
}
// Open source file
srcFile, err := os.Open(src)
if err != nil {
return fmt.Errorf("open source file %q: %w", src, err)
}
defer func() {
if closeErr := srcFile.Close(); closeErr != nil {
log.Warn().Err(closeErr).Str("path", src).Msg("Failed to close source file after copy")
}
}()
// Create destination file with same permissions
destFile, err := os.OpenFile(dest, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, info.Mode())
if err != nil {
return fmt.Errorf("open destination file %q: %w", dest, err)
}
defer func() {
if closeErr := destFile.Close(); closeErr != nil {
log.Warn().Err(closeErr).Str("path", dest).Msg("Failed to close destination file after copy")
}
}()
// Copy contents
if _, err := io.Copy(destFile, srcFile); err != nil {
return fmt.Errorf("copy %q to %q: %w", src, dest, err)
}
return nil
}
// copyDirSafe recursively copies a directory, skipping symlinks for security
func (m *Manager) copyDirSafe(src, dest string) error {
// Get source directory info
srcInfo, err := os.Stat(src)
if err != nil {
return fmt.Errorf("stat source directory %q: %w", src, err)
}
// Create destination directory
if err := os.MkdirAll(dest, srcInfo.Mode()); err != nil {
return fmt.Errorf("create destination directory %q: %w", dest, err)
}
// Read source directory entries
entries, err := os.ReadDir(src)
if err != nil {
return fmt.Errorf("read source directory %q: %w", src, err)
}
for _, entry := range entries {
srcPath, err := securityutil.JoinStorageLeaf(src, entry.Name())
if err != nil {
log.Warn().Err(err).Str("dir", src).Str("entry", entry.Name()).Msg("Skipping invalid source path during backup/restore")
continue
}
destPath, err := securityutil.JoinStorageLeaf(dest, entry.Name())
if err != nil {
log.Warn().Err(err).Str("dir", dest).Str("entry", entry.Name()).Msg("Skipping invalid destination path during backup/restore")
continue
}
// Get file info (using Lstat to detect symlinks)
info, err := os.Lstat(srcPath)
if err != nil {
log.Warn().Str("path", srcPath).Err(err).Msg("Failed to stat file during copy")
continue
}
// Skip symlinks for security
if info.Mode()&os.ModeSymlink != 0 {
log.Warn().Str("path", srcPath).Msg("Skipping symlink during backup/restore")
continue
}
if entry.IsDir() {
// Recursively copy subdirectory
if err := m.copyDirSafe(srcPath, destPath); err != nil {
return fmt.Errorf("copy subdirectory %q: %w", srcPath, err)
}
} else {
// Copy file
if err := m.copyFileSafe(srcPath, destPath); err != nil {
log.Warn().Str("file", srcPath).Err(err).Msg("Failed to copy file")
continue
}
}
}
return nil
}
// isPreV4Installation checks if this is a pre-v4 (Node.js based) installation
func isPreV4Installation() bool {
// Check for .env file (used by Node.js version)
if _, err := os.Stat("/opt/pulse/.env"); err == nil {
return true
}
// Note: pulse-backend.service is used by both v4 and pre-v4, so we can't use it as an indicator
// Only check for Node.js artifacts which are exclusive to pre-v4
// Check for Node.js artifacts
nodeArtifacts := []string{
"/opt/pulse/package.json",
"/opt/pulse/node_modules",
"/opt/pulse/server.js",
"/opt/pulse/backend",
"/opt/pulse/frontend",
}
for _, artifact := range nodeArtifacts {
if _, err := os.Stat(artifact); err == nil {
return true
}
}
return false
}