worker/internal/webapp/compose.go

package webapp

import (
	"context"
	"log"
	"os"
	"sync"
	"time"

	"rocketgit.ru/rsmon/worker/internal/compose"
)

// composeRefreshInterval is the cadence the Compose project snapshot is
// rebuilt. 60s matches the process-inventory loop so the two collectors
// stay in step; management operations are synchronous POSTs that do not
// wait for the next tick.
const composeRefreshInterval = 60 * time.Second

// ComposeRefresher owns the periodic `docker compose` discovery loop and
// exposes the latest snapshot for the operator console. It is the
// Compose counterpart of Inventory: a self-contained refresher that is
// constructed in New and started in Start.
//
// When Docker is unavailable (or compose is disabled) the refresher
// stays idle and reports an empty snapshot with an "unavailable" error,
// so /compose renders a banner instead of a half-built table.
type ComposeRefresher struct {
	enabled bool
	log     *log.Logger
	disco   composeDiscover

	mu      sync.RWMutex
	snap    compose.Snapshot
	lastAt  time.Time
	stopCh  chan struct{}
	stopWG  sync.WaitGroup
	started bool
}

// composeDiscover is the subset of the compose package the refresher
// calls. A function field so tests can inject a fixture discovery
// without exec'ing Docker.
type composeDiscover func(ctx context.Context) (*compose.DiscoveryResult, error)

// NewComposeRefresher constructs a refresher. When enabled is false the
// refresher never starts its loop and Snapshot returns an empty result;
// the page still renders (with the disabled notice). The discover hook
// defaults to compose.Discover so production needs no wiring.
func NewComposeRefresher(enabled bool, logger *log.Logger) *ComposeRefresher {
	if logger == nil {
		logger = log.New(os.Stderr, "webapp-compose: ", log.LstdFlags)
	}
	return &ComposeRefresher{
		enabled: enabled,
		log:     logger,
		disco:   compose.Discover,
		stopCh:  make(chan struct{}),
	}
}

// SetDiscovery injects the discovery function. Used by tests; production
// leaves the compose.Discover default set by NewComposeRefresher.
func (c *ComposeRefresher) SetDiscovery(f composeDiscover) {
	if f == nil {
		c.disco = compose.Discover
		return
	}
	c.disco = f
}

// Enabled reports whether compose management is turned on.
func (c *ComposeRefresher) Enabled() bool { return c != nil && c.enabled }

// Start launches the background refresh loop. The first refresh runs
// immediately so /compose has data on the first request. Idempotent.
func (c *ComposeRefresher) Start(ctx context.Context) {
	if c == nil || !c.enabled {
		return
	}
	c.mu.Lock()
	if c.started {
		c.mu.Unlock()
		return
	}
	c.started = true
	c.mu.Unlock()

	c.stopWG.Add(1)
	go c.loop(ctx)
}

// Stop cancels the refresh loop and waits for it to exit.
func (c *ComposeRefresher) Stop() {
	if c == nil {
		return
	}
	c.mu.Lock()
	if !c.started {
		c.mu.Unlock()
		return
	}
	select {
	case <-c.stopCh:
	default:
		close(c.stopCh)
	}
	c.mu.Unlock()
	c.stopWG.Wait()
}

// Snapshot returns the most recent discovery snapshot. Always safe to
// call; returns a zero-value snapshot (with the unavailable notice when
// disabled) before the first refresh completes.
func (c *ComposeRefresher) Snapshot() compose.Snapshot {
	if c == nil {
		return compose.Snapshot{Errors: []string{"compose management is disabled"}}
	}
	c.mu.RLock()
	defer c.mu.RUnlock()
	if !c.enabled {
		return compose.Snapshot{Errors: []string{"compose management is disabled"}}
	}
	out := c.snap
	out.Projects = append([]compose.ProjectSummary(nil), c.snap.Projects...)
	return out
}

// LastRefreshAt returns the time of the last successful refresh.
func (c *ComposeRefresher) LastRefreshAt() time.Time {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.lastAt
}

// Refs resolves the Compose CLI arguments for a project from the current
// snapshot. Returns ok=false when the project is unknown or its working
// directory was not discovered (Compose cannot be driven without it).
func (c *ComposeRefresher) Refs(project string) (compose.ProjectRefs, bool) {
	p := c.Snapshot().FindProject(project)
	if p == nil || p.WorkingDir == "" {
		return compose.ProjectRefs{}, false
	}
	return compose.ProjectRefs{Name: p.Name, WorkingDir: p.WorkingDir, ConfigFiles: p.ConfigFiles}, true
}

func (c *ComposeRefresher) loop(ctx context.Context) {
	defer c.stopWG.Done()
	c.refresh(ctx)
	t := time.NewTicker(composeRefreshInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-c.stopCh:
			return
		case <-t.C:
			c.refresh(ctx)
		}
	}
}

func (c *ComposeRefresher) refresh(ctx context.Context) {
	res, err := c.disco(ctx)
	if err != nil || res == nil {
		if err != nil {
			c.log.Printf("compose refresh: %v", err)
		}
		return
	}
	snap := compose.Summarize(res)
	c.mu.Lock()
	c.snap = snap
	c.lastAt = time.Now().UTC()
	c.mu.Unlock()
}