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() }