DockMV/internal/job/manager.go

package job

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"io"
	"sort"
	"sync"
	"time"
)

// Manager owns every job in the process.
type Manager struct {
	mu   sync.RWMutex
	jobs map[string]*Job
	// keep bounds how many finished jobs are retained.
	keep int
}

// NewManager creates an empty job manager.
func NewManager() *Manager {
	return &Manager{jobs: map[string]*Job{}, keep: 50}
}

// ErrNotFound is returned for an unknown job id.
var ErrNotFound = errors.New("job not found")

// Run creates a job and executes fn in the background. fn receives a context
// that is canceled when the job is canceled, and the job handle for progress
// reporting.
func (m *Manager) Run(parent context.Context, kind Kind, title string, dryRun bool, fn func(context.Context, *Job) error) *Job {
	j := newJob(newID(), kind, title, dryRun)
	ctx, cancel := context.WithCancel(parent)
	j.cancel = cancel

	m.mu.Lock()
	m.jobs[j.snap.ID] = j
	m.mu.Unlock()
	m.prune()

	go func() {
		defer cancel()
		j.start()
		err := fn(ctx, j)
		if err == nil && ctx.Err() != nil {
			err = context.Canceled
		}
		if errors.Is(err, context.Canceled) {
			err = context.Canceled
		}
		j.finish(err)
	}()
	return j
}

// Get returns a job by id.
func (m *Manager) Get(id string) (*Job, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	j, ok := m.jobs[id]
	if !ok {
		return nil, ErrNotFound
	}
	return j, nil
}

// List returns every job, newest first.
func (m *Manager) List() []Snapshot {
	m.mu.RLock()
	jobs := make([]*Job, 0, len(m.jobs))
	for _, j := range m.jobs {
		jobs = append(jobs, j)
	}
	m.mu.RUnlock()

	out := make([]Snapshot, 0, len(jobs))
	for _, j := range jobs {
		s := j.Snapshot()
		// The list view does not need the full log.
		if len(s.Log) > 5 {
			s.Log = s.Log[len(s.Log)-5:]
		}
		out = append(out, s)
	}
	sort.Slice(out, func(i, k int) bool { return out[i].CreatedAt.After(out[k].CreatedAt) })
	return out
}

// Delete removes a finished job. A running job is canceled instead.
func (m *Manager) Delete(id string) error {
	m.mu.Lock()
	j, ok := m.jobs[id]
	if !ok {
		m.mu.Unlock()
		return ErrNotFound
	}
	if !j.Snapshot().State.Terminal() {
		m.mu.Unlock()
		j.Cancel()
		return nil
	}
	delete(m.jobs, id)
	m.mu.Unlock()
	return nil
}

// prune drops the oldest finished jobs once the retention limit is exceeded.
func (m *Manager) prune() {
	m.mu.Lock()
	defer m.mu.Unlock()
	if len(m.jobs) <= m.keep {
		return
	}
	type entry struct {
		id string
		at time.Time
	}
	var finished []entry
	for id, j := range m.jobs {
		s := j.Snapshot()
		if s.State.Terminal() {
			finished = append(finished, entry{id, s.CreatedAt})
		}
	}
	sort.Slice(finished, func(i, k int) bool { return finished[i].at.Before(finished[k].at) })
	for i := 0; i < len(finished) && len(m.jobs) > m.keep; i++ {
		delete(m.jobs, finished[i].id)
	}
}

func newID() string {
	b := make([]byte, 8)
	if _, err := rand.Read(b); err != nil {
		return hex.EncodeToString([]byte(time.Now().Format("150405.000000")))
	}
	return hex.EncodeToString(b)
}

// CountingReader wraps a reader and reports every read to a job step, which is
// how transfer progress reaches the UI.
type CountingReader struct {
	R   io.Reader
	Job *Job
	St  *Step

	pending   int64
	lastFlush time.Time
}

// NewCountingReader builds a progress-reporting reader.
func NewCountingReader(r io.Reader, j *Job, st *Step) *CountingReader {
	return &CountingReader{R: r, Job: j, St: st, lastFlush: time.Now()}
}

// Read implements io.Reader, batching counter updates so a fast transfer does
// not flood subscribers with notifications.
func (c *CountingReader) Read(p []byte) (int, error) {
	n, err := c.R.Read(p)
	if n > 0 {
		c.pending += int64(n)
		if c.pending >= 4<<20 || time.Since(c.lastFlush) > 200*time.Millisecond {
			c.flush()
		}
	}
	if err != nil {
		c.flush()
	}
	return n, err
}

// Flush pushes any buffered byte count to the job.
func (c *CountingReader) Flush() { c.flush() }

func (c *CountingReader) flush() {
	if c.pending == 0 {
		return
	}
	c.Job.AddBytes(c.St, c.pending)
	c.pending = 0
	c.lastFlush = time.Now()
}