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

// Hub routes jobs to agent connections. *hub.Hub satisfies it.
type Hub interface {
	// Route picks the connection for svc: model.ErrAgentOffline or
	// model.ErrAgentTooOld when there is none.
	Route(svc model.Service) (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
}

// Updatable returns nil or a *model.NotUpdatableError. It looks only at the
// service row, so the UI can call it for every page render.
func (s *Service) Updatable(svc model.Service) 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.Kind == protocol.KindContainer:
		return &model.NotUpdatableError{Reason: model.ReasonContainer}
	}
	if _, _, ok := target(svc.ActualRef); !ok {
		return &model.NotUpdatableError{Reason: model.ReasonNoTag}
	}
	return nil
}

// Trigger starts a manual update of the service to its current tag. It
// returns the job, which may already be failed ("agent offline", "agent too
// old for updates"); a *model.NotUpdatableError when the service cannot be
// updated (no job created); the active job and model.ErrJobActive when one
// is queued or running; 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
	}
	if err := s.Updatable(svc); err != nil {
		return model.Job{}, err
	}
	repo, tag, _ := target(svc.ActualRef) // Updatable checked the tag
	ref := repo + ":" + tag

	r, routeErr := s.hub.Route(svc)
	nj := model.NewJob{
		ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref,
		Trigger: model.TriggerManual, RequestedBy: 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)
}