DockMV/internal/job/job.go

// Package job tracks long-running migrations and streams their progress to
// the UI. A job is a tree: job -> per-container item -> per-operation step,
// with byte counters on the steps that move data.
package job

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

// State is the lifecycle of a job, item or step.
type State string

const (
	StatePending   State = "pending"
	StateRunning   State = "running"
	StateSucceeded State = "succeeded"
	StateFailed    State = "failed"
	StateSkipped   State = "skipped"
	StateCanceled  State = "canceled"
)

// Terminal reports whether no further transitions are expected.
func (s State) Terminal() bool {
	switch s {
	case StateSucceeded, StateFailed, StateSkipped, StateCanceled:
		return true
	}
	return false
}

// Kind distinguishes the two migration modes.
type Kind string

const (
	KindSSH     Kind = "ssh"
	KindPackage Kind = "package"
	KindRestore Kind = "restore"
)

// Level classifies a log line.
type Level string

const (
	LevelInfo  Level = "info"
	LevelWarn  Level = "warn"
	LevelError Level = "error"
	LevelCmd   Level = "cmd" // a command that was (or would be) run on a host
)

// LogEntry is one line in the job log.
type LogEntry struct {
	Seq     int64     `json:"seq"`
	At      time.Time `json:"at"`
	Level   Level     `json:"level"`
	Item    string    `json:"item,omitempty"`
	Message string    `json:"message"`
}

// Step is one unit of work inside an item, e.g. "transfer volume pgdata".
type Step struct {
	ID         string     `json:"id"`
	Label      string     `json:"label"`
	State      State      `json:"state"`
	BytesDone  int64      `json:"bytesDone"`
	BytesTotal int64      `json:"bytesTotal"` // -1 when unknown
	Error      string     `json:"error,omitempty"`
	StartedAt  *time.Time `json:"startedAt,omitempty"`
	EndedAt    *time.Time `json:"endedAt,omitempty"`
}

// Item is the migration of one container.
type Item struct {
	ID       string   `json:"id"` // container id on the source
	Name     string   `json:"name"`
	State    State    `json:"state"`
	Error    string   `json:"error,omitempty"`
	Steps    []*Step  `json:"steps"`
	Warnings []string `json:"warnings,omitempty"`
}

// Snapshot is the serializable view of a job handed to the UI.
type Snapshot struct {
	ID         string     `json:"id"`
	Kind       Kind       `json:"kind"`
	Title      string     `json:"title"`
	State      State      `json:"state"`
	DryRun     bool       `json:"dryRun"`
	Error      string     `json:"error,omitempty"`
	CreatedAt  time.Time  `json:"createdAt"`
	StartedAt  *time.Time `json:"startedAt,omitempty"`
	EndedAt    *time.Time `json:"endedAt,omitempty"`
	Items      []*Item    `json:"items"`
	Log        []LogEntry `json:"log"`
	BytesDone  int64      `json:"bytesDone"`
	BytesTotal int64      `json:"bytesTotal"`
	// Artifact is the produced package path, for package jobs.
	Artifact string `json:"artifact,omitempty"`
	// ArtifactBytes is the package size on disk.
	ArtifactBytes int64 `json:"artifactBytes,omitempty"`
	Revision      int64 `json:"revision"`
}

// Job is a running or finished migration.
type Job struct {
	mu       sync.RWMutex
	snap     Snapshot
	seq      int64
	revision int64
	maxLog   int

	cancel context.CancelFunc
	done   chan struct{}

	subsMu sync.Mutex
	subs   map[int]chan struct{}
	nextID int
}

func newJob(id string, kind Kind, title string, dryRun bool) *Job {
	return &Job{
		snap: Snapshot{
			ID: id, Kind: kind, Title: title, State: StatePending,
			DryRun: dryRun, CreatedAt: time.Now(), Items: []*Item{}, Log: []LogEntry{},
			BytesTotal: 0,
		},
		maxLog: 5000,
		done:   make(chan struct{}),
		subs:   map[int]chan struct{}{},
	}
}

// ID returns the job identifier.
func (j *Job) ID() string { return j.snap.ID }

// Done is closed once the job reaches a terminal state.
func (j *Job) Done() <-chan struct{} { return j.done }

// Snapshot returns a deep-enough copy for JSON serialization.
func (j *Job) Snapshot() Snapshot {
	j.mu.RLock()
	defer j.mu.RUnlock()
	s := j.snap
	s.Items = make([]*Item, len(j.snap.Items))
	var done, total int64
	for i, it := range j.snap.Items {
		cp := *it
		cp.Steps = make([]*Step, len(it.Steps))
		for k, st := range it.Steps {
			sc := *st
			cp.Steps[k] = &sc
			done += sc.BytesDone
			if sc.BytesTotal > 0 {
				total += sc.BytesTotal
			}
		}
		s.Items[i] = &cp
	}
	s.Log = append([]LogEntry(nil), j.snap.Log...)
	s.BytesDone, s.BytesTotal = done, total
	s.Revision = atomic.LoadInt64(&j.revision)
	return s
}

// Subscribe returns a channel that receives a signal whenever the job changes,
// plus a function to unsubscribe.
func (j *Job) Subscribe() (<-chan struct{}, func()) {
	j.subsMu.Lock()
	defer j.subsMu.Unlock()
	id := j.nextID
	j.nextID++
	ch := make(chan struct{}, 1)
	j.subs[id] = ch
	return ch, func() {
		j.subsMu.Lock()
		defer j.subsMu.Unlock()
		if c, ok := j.subs[id]; ok {
			delete(j.subs, id)
			close(c)
		}
	}
}

func (j *Job) touch() {
	atomic.AddInt64(&j.revision, 1)
	j.subsMu.Lock()
	for _, ch := range j.subs {
		select {
		case ch <- struct{}{}:
		default: // a signal is already pending; the reader will see the latest state
		}
	}
	j.subsMu.Unlock()
}

// Logf appends a line to the job log.
func (j *Job) Logf(level Level, item, format string, args ...any) {
	j.mu.Lock()
	j.seq++
	e := LogEntry{Seq: j.seq, At: time.Now(), Level: level, Item: item, Message: fmt.Sprintf(format, args...)}
	j.snap.Log = append(j.snap.Log, e)
	if len(j.snap.Log) > j.maxLog {
		j.snap.Log = j.snap.Log[len(j.snap.Log)-j.maxLog:]
	}
	j.mu.Unlock()
	j.touch()
}

// AddItem registers a container in the job and returns its handle.
func (j *Job) AddItem(id, name string) *Item {
	j.mu.Lock()
	it := &Item{ID: id, Name: name, State: StatePending, Steps: []*Step{}}
	j.snap.Items = append(j.snap.Items, it)
	j.mu.Unlock()
	j.touch()
	return it
}

// AddStep registers a unit of work under an item. bytesTotal may be -1 when
// the size is not known ahead of time.
func (j *Job) AddStep(it *Item, id, label string, bytesTotal int64) *Step {
	j.mu.Lock()
	st := &Step{ID: id, Label: label, State: StatePending, BytesTotal: bytesTotal}
	it.Steps = append(it.Steps, st)
	j.mu.Unlock()
	j.touch()
	return st
}

// StartStep marks a step as running.
func (j *Job) StartStep(st *Step) {
	now := time.Now()
	j.mu.Lock()
	st.State = StateRunning
	st.StartedAt = &now
	j.mu.Unlock()
	j.touch()
}

// FinishStep closes a step, recording an error when one occurred.
func (j *Job) FinishStep(st *Step, err error) {
	now := time.Now()
	j.mu.Lock()
	st.EndedAt = &now
	if err != nil {
		st.State = StateFailed
		st.Error = err.Error()
	} else {
		st.State = StateSucceeded
		if st.BytesTotal < 0 {
			st.BytesTotal = st.BytesDone
		}
	}
	j.mu.Unlock()
	j.touch()
}

// SkipStep marks a step as deliberately not performed.
func (j *Job) SkipStep(st *Step, reason string) {
	now := time.Now()
	j.mu.Lock()
	st.State = StateSkipped
	st.EndedAt = &now
	st.Error = reason
	j.mu.Unlock()
	j.touch()
}

// AddBytes advances a step's byte counter. It is safe to call at high rates
// from the transfer goroutine.
func (j *Job) AddBytes(st *Step, n int64) {
	j.mu.Lock()
	st.BytesDone += n
	j.mu.Unlock()
	j.touch()
}

// SetItemState transitions an item.
func (j *Job) SetItemState(it *Item, s State, err error) {
	j.mu.Lock()
	it.State = s
	if err != nil {
		it.Error = err.Error()
	}
	j.mu.Unlock()
	j.touch()
}

// AddItemWarning attaches a non-fatal note to an item.
func (j *Job) AddItemWarning(it *Item, format string, args ...any) {
	msg := fmt.Sprintf(format, args...)
	j.mu.Lock()
	it.Warnings = append(it.Warnings, msg)
	j.mu.Unlock()
	j.Logf(LevelWarn, it.ID, "%s", msg)
}

// SetArtifact records the produced package.
func (j *Job) SetArtifact(path string, bytes int64) {
	j.mu.Lock()
	j.snap.Artifact = path
	j.snap.ArtifactBytes = bytes
	j.mu.Unlock()
	j.touch()
}

func (j *Job) start() {
	now := time.Now()
	j.mu.Lock()
	j.snap.State = StateRunning
	j.snap.StartedAt = &now
	j.mu.Unlock()
	j.touch()
}

func (j *Job) finish(err error) {
	now := time.Now()
	j.mu.Lock()
	if j.snap.State.Terminal() {
		j.mu.Unlock()
		return
	}
	j.snap.EndedAt = &now
	switch {
	case err == nil:
		j.snap.State = StateSucceeded
	case err == context.Canceled:
		j.snap.State = StateCanceled
		j.snap.Error = "canceled by operator"
	default:
		j.snap.State = StateFailed
		j.snap.Error = err.Error()
	}
	j.mu.Unlock()
	j.touch()
	close(j.done)
}

// Cancel asks the job to stop. Work already in flight unwinds through context
// cancellation.
func (j *Job) Cancel() {
	if j.cancel != nil {
		j.cancel()
	}
}