EvoBGP/internal/jobs/job.go

package jobs

import (
	"fmt"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"
)

// Status values align with OpenAPI JobStatus and DB constraint job_audit_status_chk.
const (
	StatusQueued    = "queued"
	StatusRunning   = "running"
	StatusSucceeded = "succeeded"
	StatusFailed    = "failed"
	StatusCancelled = "cancelled"
)

// Job is the API-facing job model (поля согласованы со схемой job_audit в миграциях; персистенция в БД пока не подключена).
type Job struct {
	ID              string
	TenantID        string
	Kind            string
	Status          string
	IdempotencyKey  *string
	ModuleID        *string
	CreatedAt       time.Time
	StartedAt       *time.Time
	FinishedAt      *time.Time
	Error           *string
	ProgressPct     *int16
	Meta            map[string]any
	cancelRequested bool
	mu              sync.Mutex
}

func (j *Job) MarkRunning() {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.Status != StatusQueued {
		return
	}
	now := time.Now().UTC()
	j.StartedAt = &now
	j.Status = StatusRunning
}

func (j *Job) Succeed() {
	j.mu.Lock()
	defer j.mu.Unlock()
	now := time.Now().UTC()
	j.FinishedAt = &now
	j.Status = StatusSucceeded
}

func (j *Job) Fail(msg string) {
	j.mu.Lock()
	defer j.mu.Unlock()
	now := time.Now().UTC()
	j.FinishedAt = &now
	j.Status = StatusFailed
	j.Error = &msg
}

func (j *Job) MarkCancelled() {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.Status == StatusSucceeded || j.Status == StatusFailed || j.Status == StatusCancelled {
		return
	}
	now := time.Now().UTC()
	j.FinishedAt = &now
	j.Status = StatusCancelled
}

func (j *Job) IsCancelRequested() bool {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.cancelRequested
}

func (j *Job) RequestCancel() bool {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.cancelRequested = true
	if j.Status == StatusQueued {
		now := time.Now().UTC()
		j.FinishedAt = &now
		j.Status = StatusCancelled
		return true
	}
	return false
}

// Snapshot returns a consistent view for JSON serialization (safe under concurrent worker updates).
func (j *Job) mergeMeta(kv map[string]any) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.Meta == nil {
		j.Meta = map[string]any{}
	}
	for k, v := range kv {
		j.Meta[k] = v
	}
}

// statusLocked is used by the worker defer for metrics (any stable terminal or in-flight status).
func (j *Job) statusLocked() string {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.Status
}

func (j *Job) Snapshot() map[string]any {
	j.mu.Lock()
	defer j.mu.Unlock()
	metaCopy := make(map[string]any, len(j.Meta))
	for k, v := range j.Meta {
		metaCopy[k] = v
	}
	m := map[string]any{
		"job_id": j.ID, "kind": j.Kind, "status": j.Status,
		"created_at": j.CreatedAt.UTC().Format(time.RFC3339Nano),
		"meta":       metaCopy,
	}
	if j.IdempotencyKey != nil {
		m["idempotency_key"] = *j.IdempotencyKey
	} else {
		m["idempotency_key"] = nil
	}
	if j.StartedAt != nil {
		m["started_at"] = j.StartedAt.UTC().Format(time.RFC3339Nano)
	} else {
		m["started_at"] = nil
	}
	if j.FinishedAt != nil {
		m["finished_at"] = j.FinishedAt.UTC().Format(time.RFC3339Nano)
	} else {
		m["finished_at"] = nil
	}
	if j.Error != nil {
		m["error"] = *j.Error
	} else {
		m["error"] = nil
	}
	return m
}

var jobRegistryMaxJobsOnce sync.Once
var jobRegistryMaxJobs int

// registryMaxJobsFromEnv returns EVOBGP_JOB_REGISTRY_MAX_JOBS once (0 = без лимита, только завершённые джобы вытесняются).
func registryMaxJobsFromEnv() int {
	jobRegistryMaxJobsOnce.Do(func() {
		s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_REGISTRY_MAX_JOBS"))
		if s == "" {
			return
		}
		n, err := strconv.Atoi(s)
		if err != nil || n <= 0 {
			return
		}
		jobRegistryMaxJobs = n
	})
	return jobRegistryMaxJobs
}

// Registry — in-memory очередь и индекс по idempotency в процессе, где поднят HTTP API (evobgp-api и evobgp-all).
// Отдельные воркеры в reference-профиле не разделяют память с API: scheduler дергает refresh по HTTP; см. docs/architecture.md.
// Запись задач в PostgreSQL job_audit + SKIP LOCKED / внешний брокер — планируемое расширение (архитектурный план §2, §7.10).
type Registry struct {
	mu          sync.RWMutex
	byID        map[string]*Job
	byIdempo    map[idempoKey]*Job
	workerStart func(j *Job)
}

type idempoKey struct {
	tenant string
	key    string
}

func NewRegistry(workerStart func(j *Job)) *Registry {
	return &Registry{
		byID:        make(map[string]*Job),
		byIdempo:    make(map[idempoKey]*Job),
		workerStart: workerStart,
	}
}

// pruneTerminalIfOver удаляет самые старые завершённые джобы (succeeded/failed/cancelled), пока len(byID) > maxJobs.
func (r *Registry) pruneTerminalIfOver(maxJobs int) {
	if r == nil || maxJobs <= 0 || len(r.byID) <= maxJobs {
		return
	}
	type fin struct {
		j *Job
		t time.Time
	}
	var cands []fin
	for _, j := range r.byID {
		st := j.statusLocked()
		if st != StatusSucceeded && st != StatusFailed && st != StatusCancelled {
			continue
		}
		j.mu.Lock()
		ft := j.FinishedAt
		j.mu.Unlock()
		if ft == nil {
			continue
		}
		cands = append(cands, fin{j: j, t: *ft})
	}
	need := len(r.byID) - maxJobs
	if need <= 0 || len(cands) == 0 {
		return
	}
	sort.Slice(cands, func(i, j int) bool { return cands[i].t.Before(cands[j].t) })
	if need > len(cands) {
		need = len(cands)
	}
	for i := 0; i < need; i++ {
		v := cands[i].j
		delete(r.byID, v.ID)
		if v.IdempotencyKey != nil && *v.IdempotencyKey != "" {
			delete(r.byIdempo, idempoKey{tenant: v.TenantID, key: *v.IdempotencyKey})
		}
	}
}

// Enqueue creates a job or returns an existing one for the same idempotency key.
func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, moduleID *string, meta map[string]any) (*Job, bool, error) {
	r.mu.Lock()
	defer r.mu.Unlock()

	maxJobs := registryMaxJobsFromEnv()
	r.pruneTerminalIfOver(maxJobs)

	if idempotencyKey != nil && *idempotencyKey != "" {
		k := idempoKey{tenant: tenantID, key: *idempotencyKey}
		if existing, ok := r.byIdempo[k]; ok {
			return existing, false, nil
		}
	}

	j := &Job{
		ID:             uuid.NewString(),
		TenantID:       tenantID,
		Kind:           kind,
		Status:         StatusQueued,
		IdempotencyKey: idempotencyKey,
		ModuleID:       moduleID,
		CreatedAt:      time.Now().UTC(),
		Meta:           cloneMeta(meta),
	}
	if idempotencyKey != nil && *idempotencyKey != "" {
		r.byIdempo[idempoKey{tenant: tenantID, key: *idempotencyKey}] = j
	}
	r.byID[j.ID] = j
	r.pruneTerminalIfOver(maxJobs)

	if r.workerStart != nil {
		go r.workerStart(j)
	}
	return j, true, nil
}

func cloneMeta(m map[string]any) map[string]any {
	if m == nil {
		return map[string]any{}
	}
	out := make(map[string]any, len(m))
	for k, v := range m {
		out[k] = v
	}
	return out
}

func (r *Registry) Get(tenantID, jobID string) (*Job, error) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	j, ok := r.byID[jobID]
	if !ok || j.TenantID != tenantID {
		return nil, ErrNotFound
	}
	return j, nil
}

func (r *Registry) List(tenantID, statusFilter, kindFilter, cursor string, limit int) ([]*Job, string, bool) {
	if limit <= 0 {
		limit = 50
	}
	// Build candidate set under r.mu, but read mutable status fields (j.Status) outside
	// of r.mu so we don't race with Job.mu-protected writes (MarkRunning/Succeed/Fail).
	r.mu.RLock()
	all := make([]*Job, 0, len(r.byID))
	for _, j := range r.byID {
		if j.TenantID == tenantID {
			all = append(all, j)
		}
	}
	r.mu.RUnlock()

	// Apply filters outside of r.mu to synchronize with Job.mu.
	if statusFilter != "" || kindFilter != "" {
		filtered := all[:0]
		for _, j := range all {
			if kindFilter != "" && j.Kind != kindFilter {
				continue
			}
			if statusFilter != "" && j.statusLocked() != statusFilter {
				continue
			}
			filtered = append(filtered, j)
		}
		all = filtered
	}

	sort.Slice(all, func(i, j int) bool {
		return all[i].CreatedAt.After(all[j].CreatedAt)
	})
	off := 0
	if cursor != "" {
		_ = parseCursor(cursor, &off)
	}
	end := off + limit
	next := ""
	hasMore := false
	if end > len(all) {
		end = len(all)
	} else {
		hasMore = true
		next = formatCursor(end)
	}
	if off >= len(all) {
		return nil, "", false
	}
	return all[off:end], next, hasMore
}

// CountOtherActiveModuleRefresh returns how many module_refresh jobs for the tenant are still
// queued or running, excluding excludeJobID (the current job). Used to batch deploy_apply.
func (r *Registry) CountOtherActiveModuleRefresh(tenantID, excludeJobID string) int {
	return r.CountOtherActiveRefresh(tenantID, excludeJobID)
}

// CountOtherActiveRefresh counts queued/running module_refresh and tenant_refresh jobs for the tenant.
func (r *Registry) CountOtherActiveRefresh(tenantID, excludeJobID string) int {
	if r == nil {
		return 0
	}
	r.mu.RLock()
	candidates := make([]*Job, 0, 8)
	for _, j := range r.byID {
		if j.TenantID != tenantID {
			continue
		}
		if j.Kind != KindModuleRefresh && j.Kind != KindTenantRefresh {
			continue
		}
		if j.ID == excludeJobID {
			continue
		}
		candidates = append(candidates, j)
	}
	r.mu.RUnlock()

	n := 0
	for _, j := range candidates {
		st := j.statusLocked()
		if st == StatusQueued || st == StatusRunning {
			n++
		}
	}
	return n
}

func parseCursor(s string, off *int) error {
	_, err := fmt.Sscanf(s, "%d", off)
	return err
}

func formatCursor(off int) string {
	return fmt.Sprintf("%d", off)
}

// RequestCancel marks a job for cancellation (best-effort).
func (r *Registry) RequestCancel(tenantID, jobID string) (*Job, error) {
	j, err := r.Get(tenantID, jobID)
	if err != nil {
		return nil, err
	}
	j.RequestCancel()
	return j, nil
}