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

// Package jobs decides whether a service may be updated and starts update
// jobs on the right agent.
package jobs

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

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

// Store is the storage jobs needs. *store.Store satisfies it.
type Store interface {
	GetService(ctx context.Context, id int64) (model.Service, error)
	CreateJob(ctx context.Context, j model.NewJob) (model.Job, error)
	FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error)
	GetJob(ctx context.Context, id int64) (model.Job, error)
	GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error)
	SetLastTagDigest(ctx context.Context, id int64, digest string) error
}

// Hub routes jobs to agent connections. *hub.Hub satisfies it.
type Hub interface {
	// Route picks the connection for svc among agents speaking at least
	// protocol version minProtocol: model.ErrAgentOffline or
	// model.ErrAgentTooOld when there is none.
	Route(svc model.Service, minProtocol int) (model.Route, error)
	// StartJob sends job.start over exactly the routed connection.
	// model.ErrAgentOffline if that connection is gone or was replaced.
	StartJob(r model.Route, jobID int64, start protocol.JobStart) error
}

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

// Service starts update jobs.
type Service struct {
	store    Store
	hub      Hub
	pub      Publisher
	selfRepo string
	log      *slog.Logger
	now      func() time.Time
}

// New returns a Service. agentImage is the master's AGENT_IMAGE; services
// whose image repo equals its repo are the updater itself and are refused.
func New(store Store, hub Hub, pub Publisher, agentImage string, log *slog.Logger) *Service {
	return &Service{
		store: store, hub: hub, pub: pub, log: log, now: time.Now,
		selfRepo: imageref.Parse(agentImage).Repo,
	}
}

// target splits a service's actual ref into the repo and tag to deploy. A bare
// image ("postgres", no tag and no digest) means "latest", as it does for
// Docker. A ref with a digest but no tag follows no tag (ok is false).
func target(actualRef string) (repo, tag string, ok bool) {
	ref := imageref.Parse(actualRef)
	switch {
	case ref.Tag != "":
		return ref.Repo, ref.Tag, true
	case ref.Digest == "":
		return ref.Repo, "latest", true
	}
	return ref.Repo, "", false
}

// FollowedRef returns "repo:tag" for the tag a service with this actual ref
// follows (a bare image follows "latest"), or "" when it follows no tag.
func FollowedRef(actualRef string) string {
	repo, tag, ok := target(actualRef)
	if !ok {
		return ""
	}
	return repo + ":" + tag
}

// Updatable returns nil or a *model.NotUpdatableError. It looks only at the
// service row, so the UI can call it for every page render. It answers for an
// update to the service's own followed tag (the Update button).
func (s *Service) Updatable(svc model.Service) error {
	return s.updatable(svc, false)
}

// AutoAllowed returns nil or a *model.NotUpdatableError saying why the
// service can never be updated automatically. Unlike Updatable it does not
// refuse a service that follows no tag: a webhook that names a tag can still
// update it.
func (s *Service) AutoAllowed(svc model.Service) error {
	return s.updatable(svc, true)
}

// updatable checks the service row. explicitRef means the caller supplies the
// target ref, so a service that follows no tag is fine.
func (s *Service) updatable(svc model.Service, explicitRef bool) error {
	switch {
	case svc.RemovedAt != nil:
		return &model.NotUpdatableError{Reason: model.ReasonRemoved}
	case svc.Self || imageref.Parse(svc.ActualRef).Repo == s.selfRepo:
		return &model.NotUpdatableError{Reason: model.ReasonSelf}
	case svc.Built:
		return &model.NotUpdatableError{Reason: model.ReasonBuilt}
	case svc.Kind == protocol.KindContainer:
		return &model.NotUpdatableError{Reason: model.ReasonContainer}
	}
	if _, _, ok := target(svc.ActualRef); !ok && !explicitRef {
		return &model.NotUpdatableError{Reason: model.ReasonNoTag}
	}
	return nil
}

// Restorable returns nil or a *model.NotUpdatableError saying why the service
// cannot be restored to its desired version: it must have drifted, and the
// desired version is deployed like an explicit ref, so a service that follows
// no tag is fine. The service page calls it for every render.
func (s *Service) Restorable(svc model.Service) error {
	if svc.RemovedAt != nil {
		return &model.NotUpdatableError{Reason: model.ReasonRemoved}
	}
	if !svc.Drifted() {
		return &model.NotUpdatableError{Reason: model.ReasonNotDrifted}
	}
	if err := s.updatable(svc, true); err != nil {
		return err
	}
	if imageref.Parse(svc.DesiredRef).Repo == s.selfRepo {
		return &model.NotUpdatableError{Reason: model.ReasonSelf}
	}
	return nil
}

// RestoreDesired starts a manual job that deploys the drifted service's
// desired ref again. The ref is read from the row, never from the caller. It
// returns model.ErrNotFound for an unknown service and otherwise what Start
// returns, with a *model.NotUpdatableError from Restorable if it cannot be
// restored.
//
// When the service had drifted to another digest of the same tag, that digest
// becomes last_tag_digest, so the poller does not deploy it again before the
// registry shows a newer one (the same rule as for any rollback). It is not
// written when last_tag_digest holds back a newer digest, that is when it is
// set and differs from the digest the desired ref names.
func (s *Service) RestoreDesired(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) {
	svc, err := s.store.GetService(ctx, serviceID)
	if err != nil {
		return model.Job{}, err
	}
	if err := s.Restorable(svc); err != nil {
		return model.Job{}, err
	}
	job, err := s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: svc.DesiredRef})
	if err != nil {
		return job, err
	}
	if d := driftedTagDigest(svc); d != "" && d != svc.LastTagDigest && holdsNothingNewer(svc) {
		if err := s.store.SetLastTagDigest(ctx, svc.ID, d); err != nil {
			s.log.Warn("last_tag_digest_not_set", "service_id", svc.ID, "error", err)
		}
	}
	return job, nil
}

// holdsNothingNewer reports whether last_tag_digest may be overwritten by a
// restore: it is empty or equals the digest the desired ref names. Anything
// else is a digest the poller must keep holding back.
func holdsNothingNewer(svc model.Service) bool {
	return svc.LastTagDigest == "" || svc.LastTagDigest == imageref.Parse(svc.DesiredRef).Digest
}

// driftedTagDigest returns the digest a drifted service runs when it runs the
// same tag as desired at another digest, else "".
func driftedTagDigest(svc model.Service) string {
	tag := FollowedRef(svc.DesiredRef)
	if tag == "" || tag != FollowedRef(svc.ActualRef) {
		return ""
	}
	d := imageref.Parse(svc.ActualRef).Digest
	if d == "" {
		d = svc.ActualDigest
	}
	if !imageref.IsDigest(d) || d == imageref.Parse(svc.DesiredRef).Digest {
		return ""
	}
	return d
}

// StartRequest says who asks for an update and what to deploy.
type StartRequest struct {
	Trigger     model.JobTrigger
	RequestedBy string // OIDC subject for manual jobs; "" otherwise
	// Ref is the ref to deploy ("repo:tag", "repo:tag@digest" or
	// "repo@digest"). Empty means the service's followed tag, derived from
	// its actual ref.
	Ref string
}

// Trigger starts a manual update of the service to its current tag. It
// returns what Start returns, and model.ErrNotFound for an unknown service.
func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) {
	svc, err := s.store.GetService(ctx, serviceID)
	if err != nil {
		return model.Job{}, err
	}
	return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy})
}

// TriggerVersion starts a manual update of the service to exactly the image
// version versionID, which must be a version of the service's own repo; the
// ref is built from the stored version, never taken from the caller. It
// returns model.ErrNotFound for an unknown service, an unknown version or a
// version of another repo, and otherwise what Start returns. A version
// without a tag leaves the service following no tag.
func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error) {
	svc, err := s.store.GetService(ctx, serviceID)
	if err != nil {
		return model.Job{}, err
	}
	v, err := s.store.GetImageVersion(ctx, versionID)
	if err != nil {
		return model.Job{}, err
	}
	if v.Repo != svc.ImageRepo {
		return model.Job{}, model.ErrNotFound
	}
	return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: v.Ref()})
}

// Start starts an update of svc. It returns the job, which may already be
// failed ("agent offline", "agent too old for updates", or "agent too old to
// deploy a pinned version" when the ref has a digest or no tag and the agent
// speaks protocol version 2); a
// *model.NotUpdatableError when the service cannot be updated (no job
// created); the active job and model.ErrJobActive when one is queued or
// running. Poll and webhook jobs are refused for a drifted service; only a
// person starts one then.
func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest) (model.Job, error) {
	if req.Trigger != model.TriggerManual && svc.Drifted() {
		return model.Job{}, &model.NotUpdatableError{Reason: model.ReasonDrifted}
	}
	if err := s.updatable(svc, req.Ref != ""); err != nil {
		return model.Job{}, err
	}
	ref := req.Ref
	if ref == "" {
		ref = FollowedRef(svc.ActualRef) // updatable checked the tag
	}

	minProtocol := protocol.MinJobVersion
	if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" {
		minProtocol = protocol.MinPinnedRefVersion
	}
	r, routeErr := s.hub.Route(svc, minProtocol)
	nj := model.NewJob{
		ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref,
		Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(),
	}
	if routeErr == nil {
		nj.AgentID = &r.AgentID
	}
	job, err := s.store.CreateJob(ctx, nj)
	if errors.Is(err, model.ErrAgentOffline) {
		// The routed agent was revoked after Route. Record the attempt as a
		// failed job that was never sent anywhere.
		routeErr, nj.AgentID = err, nil
		job, err = s.store.CreateJob(ctx, nj)
	}
	if err != nil {
		return job, err
	}

	failure := routeErr
	if failure == nil {
		start := protocol.JobStart{Kind: svc.Kind, Ref: ref}
		if svc.Kind == protocol.KindSwarm {
			start.SwarmClusterID, start.SwarmService = svc.SwarmClusterID, svc.SwarmService
		} else {
			start.ComposeProject, start.ComposeService = svc.ComposeProject, svc.ComposeService
		}
		if err := s.hub.StartJob(r, job.ID, start); err != nil {
			s.log.Warn("job_start_failed", "job_id", job.ID, "error", err)
			failure = model.ErrAgentOffline
		}
	}
	if failure != nil {
		if _, err := s.store.FailJob(ctx, job.ID, failure.Error(), s.now()); err != nil {
			return model.Job{}, err
		}
	}
	s.pub.Publish(events.Jobs)
	s.pub.Publish(events.JobTopic(job.ID))
	return s.store.GetJob(ctx, job.ID)
}