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

package poller

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

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

type availableCall struct {
	id int64
	a  model.Available
}

type pollErrorCall struct {
	ids []int64
	msg string
}

type lastTagCall struct {
	id     int64
	digest string
}

type fakeStore struct {
	mu         sync.Mutex
	services   []model.Service
	fresh      map[int64]model.Service // what GetService returns instead of the listed row
	lastTag    []lastTagCall
	listErr    error
	available  []availableCall
	pollErrors []pollErrorCall
	ctxErrs    []error // ctx.Err() seen by every write
}

func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) {
	return f.services, f.listErr
}

func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if s, ok := f.fresh[id]; ok {
		return s, nil
	}
	for _, s := range f.services {
		if s.ID == id {
			return s, nil
		}
	}
	return model.Service{}, model.ErrNotFound
}

// SetLastTagDigest records the call and, like the database, changes the
// row the next cycle lists.
func (f *fakeStore) SetLastTagDigest(_ context.Context, id int64, digest string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.lastTag = append(f.lastTag, lastTagCall{id, digest})
	for i := range f.services {
		if f.services[i].ID == id {
			f.services[i].LastTagDigest = digest
		}
	}
	return nil
}

func (f *fakeStore) SetAvailable(ctx context.Context, id int64, a model.Available) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.ctxErrs = append(f.ctxErrs, ctx.Err())
	f.available = append(f.available, availableCall{id, a})
	return nil
}

func (f *fakeStore) SetPollError(ctx context.Context, ids []int64, msg string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.ctxErrs = append(f.ctxErrs, ctx.Err())
	f.pollErrors = append(f.pollErrors, pollErrorCall{ids, msg})
	return nil
}

type startCall struct {
	svcID int64
	req   jobs.StartRequest
}

type fakeJobs struct {
	mu      sync.Mutex
	starts  []startCall
	errFor  map[int64]error // by service ID
	ctxErrs []error
}

func (f *fakeJobs) Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.ctxErrs = append(f.ctxErrs, ctx.Err())
	f.starts = append(f.starts, startCall{svc.ID, req})
	return model.Job{ID: int64(len(f.starts))}, f.errFor[svc.ID]
}

type fakeRegistry struct {
	mu       sync.Mutex
	digests  map[string]string // ref -> digest
	errs     map[string]error  // ref -> error
	calls    []string
	inFlight int
	maxSeen  int
	delay    time.Duration
	started  chan struct{} // receives a value whenever a lookup starts, if set
	release  chan struct{} // lookups wait for it to be closed, if set
	onResult func()        // called after a lookup computed its answer
}

func (f *fakeRegistry) Digest(ctx context.Context, ref string) (string, error) {
	f.mu.Lock()
	f.calls = append(f.calls, ref)
	f.inFlight++
	if f.inFlight > f.maxSeen {
		f.maxSeen = f.inFlight
	}
	f.mu.Unlock()
	defer func() {
		f.mu.Lock()
		f.inFlight--
		f.mu.Unlock()
	}()

	if f.started != nil {
		f.started <- struct{}{}
	}
	if f.release != nil {
		select {
		case <-f.release:
		case <-ctx.Done():
			return "", ctx.Err()
		}
	}
	if f.delay > 0 {
		time.Sleep(f.delay)
	}
	if f.onResult != nil {
		f.onResult()
	}
	if err := f.errs[ref]; err != nil {
		return "", err
	}
	return f.digests[ref], nil
}

func (f *fakeRegistry) sortedCalls() []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := append([]string(nil), f.calls...)
	sort.Strings(out)
	return out
}

type fakePub struct {
	mu     sync.Mutex
	topics []events.Topic
}

func (p *fakePub) Publish(t events.Topic) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.topics = append(p.topics, t)
}

const (
	repoA = "gitea.shcizo.se/shcizo/slusk"
	repoB = "gitea/gitea"
)

// svc builds a polled service whose running version has digest sha256:old.
func svc(id int64, actualRef string) model.Service {
	return model.Service{ID: id, Kind: "compose", ActualRef: actualRef, ActualDigest: "sha256:old", UpdateMode: model.ModeManual}
}

func newPoller(st *fakeStore, j *fakeJobs, reg *fakeRegistry, cfg Config) (*Poller, *fakePub) {
	pub := &fakePub{}
	if cfg.Workers == 0 {
		cfg.Workers = 4
	}
	p := New(st, j, reg, pub, cfg, slog.New(slog.NewTextHandler(io.Discard, nil)))
	p.now = func() time.Time { return time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) }
	return p, pub
}

func TestSkipReason(t *testing.T) {
	tests := []struct {
		name string
		svc  model.Service
		want string
	}{
		{"tag and digest", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a"}, ""},
		{"bare image follows latest", model.Service{ActualRef: "r", ActualDigest: "sha256:a"}, ""},
		{"digest only follows no tag", model.Service{ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, model.ReasonNoTag},
		{"built", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Built: true}, "built from source"},
		{"no digest", model.Service{ActualRef: "r:1"}, "no registry digest"},
		{"self and container are polled", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Self: true, Kind: "container"}, ""},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			assert.Equal(t, tt.want, SkipReason(tt.svc))
		})
	}
}

func TestOneHeadPerDistinctRef(t *testing.T) {
	st := &fakeStore{services: []model.Service{
		svc(1, repoA+":1.4"),
		svc(2, repoA+":1.4@sha256:old"),
		svc(3, repoB),
		svc(4, repoB+":latest"),
		svc(5, repoB+":1.27"),
	}}
	reg := &fakeRegistry{digests: map[string]string{}}
	p, _ := newPoller(st, &fakeJobs{}, reg, Config{})

	p.Cycle(context.Background())

	assert.Equal(t, []string{repoA + ":1.4", repoB + ":1.27", repoB + ":latest"}, reg.sortedCalls())
}

func TestSkipped(t *testing.T) {
	removed := svc(4, "r:1")
	removed.RemovedAt = ptr(time.Now())
	st := &fakeStore{services: []model.Service{
		{ID: 1, ActualRef: "r@sha256:a", ActualDigest: "sha256:a"},
		{ID: 2, ActualRef: "r:1", ActualDigest: "sha256:a", Built: true},
		{ID: 3, ActualRef: "r:1"},
		removed,
	}}
	reg := &fakeRegistry{}
	j := &fakeJobs{}
	p, pub := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())

	assert.Empty(t, reg.calls, "no lookups")
	assert.Empty(t, st.available)
	assert.Empty(t, st.pollErrors)
	assert.Empty(t, j.starts)
	assert.Empty(t, pub.topics)
}

func TestNewDigest(t *testing.T) {
	ref := repoA + ":1.4"
	tests := []struct {
		name      string
		mode      string
		wantStart bool
	}{
		{"auto gets one poll job to repo:tag", model.ModeAuto, true},
		{"manual is only marked", model.ModeManual, false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			s := svc(1, ref+"@sha256:old")
			s.UpdateMode = tt.mode
			st := &fakeStore{services: []model.Service{s}}
			reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
			j := &fakeJobs{}
			p, pub := newPoller(st, j, reg, Config{})

			p.Cycle(context.Background())

			require.Len(t, st.available, 1)
			assert.Equal(t, int64(1), st.available[0].id)
			assert.Equal(t, model.Available{Ref: ref, Digest: "sha256:new", Source: model.SourcePoll,
				At: time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC)}, st.available[0].a)
			if tt.wantStart {
				assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts)
				assert.Equal(t, []lastTagCall{{1, "sha256:new"}}, st.lastTag)
			} else {
				assert.Empty(t, j.starts)
				assert.Empty(t, st.lastTag)
			}
			assert.Equal(t, []events.Topic{events.Services}, pub.topics)
		})
	}
}

func TestSameSuggestionIsIdempotent(t *testing.T) {
	ref := repoA + ":1.4"
	running := svc(1, ref+"@sha256:old")
	running.UpdateMode = model.ModeAuto
	running.ActualDigest = "sha256:new" // already on the registry's digest
	suggested := svc(2, ref)
	suggested.UpdateMode = model.ModeAuto
	suggested.AvailableRef, suggested.AvailableDigest = ref, "sha256:new"
	suggested.LastTagDigest = "sha256:new" // its job was started in an earlier cycle
	staleSuggestion := svc(3, ref)
	staleSuggestion.AvailableRef, staleSuggestion.AvailableDigest = ref, "sha256:older"
	st := &fakeStore{services: []model.Service{running, suggested, staleSuggestion}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	j := &fakeJobs{}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())

	require.Len(t, st.available, 1, "only the service with an older suggestion is updated")
	assert.Equal(t, int64(3), st.available[0].id)
	assert.Empty(t, j.starts, "no new job for an already suggested digest")
}

func TestStartErrorsDoNotAbortCycle(t *testing.T) {
	ref := repoA + ":1.4"
	var svcs []model.Service
	for id := int64(1); id <= 3; id++ {
		s := svc(id, ref)
		s.UpdateMode = model.ModeAuto
		svcs = append(svcs, s)
	}
	st := &fakeStore{services: svcs}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	j := &fakeJobs{errFor: map[int64]error{
		1: model.ErrJobActive,
		2: &model.NotUpdatableError{Reason: model.ReasonSelf},
		3: errors.New("database down"),
	}}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())

	assert.Len(t, st.available, 3)
	assert.Len(t, j.starts, 3)
	assert.Empty(t, st.lastTag, "a job that was not created is retried next cycle")
}

func TestRollbackOfAutoServiceIsNotUndone(t *testing.T) {
	ref := repoA + ":main"
	// Updated to sha256:new automatically, then rolled back to sha256:old by hand.
	s := svc(1, ref+"@sha256:old")
	s.UpdateMode = model.ModeAuto
	s.LastTagDigest = "sha256:new"
	s.DesiredRef = ref + "@sha256:old"
	st := &fakeStore{services: []model.Service{s}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	j := &fakeJobs{}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())
	p.Cycle(context.Background())

	assert.Empty(t, j.starts, "same registry digest: the rollback stays")
	assert.Equal(t, "sha256:new", st.available[0].a.Digest, "the newer version is still offered")

	reg.digests[ref] = "sha256:newer"
	p.Cycle(context.Background())

	assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts, "a new digest moves it again")
	assert.Equal(t, []lastTagCall{{1, "sha256:newer"}}, st.lastTag)
}

func TestActiveJobIsRetriedNextCycle(t *testing.T) {
	ref := repoA + ":1.4"
	s := svc(1, ref)
	s.UpdateMode = model.ModeAuto
	st := &fakeStore{services: []model.Service{s}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	j := &fakeJobs{errFor: map[int64]error{1: model.ErrJobActive}}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())
	require.Len(t, j.starts, 1)
	assert.Empty(t, st.lastTag)

	j.errFor = nil
	p.Cycle(context.Background())

	assert.Len(t, j.starts, 2)
	assert.Equal(t, []lastTagCall{{1, "sha256:new"}}, st.lastTag)

	p.Cycle(context.Background())
	assert.Len(t, j.starts, 2, "started once the job exists")
}

func TestJobThatAlreadyDeployedTheDigestIsNotStartedAgain(t *testing.T) {
	ref := repoA + ":1.4"
	// FinishJob set desired_ref; the agent has not reported the new digest yet.
	s := svc(1, ref+"@sha256:old")
	s.UpdateMode = model.ModeAuto
	s.DesiredRef = ref + "@sha256:new"
	st := &fakeStore{services: []model.Service{s}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	j := &fakeJobs{}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())

	assert.Empty(t, j.starts)
}

func TestServiceSwitchedToManualDuringTheCycleIsNotStarted(t *testing.T) {
	ref := repoA + ":1.4"
	s := svc(1, ref)
	s.UpdateMode = model.ModeAuto
	manual := s
	manual.UpdateMode = model.ModeManual
	st := &fakeStore{services: []model.Service{s}, fresh: map[int64]model.Service{1: manual}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	j := &fakeJobs{}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())

	assert.Len(t, st.available, 1, "still suggested")
	assert.Empty(t, j.starts)
	assert.Empty(t, st.lastTag)
}

func TestErrorIsPerRef(t *testing.T) {
	bad, good := repoA+":1.4", repoB+":1.27"
	badOld := svc(1, bad)
	badSame := svc(2, bad)
	badSame.LastPollError = "head: 401 unauthorized"
	goodSvc := svc(3, good)
	st := &fakeStore{services: []model.Service{badOld, badSame, goodSvc}}
	reg := &fakeRegistry{
		errs:    map[string]error{bad: errors.New("head: 401 unauthorized")},
		digests: map[string]string{good: "sha256:new"},
	}
	p, pub := newPoller(st, &fakeJobs{}, reg, Config{})

	p.Cycle(context.Background())

	assert.Equal(t, []pollErrorCall{{[]int64{1}, "head: 401 unauthorized"}}, st.pollErrors,
		"only the service whose stored error differs is written")
	require.Len(t, st.available, 1, "the other ref is still processed")
	assert.Equal(t, int64(3), st.available[0].id)
	assert.Equal(t, []events.Topic{events.Services}, pub.topics)
}

func TestErrorTextIsTruncated(t *testing.T) {
	ref := repoA + ":1.4"
	st := &fakeStore{services: []model.Service{svc(1, ref)}}
	reg := &fakeRegistry{errs: map[string]error{ref: errors.New(strings.Repeat("é", 400) + "\x00")}}
	p, _ := newPoller(st, &fakeJobs{}, reg, Config{})

	p.Cycle(context.Background())

	require.Len(t, st.pollErrors, 1)
	msg := st.pollErrors[0].msg
	assert.LessOrEqual(t, len(msg), maxErrorBytes)
	assert.Equal(t, strings.Repeat("é", 250), msg, "cut on a character boundary, NUL removed")
}

func TestErrorClearedOnSuccess(t *testing.T) {
	ref := repoA + ":1.4"
	failing := svc(1, ref)
	failing.LastPollError = "head: boom"
	fine := svc(2, ref)
	failing.ActualDigest, fine.ActualDigest = "sha256:new", "sha256:new"
	st := &fakeStore{services: []model.Service{failing, fine}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	p, pub := newPoller(st, &fakeJobs{}, reg, Config{})

	p.Cycle(context.Background())

	assert.Equal(t, []pollErrorCall{{[]int64{1}, ""}}, st.pollErrors)
	assert.Empty(t, st.available)
	assert.Equal(t, []events.Topic{events.Services}, pub.topics)
}

func TestNothingWrittenNothingPublished(t *testing.T) {
	ref := repoA + ":1.4"
	s := svc(1, ref)
	s.ActualDigest = "sha256:new"
	st := &fakeStore{services: []model.Service{s}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	p, pub := newPoller(st, &fakeJobs{}, reg, Config{})

	p.Cycle(context.Background())

	assert.Empty(t, st.pollErrors)
	assert.Empty(t, st.available)
	assert.Empty(t, pub.topics)
}

func TestListErrorEndsTheCycle(t *testing.T) {
	st := &fakeStore{listErr: errors.New("db down")}
	reg := &fakeRegistry{}
	p, _ := newPoller(st, &fakeJobs{}, reg, Config{})

	p.Cycle(context.Background())

	assert.Empty(t, reg.calls)
}

func TestConcurrencyLimit(t *testing.T) {
	const workers = 3
	var svcs []model.Service
	for i := 0; i < 8; i++ {
		svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i)))
	}
	st := &fakeStore{services: svcs}
	reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 8)}
	p, _ := newPoller(st, &fakeJobs{}, reg, Config{Workers: workers})
	done := make(chan struct{})
	go func() { p.Cycle(context.Background()); close(done) }()

	for i := 0; i < workers; i++ {
		select {
		case <-reg.started:
		case <-time.After(2 * time.Second):
			t.Fatalf("only %d lookups started, want %d", i, workers)
		}
	}
	select {
	case <-reg.started:
		t.Fatal("more lookups in flight than Workers")
	case <-time.After(100 * time.Millisecond):
	}
	close(reg.release)

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("Cycle did not finish")
	}
	assert.Len(t, reg.calls, 8)
}

func TestCycleStopsOnCancel(t *testing.T) {
	var svcs []model.Service
	for i := 0; i < 6; i++ {
		svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i)))
	}
	st := &fakeStore{services: svcs}
	reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 6)}
	p, pub := newPoller(st, &fakeJobs{}, reg, Config{Workers: 2})
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { p.Cycle(ctx); close(done) }()
	<-reg.started
	<-reg.started

	cancel()

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("Cycle did not return after cancel")
	}
	assert.Len(t, reg.calls, 2, "no new lookups after cancel")
	assert.Empty(t, st.pollErrors, "cancelled lookups are not recorded as registry errors")
	assert.Empty(t, st.available)
	assert.Empty(t, pub.topics)
}

func TestMarkedJobsAreStartedEvenIfCancelledDuringTheCycle(t *testing.T) {
	ref := repoA + ":1.4"
	s := svc(1, ref)
	s.UpdateMode = model.ModeAuto
	st := &fakeStore{services: []model.Service{s}}
	ctx, cancel := context.WithCancel(context.Background())
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}, onResult: cancel}
	j := &fakeJobs{}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(ctx)

	require.Len(t, st.available, 1)
	require.Len(t, j.starts, 1)
	assert.NoError(t, st.ctxErrs[0], "writes do not see the cancelled context")
	assert.NoError(t, j.ctxErrs[0])
}

func TestRun(t *testing.T) {
	ref := repoA + ":1.4"
	st := &fakeStore{services: []model.Service{svc(1, ref)}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:old"}, delay: 30 * time.Millisecond}
	p, _ := newPoller(st, &fakeJobs{}, reg, Config{Interval: 5 * time.Millisecond, FirstDelay: 60 * time.Millisecond, Workers: 1})
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	began := time.Now()
	go func() { p.Run(ctx); close(done) }()

	require.Eventually(t, func() bool {
		reg.mu.Lock()
		defer reg.mu.Unlock()
		return len(reg.calls) >= 3
	}, 5*time.Second, 5*time.Millisecond)
	cancel()

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("Run did not return after cancel")
	}
	assert.GreaterOrEqual(t, time.Since(began), 60*time.Millisecond+2*30*time.Millisecond, "first cycle waits FirstDelay")
	assert.Equal(t, 1, reg.maxSeen, "cycles never overlap")
}

func TestRunReturnsDuringFirstDelay(t *testing.T) {
	p, _ := newPoller(&fakeStore{}, &fakeJobs{}, &fakeRegistry{}, Config{Interval: time.Hour, FirstDelay: time.Hour})
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { p.Run(ctx); close(done) }()

	cancel()

	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("Run did not return after cancel")
	}
}

func ptr[T any](v T) *T { return &v }

func TestSeedsTheTagDigestOfARunningService(t *testing.T) {
	ref := repoA + ":main"
	s := svc(1, ref+"@sha256:new")
	s.UpdateMode = model.ModeAuto
	s.ActualDigest = "sha256:new" // updated outside the updater: last_tag_digest is empty
	st := &fakeStore{services: []model.Service{s}}
	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
	j := &fakeJobs{}
	p, _ := newPoller(st, j, reg, Config{})

	p.Cycle(context.Background())

	assert.Equal(t, []lastTagCall{{1, "sha256:new"}}, st.lastTag)
	assert.Empty(t, j.starts)

	// Rolled back by hand to the older digest.
	st.services[0].ActualRef, st.services[0].ActualDigest = ref+"@sha256:old", "sha256:old"
	st.services[0].DesiredRef = ref + "@sha256:old"
	p.Cycle(context.Background())

	assert.Empty(t, j.starts, "the rollback stays")
	assert.Len(t, st.lastTag, 1, "nothing more to seed")
}