EvoBGP/internal/jobs/durable.go

package jobs

import (
	"context"
	"os"
	"strconv"
	"strings"
	"time"

	"evobgp/internal/logging"
	"evobgp/internal/observability"
	"evobgp/internal/repository"
)

// Adopt registers a durable job_audit claim into the in-process registry and starts the worker.
// Used after SKIP LOCKED claim so two API processes can share the queue without losing work on restart.
func (r *Registry) Adopt(j *Job) bool {
	if r == nil || j == nil || j.ID == "" {
		return false
	}
	r.mu.Lock()
	if _, exists := r.byID[j.ID]; exists {
		r.mu.Unlock()
		return false
	}
	if j.Status == "" {
		j.Status = StatusQueued
	}
	if j.Meta == nil {
		j.Meta = map[string]any{}
	}
	r.byID[j.ID] = j
	if j.IdempotencyKey != nil && *j.IdempotencyKey != "" {
		r.byIdempo[idempoKey{tenant: j.TenantID, key: *j.IdempotencyKey}] = j
	}
	if isRefreshKind(j.Kind) {
		r.inflightRefresh[j.TenantID]++
	}
	workerStart := r.workerStart
	r.mu.Unlock()

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

// StartDurableQueueLoop periodically reclaims stale running rows and claims queued job_audit work (PG SKIP LOCKED).
// No-op when audit is nil. Interval from EVOBGP_JOB_RECLAIM_INTERVAL (default 20s); stale from EVOBGP_JOB_STALE_AFTER (default 15m).
func StartDurableQueueLoop(ctx context.Context, reg *Registry, audit *repository.JobAuditWriter) {
	if ctx == nil || reg == nil || audit == nil {
		return
	}
	interval := 20 * time.Second
	if s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_RECLAIM_INTERVAL")); s != "" {
		if d, err := time.ParseDuration(s); err == nil && d > 0 {
			interval = d
		}
	}
	staleAfter := 15 * time.Minute
	if s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_STALE_AFTER")); s != "" {
		if d, err := time.ParseDuration(s); err == nil && d > 0 {
			staleAfter = d
		}
	}
	limit := 8
	if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_JOB_CLAIM_LIMIT"))); err == nil && n > 0 {
		limit = n
	}
	grace := 30 * time.Second
	if s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_CLAIM_GRACE")); s != "" {
		if d, err := time.ParseDuration(s); err == nil && d >= 0 {
			grace = d
		}
	}
	log := logging.With("component", "jobs.durable")
	run := func() {
		cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
		defer cancel()
		n, err := audit.ReclaimStaleRunning(cctx, staleAfter)
		if err != nil {
			log.Warn("reclaim stale running failed", "err", err)
		} else if n > 0 {
			log.Info("reclaimed stale running jobs", "count", n)
		}
		claimed, err := audit.ClaimQueued(cctx, limit, grace)
		if err != nil {
			log.Warn("claim queued failed", "err", err)
			return
		}
		for i := range claimed {
			c := claimed[i]
			j := &Job{
				ID:             c.ID,
				TenantID:       c.TenantID,
				Kind:           c.Kind,
				Status:         StatusQueued,
				IdempotencyKey: c.IdempotencyKey,
				ModuleID:       c.ModuleID,
				CreatedAt:      c.CreatedAt,
				Meta:           c.Meta,
			}
			if !reg.Adopt(j) {
				// Already local — leave DB running; local worker owns it.
				continue
			}
			log.Info("adopted durable job", "job_id", j.ID, "kind", j.Kind, "tenant_id", j.TenantID)
		}
		counts, err := audit.CountByStatus(cctx)
		if err == nil {
			observability.RecordJobAuditDepth(counts)
		}
	}
	go func() {
		run()
		t := time.NewTicker(interval)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				run()
			}
		}
	}()
}