EvoBGP/internal/jobs/job.go

package jobs

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

	"evobgp/internal/observability"

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

// metaString returns a string meta field under lock (safe vs concurrent mergeMeta).
func (j *Job) metaString(key string) string {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.Meta == nil {
		return ""
	}
	s, _ := j.Meta[key].(string)
	return s
}

// metaBool returns a bool meta field under lock.
func (j *Job) metaBool(key string) bool {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.Meta == nil {
		return false
	}
	b, _ := j.Meta[key].(bool)
	return b
}

// metaCopy returns a shallow copy of job meta under lock.
func (j *Job) metaCopy() map[string]any {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.Meta == nil {
		return nil
	}
	out := make(map[string]any, len(j.Meta))
	for k, v := range j.Meta {
		out[k] = v
	}
	return out
}

// 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)
	workerSem   chan struct{}
	onTerminal  func(j *Job)
	onEnqueued  func(j *Job)
	onRunning   func(j *Job)
	// inflightRefresh counts refresh-kind jobs (module_refresh, tenant_refresh) per tenant that
	// have been enqueued but not yet finalized in finishModuleRefreshSuccess. Used for deterministic
	// deploy coalescing under tenantRefreshMu (instead of polling job statuses).
	inflightRefresh map[string]int
}

type idempoKey struct {
	tenant string
	key    string
}

func NewRegistry(workerStart func(j *Job)) *Registry {
	maxWorkers := registryMaxConcurrentJobs()
	return &Registry{
		byID:            make(map[string]*Job),
		byIdempo:        make(map[idempoKey]*Job),
		workerStart:     workerStart,
		workerSem:       make(chan struct{}, maxWorkers),
		inflightRefresh: make(map[string]int),
	}
}

// SetTerminalHook registers a best-effort callback when jobs reach a terminal state.
func (r *Registry) SetTerminalHook(fn func(j *Job)) {
	if r == nil {
		return
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.onTerminal = fn
}

// SetPersistHooks registers best-effort callbacks for job lifecycle persistence.
func (r *Registry) SetPersistHooks(onEnqueued, onRunning, onTerminal func(j *Job)) {
	if r == nil {
		return
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.onEnqueued = onEnqueued
	r.onRunning = onRunning
	if onTerminal != nil {
		r.onTerminal = onTerminal
	}
}

func (r *Registry) fireEnqueued(j *Job) {
	if r == nil || j == nil {
		return
	}
	r.mu.RLock()
	fn := r.onEnqueued
	r.mu.RUnlock()
	if fn != nil {
		fn(j)
	}
}

func (r *Registry) fireRunning(j *Job) {
	if r == nil || j == nil {
		return
	}
	r.mu.RLock()
	fn := r.onRunning
	r.mu.RUnlock()
	if fn != nil {
		fn(j)
	}
}

func (r *Registry) fireTerminal(j *Job) {
	if r == nil || j == nil {
		return
	}
	r.mu.RLock()
	fn := r.onTerminal
	r.mu.RUnlock()
	if fn != nil {
		fn(j)
	}
}

func registryMaxConcurrentJobs() int {
	if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_JOB_MAX_CONCURRENT"))); err == nil && n > 0 {
		return n
	}
	return 8
}

// 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()
	maxJobs := registryMaxJobsFromEnv()
	r.pruneTerminalIfOver(maxJobs)

	if idempotencyKey != nil && *idempotencyKey != "" {
		k := idempoKey{tenant: tenantID, key: *idempotencyKey}
		if existing, ok := r.byIdempo[k]; ok {
			st := existing.statusLocked()
			if st == StatusQueued || st == StatusRunning {
				r.mu.Unlock()
				return existing, false, nil
			}
			delete(r.byIdempo, k)
		}
	}

	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
	if isRefreshKind(kind) {
		r.inflightRefresh[tenantID]++
	}
	r.pruneTerminalIfOver(maxJobs)
	enqueuedHook := r.onEnqueued
	workerStart := r.workerStart
	r.mu.Unlock()

	if enqueuedHook != nil {
		enqueuedHook(j)
	}
	if workerStart != nil {
		go func() {
			r.workerSem <- struct{}{}
			active := len(r.workerSem)
			capacity := cap(r.workerSem)
			observability.RecordJobQueueDepth(active, capacity)
			defer func() {
				<-r.workerSem
				observability.RecordJobQueueDepth(len(r.workerSem), capacity)
			}()
			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
}

// isRefreshKind reports whether a job kind participates in deploy coalescing.
func isRefreshKind(kind string) bool {
	return kind == KindModuleRefresh || kind == KindTenantRefresh
}

// finalizeRefreshCoalesce is called from finishModuleRefreshSuccess under tenantRefreshMu.
// It atomically decrements the per-tenant inflight refresh counter and reports whether the
// caller is the last outstanding refresh for the tenant (and therefore should render+deploy).
//
// Unlike CountOtherActiveRefresh (which polls job statuses and races under -race), this counter
// is incremented in Enqueue under r.mu and decremented here, so the "last one" decision is
// deterministic regardless of how fast each refresh's ingest completes.
func (r *Registry) finalizeRefreshCoalesce(tenantID string) bool {
	if r == nil {
		return true
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	n := r.inflightRefresh[tenantID]
	if n <= 1 {
		// Last (or already-balanced to zero) — clear the slot and let the caller deploy.
		delete(r.inflightRefresh, tenantID)
		return true
	}
	r.inflightRefresh[tenantID] = n - 1
	return false
}

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
}

// RequestCancelAll requests cancellation of all non-terminal jobs (all tenants).
func (r *Registry) RequestCancelAll() int {
	if r == nil {
		return 0
	}
	r.mu.RLock()
	defer r.mu.RUnlock()
	n := 0
	for _, j := range r.byID {
		if j == nil {
			continue
		}
		st := j.statusLocked()
		if st == StatusSucceeded || st == StatusFailed || st == StatusCancelled {
			continue
		}
		j.RequestCancel()
		n++
	}
	return n
}

// ActiveCount returns the number of queued or running jobs.
func (r *Registry) ActiveCount() int {
	if r == nil {
		return 0
	}
	r.mu.RLock()
	defer r.mu.RUnlock()
	n := 0
	for _, j := range r.byID {
		if j == nil {
			continue
		}
		st := j.statusLocked()
		if st == StatusQueued || st == StatusRunning {
			n++
		}
	}
	return n
}

// Drain waits until no queued/running jobs remain or ctx is done.
// Call RequestCancelAll first for a cooperative shutdown.
func (r *Registry) Drain(ctx context.Context) error {
	if r == nil {
		return nil
	}
	ticker := time.NewTicker(50 * time.Millisecond)
	defer ticker.Stop()
	for {
		if r.ActiveCount() == 0 {
			return nil
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}