package jobs

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"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/model"
	"github.com/shcizo/package-updater/internal/protocol"
)

const selfImage = "gitea.shcizo.se/shcizo/package-updater:latest"

type fakeStore struct {
	services map[int64]model.Service
	jobs     map[int64]model.Job
	created  []model.NewJob
	active   *model.Job // returned with ErrJobActive when set
	versions map[int64]model.ImageVersion

	revokedAgents map[int64]bool   // CreateJob returns ErrAgentOffline for these
	lastTag       map[int64]string // SetLastTagDigest calls: service ID to digest
}

func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) {
	svc, ok := f.services[id]
	if !ok {
		return model.Service{}, model.ErrNotFound
	}
	return svc, nil
}

func (f *fakeStore) CreateJob(_ context.Context, nj model.NewJob) (model.Job, error) {
	if f.active != nil {
		return *f.active, model.ErrJobActive
	}
	if nj.AgentID != nil && f.revokedAgents[*nj.AgentID] {
		return model.Job{}, model.ErrAgentOffline
	}
	f.created = append(f.created, nj)
	j := model.Job{ID: int64(len(f.created)), ServiceID: nj.ServiceID, AgentID: nj.AgentID, FromRef: nj.FromRef,
		ToRef: nj.ToRef, Trigger: nj.Trigger, RequestedBy: nj.RequestedBy, Status: model.JobQueued, CreatedAt: nj.CreatedAt}
	f.jobs[j.ID] = j
	return j, nil
}

func (f *fakeStore) FailJob(_ context.Context, id int64, msg string, at time.Time) (bool, error) {
	j := f.jobs[id]
	j.Status, j.Error, j.FinishedAt = model.JobFailed, msg, &at
	f.jobs[id] = j
	return true, nil
}

func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) {
	v, ok := f.versions[id]
	if !ok {
		return model.ImageVersion{}, model.ErrNotFound
	}
	return v, nil
}

func (f *fakeStore) SetLastTagDigest(_ context.Context, id int64, digest string) error {
	if f.lastTag == nil {
		f.lastTag = map[int64]string{}
	}
	f.lastTag[id] = digest
	return nil
}

func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil }

type startCall struct {
	route model.Route
	id    int64
	start protocol.JobStart
}

type fakeHub struct {
	route         model.Route
	routeErr      error
	startErr      error
	starts        []startCall
	agentProtocol int   // when set, Route fails with ErrAgentTooOld for a higher minProtocol
	mins          []int // minProtocol of every Route call
}

func (h *fakeHub) Route(_ model.Service, minProtocol int) (model.Route, error) {
	h.mins = append(h.mins, minProtocol)
	if h.agentProtocol != 0 && minProtocol > h.agentProtocol {
		if minProtocol >= protocol.MinSelfUpdateVersion {
			return model.Route{}, model.ErrAgentTooOldForSelfUpdate
		}
		if minProtocol >= protocol.MinPinnedRefVersion {
			return model.Route{}, model.ErrAgentTooOldForPinned
		}
		return model.Route{}, model.ErrAgentTooOld
	}
	return h.route, h.routeErr
}

func (h *fakeHub) StartJob(r model.Route, id int64, s protocol.JobStart) error {
	h.starts = append(h.starts, startCall{r, id, s})
	return h.startErr
}

type fakePub struct{ topics []events.Topic }

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

var (
	composeSvc = model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
		AgentID: ptr(int64(7)), ActualRef: "gitea.shcizo.se/shcizo/slusk:1.4.2@sha256:d1"}
	bareSvc = model.Service{ID: 3, Kind: protocol.KindCompose, ComposeProject: "db", ComposeService: "pg",
		AgentID: ptr(int64(7)), ActualRef: "postgres"}
	swarmSvc = model.Service{ID: 2, Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea",
		ActualRef: "gitea/gitea:1.27.3@sha256:87a6"}
)

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

func newService(st *fakeStore, h *fakeHub) (*Service, *fakePub) {
	pub := &fakePub{}
	return New(st, h, pub, selfImage, slog.New(slog.NewTextHandler(io.Discard, nil))), pub
}

func TestUpdatable(t *testing.T) {
	tests := []struct {
		name   string
		svc    model.Service
		reason string
	}{
		{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
		{"agent's own Compose service", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, ""},
		{"agent's own Swarm service", model.Service{Kind: protocol.KindSwarm, ActualRef: "a:1", Self: true}, model.ReasonSelfSwarm},
		{"same repo as the agent image", model.Service{Kind: protocol.KindCompose,
			ActualRef: "gitea.shcizo.se/shcizo/package-updater:1.2.0@sha256:abc"}, ""},
		{"built from source", model.Service{Kind: protocol.KindCompose, ActualRef: "slusk:1", Built: true}, model.ReasonBuilt},
		{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
		{"digest without tag", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, model.ReasonNoTag},
		{"bare image means latest", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres"}, ""},
		{"bare image on a registry with port", model.Service{Kind: protocol.KindCompose, ActualRef: "localhost:5000/foo"}, ""},
		{"tag", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:16"}, ""},
		{"tag and digest", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:17.10@sha256:abc"}, ""},
		{"registry with port and tag", model.Service{Kind: protocol.KindSwarm, ActualRef: "localhost:5000/foo:v1"}, ""},
	}
	s, _ := newService(&fakeStore{}, &fakeHub{})
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := s.Updatable(tt.svc)
			if tt.reason == "" {
				assert.NoError(t, err)
				return
			}
			var nu *model.NotUpdatableError
			require.True(t, errors.As(err, &nu))
			assert.Equal(t, tt.reason, nu.Reason)
		})
	}
}

func TestTrigger(t *testing.T) {
	ok := model.Route{AgentID: 7, ConnID: 70}
	activeJob := model.Job{ID: 99, Status: model.JobRunning}

	tests := []struct {
		name       string
		serviceID  int64
		hub        *fakeHub
		active     *model.Job
		wantErr    error
		wantStatus model.JobStatus
		wantError  string
		wantStart  *protocol.JobStart
		wantCreate bool
	}{
		{name: "compose happy path", serviceID: 1, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true,
			wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4.2"}},
		{name: "swarm", serviceID: 2, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true,
			wantStart: &protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27.3"}},
		{name: "bare image deploys latest", serviceID: 3, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true,
			wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "db", ComposeService: "pg", Ref: "postgres:latest"}},
		{name: "active job", serviceID: 1, hub: &fakeHub{route: ok}, active: &activeJob, wantErr: model.ErrJobActive, wantStatus: model.JobRunning},
		{name: "agent offline", serviceID: 1, hub: &fakeHub{routeErr: model.ErrAgentOffline}, wantStatus: model.JobFailed,
			wantError: "agent offline", wantCreate: true},
		{name: "agent too old", serviceID: 1, hub: &fakeHub{routeErr: model.ErrAgentTooOld}, wantStatus: model.JobFailed,
			wantError: "agent too old for updates", wantCreate: true},
		{name: "start fails", serviceID: 1, hub: &fakeHub{route: ok, startErr: model.ErrAgentOffline}, wantStatus: model.JobFailed,
			wantError: "agent offline", wantCreate: true,
			wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4.2"}},
		{name: "unknown service", serviceID: 5, hub: &fakeHub{route: ok}, wantErr: model.ErrNotFound},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			st := &fakeStore{
				services: map[int64]model.Service{1: composeSvc, 2: swarmSvc, 3: bareSvc},
				jobs:     map[int64]model.Job{99: activeJob},
				active:   tt.active,
			}
			s, pub := newService(st, tt.hub)

			job, err := s.Trigger(context.Background(), tt.serviceID, "sub-1")

			if tt.wantErr != nil {
				assert.ErrorIs(t, err, tt.wantErr)
			} else {
				require.NoError(t, err)
			}
			if tt.wantErr == nil || tt.active != nil {
				assert.Equal(t, tt.wantStatus, job.Status)
				assert.Equal(t, tt.wantError, job.Error)
			}
			if tt.wantStart == nil {
				assert.Empty(t, tt.hub.starts)
			} else {
				require.Len(t, tt.hub.starts, 1)
				assert.Equal(t, *tt.wantStart, tt.hub.starts[0].start)
				assert.Equal(t, ok, tt.hub.starts[0].route)
				assert.Equal(t, job.ID, tt.hub.starts[0].id)
			}
			if tt.wantCreate {
				require.Len(t, st.created, 1)
				assert.Equal(t, "sub-1", st.created[0].RequestedBy)
				assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
				assert.Equal(t, st.services[tt.serviceID].ActualRef, st.created[0].FromRef)
				assert.Contains(t, pub.topics, events.Jobs)
				assert.Contains(t, pub.topics, events.JobTopic(job.ID))
			} else {
				assert.Empty(t, st.created)
				assert.Empty(t, pub.topics)
			}
		})
	}
}

// The routed agent can be revoked between Route and CreateJob; the store then
// refuses the job. Trigger records a failed job that was never sent.
func TestTrigger_AgentRevokedAfterRoute(t *testing.T) {
	st := &fakeStore{
		services:      map[int64]model.Service{1: composeSvc},
		jobs:          map[int64]model.Job{},
		revokedAgents: map[int64]bool{7: true},
	}
	h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
	s, _ := newService(st, h)

	job, err := s.Trigger(context.Background(), 1, "sub-1")

	require.NoError(t, err)
	assert.Equal(t, model.JobFailed, job.Status)
	assert.Equal(t, "agent offline", job.Error)
	assert.Nil(t, job.AgentID)
	assert.Empty(t, h.starts)
}

func TestTrigger_NotUpdatableCreatesNoJob(t *testing.T) {
	st := &fakeStore{
		services: map[int64]model.Service{1: {ID: 1, Kind: protocol.KindContainer, ActualRef: "redis:7"}},
		jobs:     map[int64]model.Job{},
	}
	h := &fakeHub{}
	s, pub := newService(st, h)

	_, err := s.Trigger(context.Background(), 1, "sub-1")

	var nu *model.NotUpdatableError
	require.True(t, errors.As(err, &nu))
	assert.Empty(t, st.created)
	assert.Empty(t, h.starts)
	assert.Empty(t, pub.topics)
}

func TestStart_WebhookTriggerWithExplicitRef(t *testing.T) {
	st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
	h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
	s, _ := newService(st, h)

	job, err := s.Start(context.Background(), composeSvc, StartRequest{
		Trigger: model.TriggerWebhook, Ref: "gitea.shcizo.se/shcizo/slusk:2.0.0"})

	require.NoError(t, err)
	assert.Equal(t, model.JobQueued, job.Status)
	require.Len(t, st.created, 1)
	assert.Equal(t, model.TriggerWebhook, st.created[0].Trigger)
	assert.Empty(t, st.created[0].RequestedBy)
	assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", st.created[0].ToRef)
	require.Len(t, h.starts, 1)
	assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", h.starts[0].start.Ref)
}

func TestStart_ExplicitRefBypassesOnlyTheNoTagRefusal(t *testing.T) {
	tests := []struct {
		name   string
		svc    model.Service
		reason string // "" means the job starts
	}{
		{"digest only", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""},
		{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
		{"own Swarm service", model.Service{Kind: protocol.KindSwarm, ActualRef: "a:1", Self: true}, model.ReasonSelfSwarm},
		{"own Compose service", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, ""},
		{"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt},
		{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.svc.ID = 1
			st := &fakeStore{services: map[int64]model.Service{1: tt.svc}, jobs: map[int64]model.Job{}}
			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
			s, _ := newService(st, h)

			_, err := s.Start(context.Background(), tt.svc, StartRequest{Trigger: model.TriggerWebhook, Ref: "x/y:2"})

			if tt.reason == "" {
				require.NoError(t, err)
				assert.Len(t, st.created, 1)
				return
			}
			var nu *model.NotUpdatableError
			require.True(t, errors.As(err, &nu))
			assert.Equal(t, tt.reason, nu.Reason)
			assert.Empty(t, st.created)
		})
	}
}

func TestStart_WithoutRefStillRefusesNoTag(t *testing.T) {
	svc := model.Service{ID: 1, Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}
	s, _ := newService(&fakeStore{jobs: map[int64]model.Job{}}, &fakeHub{})

	_, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerWebhook})

	var nu *model.NotUpdatableError
	require.True(t, errors.As(err, &nu))
	assert.Equal(t, model.ReasonNoTag, nu.Reason)
}

func TestAutoAllowed(t *testing.T) {
	tests := []struct {
		name   string
		svc    model.Service
		reason string
	}{
		{"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved},
		{"own Swarm service", model.Service{Kind: protocol.KindSwarm, ActualRef: "a:1", Self: true}, model.ReasonSelfSwarm},
		{"own Compose service", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, ""},
		{"agent image repo", model.Service{Kind: protocol.KindCompose, ActualRef: "gitea.shcizo.se/shcizo/package-updater:latest"}, ""},
		{"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt},
		{"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer},
		{"digest without tag is allowed", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""},
		{"tag", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:16"}, ""},
	}
	s, _ := newService(&fakeStore{}, &fakeHub{})
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			err := s.AutoAllowed(tt.svc)
			if tt.reason == "" {
				assert.NoError(t, err)
				return
			}
			var nu *model.NotUpdatableError
			require.True(t, errors.As(err, &nu))
			assert.Equal(t, tt.reason, nu.Reason)
		})
	}
}

func TestFollowedRef(t *testing.T) {
	assert.Equal(t, "postgres:16", FollowedRef("postgres:16@sha256:abc"))
	assert.Equal(t, "postgres:latest", FollowedRef("postgres"))
	assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo"))
	assert.Empty(t, FollowedRef("redis@sha256:abc"))
}

func TestStart_RoutesWithTheMinimumProtocolOfTheRef(t *testing.T) {
	tests := []struct {
		name string
		ref  string // "" means the followed tag
		want int
	}{
		{"followed tag", "", protocol.MinJobVersion},
		{"repo:tag", "x/y:2", protocol.MinJobVersion},
		{"repo:tag@digest", "x/y:2@sha256:a", protocol.MinPinnedRefVersion},
		{"repo@digest", "x/y@sha256:a", protocol.MinPinnedRefVersion},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
			s, _ := newService(st, h)

			_, err := s.Start(context.Background(), composeSvc, StartRequest{Trigger: model.TriggerManual, Ref: tt.ref})

			require.NoError(t, err)
			assert.Equal(t, []int{tt.want}, h.mins)
		})
	}
}

func TestStart_SelfServiceNeedsASelfUpdatingAgent(t *testing.T) {
	self := composeSvc
	self.Self = true
	ok := model.Route{AgentID: 7, ConnID: 70}

	t.Run("routes with the self-update protocol and marks the job", func(t *testing.T) {
		st := &fakeStore{services: map[int64]model.Service{1: self}, jobs: map[int64]model.Job{}}
		h := &fakeHub{route: ok}
		s, _ := newService(st, h)

		_, err := s.Start(context.Background(), self, StartRequest{Trigger: model.TriggerManual, Ref: "x/y:2@sha256:a"})

		require.NoError(t, err)
		assert.Equal(t, []int{protocol.MinSelfUpdateVersion}, h.mins, "also above the pinned-ref minimum")
		require.Len(t, h.starts, 1)
		assert.True(t, h.starts[0].start.Self)
	})
	t.Run("a service with the agent image repo needs a self-updating agent but is not marked", func(t *testing.T) {
		svc := composeSvc
		svc.ActualRef = selfImage
		st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}}
		h := &fakeHub{route: ok}
		s, _ := newService(st, h)

		_, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerManual})

		require.NoError(t, err)
		assert.Equal(t, []int{protocol.MinSelfUpdateVersion}, h.mins)
		require.Len(t, h.starts, 1)
		assert.False(t, h.starts[0].start.Self)
	})
	t.Run("a service that is not the agent's own is not marked", func(t *testing.T) {
		st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}}
		h := &fakeHub{route: ok}
		s, _ := newService(st, h)

		_, err := s.Start(context.Background(), composeSvc, StartRequest{Trigger: model.TriggerManual})

		require.NoError(t, err)
		assert.False(t, h.starts[0].start.Self)
	})
	t.Run("an older agent fails the job with a clear error", func(t *testing.T) {
		st := &fakeStore{services: map[int64]model.Service{1: self}, jobs: map[int64]model.Job{}}
		h := &fakeHub{route: ok, agentProtocol: 3}
		s, _ := newService(st, h)

		job, err := s.Start(context.Background(), self, StartRequest{Trigger: model.TriggerManual})

		require.NoError(t, err)
		assert.Equal(t, model.JobFailed, job.Status)
		assert.Equal(t, "agent too old to update itself", job.Error)
		assert.Empty(t, h.starts)
	})
}

func TestTriggerVersion_Ref(t *testing.T) {
	const repo = "gitea.shcizo.se/shcizo/slusk"
	noTag := model.Service{ID: 4, Kind: protocol.KindCompose, ComposeProject: "n", ComposeService: "n",
		AgentID: ptr(int64(7)), ImageRepo: repo, ActualRef: repo + "@sha256:d1"}
	svc := composeSvc
	svc.ImageRepo = repo
	tests := []struct {
		name    string
		svc     model.Service
		version model.ImageVersion
		wantRef string
		wantMin int
	}{
		{"tag and digest", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
		{"tag only", svc, model.ImageVersion{ID: 10, Repo: repo, Tag: "main"}, repo + ":main", protocol.MinJobVersion},
		{"digest only", svc, model.ImageVersion{ID: 10, Repo: repo, Digest: "sha256:old"}, repo + "@sha256:old", protocol.MinPinnedRefVersion},
		{"service that follows no tag", noTag, model.ImageVersion{ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"}, repo + ":1.3.0@sha256:old", protocol.MinPinnedRefVersion},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			st := &fakeStore{
				services: map[int64]model.Service{tt.svc.ID: tt.svc},
				jobs:     map[int64]model.Job{},
				versions: map[int64]model.ImageVersion{10: tt.version},
			}
			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
			s, _ := newService(st, h)

			job, err := s.TriggerVersion(context.Background(), tt.svc.ID, 10, "sub-1")

			require.NoError(t, err)
			assert.Equal(t, model.JobQueued, job.Status)
			require.Len(t, st.created, 1)
			assert.Equal(t, model.TriggerManual, st.created[0].Trigger)
			assert.Equal(t, "sub-1", st.created[0].RequestedBy)
			assert.Equal(t, tt.wantRef, st.created[0].ToRef)
			require.Len(t, h.starts, 1)
			assert.Equal(t, tt.wantRef, h.starts[0].start.Ref)
			assert.Equal(t, []int{tt.wantMin}, h.mins)
		})
	}
}

func TestTriggerVersion_NotFound(t *testing.T) {
	const repo = "gitea.shcizo.se/shcizo/slusk"
	svc := composeSvc
	svc.ImageRepo = repo
	tests := []struct {
		name      string
		serviceID int64
		versionID int64
	}{
		{"unknown service", 9, 10},
		{"unknown version", 1, 99},
		{"version of another repo", 1, 11},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			st := &fakeStore{
				services: map[int64]model.Service{1: svc},
				jobs:     map[int64]model.Job{},
				versions: map[int64]model.ImageVersion{
					10: {ID: 10, Repo: repo, Tag: "1"},
					11: {ID: 11, Repo: "evil/other", Tag: "1"},
				},
			}
			h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}
			s, pub := newService(st, h)

			_, err := s.TriggerVersion(context.Background(), tt.serviceID, tt.versionID, "sub-1")

			assert.ErrorIs(t, err, model.ErrNotFound)
			assert.Empty(t, st.created, "no job")
			assert.Empty(t, h.starts)
			assert.Empty(t, pub.topics)
		})
	}
}

func TestTriggerVersion_RequiresPinnedProtocol(t *testing.T) {
	const repo = "gitea.shcizo.se/shcizo/slusk"
	svc := composeSvc
	svc.ImageRepo = repo
	st := &fakeStore{
		services: map[int64]model.Service{1: svc},
		jobs:     map[int64]model.Job{},
		versions: map[int64]model.ImageVersion{
			10: {ID: 10, Repo: repo, Tag: "1.3.0", Digest: "sha256:old"},
			11: {ID: 11, Repo: repo, Tag: "main"},
		},
	}
	h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2}
	s, _ := newService(st, h)

	pinned, err := s.TriggerVersion(context.Background(), 1, 10, "sub-1")
	require.NoError(t, err)
	assert.Equal(t, model.JobFailed, pinned.Status)
	assert.Equal(t, "agent too old to deploy a pinned version (needs protocol 3)", pinned.Error)
	assert.Empty(t, h.starts, "nothing sent to the old agent")

	plain, err := s.TriggerVersion(context.Background(), 1, 11, "sub-1")
	require.NoError(t, err)
	assert.Equal(t, model.JobQueued, plain.Status)
}