package-updater/internal/master/poller/poller.go

// Package poller checks the registry for new digests of the tags the
// services follow. A new digest becomes the service's available version, and
// a service in auto mode gets an update job.
package poller

import (
	"context"
	"errors"
	"log/slog"
	"strings"
	"sync"
	"time"

	"github.com/shcizo/package-updater/internal/master/events"
	"github.com/shcizo/package-updater/internal/master/jobs"
	"github.com/shcizo/package-updater/internal/master/model"
)

// lookupTimeout bounds one registry lookup.
const lookupTimeout = 30 * time.Second

// maxErrorBytes caps the registry error stored in services.last_poll_error.
const maxErrorBytes = 500

// Store is the storage the poller needs. *store.Store satisfies it.
type Store interface {
	ListServices(ctx context.Context) ([]model.Service, error)
	SetAvailable(ctx context.Context, id int64, a model.Available) error
	SetPollError(ctx context.Context, ids []int64, msg string) error
}

// Jobs starts updates. *jobs.Service satisfies it.
type Jobs interface {
	Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error)
}

// Registry looks up the digest a "repo:tag" ref points at. *registry.Client
// satisfies it.
type Registry interface {
	Digest(ctx context.Context, ref string) (string, error)
}

// Publisher receives change signals. *events.Broker satisfies it.
type Publisher interface{ Publish(t events.Topic) }

// Config holds the poller settings.
type Config struct {
	Interval   time.Duration // time between the end of a cycle and the start of the next (POLL_INTERVAL)
	FirstDelay time.Duration // wait before the first cycle; 1 minute in production
	Workers    int           // registry lookups in flight at once; 4 in production
}

// Poller runs polling cycles.
type Poller struct {
	store Store
	jobs  Jobs
	reg   Registry
	pub   Publisher
	cfg   Config
	log   *slog.Logger
	now   func() time.Time
}

// New returns a Poller.
func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Logger) *Poller {
	if cfg.Workers < 1 {
		cfg.Workers = 1
	}
	return &Poller{store: st, jobs: j, reg: reg, pub: pub, cfg: cfg, log: log, now: time.Now}
}

// SkipReason says why svc is not polled: "follows no tag", "built from
// source", "no registry digest", or "" when it is polled.
func SkipReason(svc model.Service) string {
	switch {
	case jobs.FollowedRef(svc.ActualRef) == "":
		return "follows no tag"
	case svc.Built:
		return "built from source"
	case svc.ActualDigest == "":
		return "no registry digest"
	}
	return ""
}

// Run polls until ctx is done. The first cycle starts Config.FirstDelay after
// the call and each following one Config.Interval after the previous cycle
// ended, so cycles never overlap. It blocks.
func (p *Poller) Run(ctx context.Context) {
	timer := time.NewTimer(p.cfg.FirstDelay)
	defer timer.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-timer.C:
		}
		p.Cycle(ctx)
		timer.Reset(p.cfg.Interval)
	}
}

// lookup is the registry's answer for one followed ref.
type lookup struct {
	ref    string
	digest string
	err    error
}

// cycleStats is what one cycle did, for the poll_cycle log line.
type cycleStats struct {
	refs, ok, failed, suggested, jobs, skipped int
	wrote                                      bool
}

// Cycle runs one poll over all services. It looks up every distinct followed
// ref once, with at most Config.Workers lookups at a time, then records the
// results. Writing results and starting jobs is not cut short by ctx ending
// during the lookups, so a job that was marked is always started.
func (p *Poller) Cycle(ctx context.Context) {
	began := p.now()
	svcs, err := p.store.ListServices(ctx)
	if err != nil {
		p.log.Error("poll_list_failed", "error", err)
		return
	}
	var stats cycleStats
	byRef := map[string][]model.Service{}
	var refs []string
	for _, svc := range svcs {
		if svc.RemovedAt != nil {
			continue
		}
		if SkipReason(svc) != "" {
			stats.skipped++
			continue
		}
		ref := jobs.FollowedRef(svc.ActualRef)
		if _, seen := byRef[ref]; !seen {
			refs = append(refs, ref)
		}
		byRef[ref] = append(byRef[ref], svc)
	}
	stats.refs = len(refs)

	write := context.WithoutCancel(ctx)
	for _, l := range p.lookupAll(ctx, refs) {
		if l.err != nil && ctx.Err() != nil {
			continue // cancelled mid-lookup: not the registry's fault
		}
		p.record(write, l, byRef[l.ref], &stats)
	}
	if stats.wrote {
		p.pub.Publish(events.Services)
	}
	p.log.Info("poll_cycle", "refs", stats.refs, "ok", stats.ok, "failed", stats.failed,
		"suggested", stats.suggested, "jobs", stats.jobs, "skipped", stats.skipped,
		"duration_ms", p.now().Sub(began).Milliseconds())
}

// lookupAll asks the registry for every ref, Config.Workers at a time. It
// returns the answers in ref order and stops starting new lookups once ctx
// is done; refs never looked up are missing from the result.
func (p *Poller) lookupAll(ctx context.Context, refs []string) []lookup {
	results := make([]*lookup, len(refs))
	next := make(chan int)
	var wg sync.WaitGroup
	for w := 0; w < p.cfg.Workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := range next {
				lctx, cancel := context.WithTimeout(ctx, lookupTimeout)
				d, err := p.reg.Digest(lctx, refs[i])
				cancel()
				results[i] = &lookup{ref: refs[i], digest: d, err: err}
			}
		}()
	}
feed:
	for i := range refs {
		select {
		case next <- i:
		case <-ctx.Done():
			break feed
		}
	}
	close(next)
	wg.Wait()
	var out []lookup
	for _, r := range results {
		if r != nil {
			out = append(out, *r)
		}
	}
	return out
}

// record stores one lookup result for the services that follow l.ref.
func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, stats *cycleStats) {
	if l.err != nil {
		stats.failed++
		msg := truncate(l.err.Error())
		p.log.Warn("poll_failed", "ref", l.ref, "error", msg)
		p.setPollError(ctx, svcs, msg, stats, func(s model.Service) bool { return s.LastPollError != msg })
		return
	}
	stats.ok++
	p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" })
	for _, svc := range svcs {
		if svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) {
			continue
		}
		err := p.store.SetAvailable(ctx, svc.ID, model.Available{Ref: l.ref, Digest: l.digest, Source: model.SourcePoll, At: p.now()})
		if err != nil {
			p.log.Error("poll_set_available_failed", "service_id", svc.ID, "ref", l.ref, "error", err)
			continue
		}
		stats.wrote = true
		stats.suggested++
		if svc.UpdateMode != model.ModeAuto {
			continue
		}
		p.startAuto(ctx, svc, l.ref, stats)
	}
}

// startAuto starts the update job of an auto service to ref. A service that
// already has a job or cannot be updated is not an error for the cycle.
func (p *Poller) startAuto(ctx context.Context, svc model.Service, ref string, stats *cycleStats) {
	_, err := p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref})
	var notUpdatable *model.NotUpdatableError
	switch {
	case err == nil:
		stats.jobs++
	case errors.Is(err, model.ErrJobActive), errors.As(err, &notUpdatable):
		p.log.Debug("poll_job_not_started", "service_id", svc.ID, "ref", ref, "reason", err.Error())
	default:
		p.log.Warn("poll_job_failed", "service_id", svc.ID, "ref", ref, "error", err)
	}
}

// setPollError sets msg as last_poll_error on the services for which
// changed returns true; with none, it writes nothing.
func (p *Poller) setPollError(ctx context.Context, svcs []model.Service, msg string, stats *cycleStats, changed func(model.Service) bool) {
	var ids []int64
	for _, svc := range svcs {
		if changed(svc) {
			ids = append(ids, svc.ID)
		}
	}
	if len(ids) == 0 {
		return
	}
	if err := p.store.SetPollError(ctx, ids, msg); err != nil {
		p.log.Error("poll_set_error_failed", "error", err)
		return
	}
	stats.wrote = true
}

// truncate cuts msg to maxErrorBytes without leaving a broken UTF-8
// sequence, and drops NUL bytes, which Postgres text cannot hold.
func truncate(msg string) string {
	msg = strings.ReplaceAll(msg, "\x00", "")
	if len(msg) > maxErrorBytes {
		msg = strings.ToValidUTF8(msg[:maxErrorBytes], "")
	}
	return msg
}