feat: enhance evobgp with new command-line tools for bundle management, including pull, verify, and apply functionalities. Update go.mod to include necessary dependencies and complete todos in architecture plan for improved observability and deployment practices.

2026-04-05 14:07:45 +07:00
parent 272542b92a
commit bf52b21150
131 changed files with 9222 additions and 17 deletions
@@ -0,0 +1,5 @@
+package jobs
+
+import "errors"
+
+var ErrNotFound = errors.New("jobs: not found")
@@ -0,0 +1,285 @@
+package jobs
+
+import (
+	"fmt"
+	"sort"
+	"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 (maps to 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
+}
+
+// Registry tracks jobs in memory (microVPS-style single process; swap for PG + SKIP LOCKED later).
+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,
+	}
+}
+
+// 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()
+
+	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
+
+	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
+	}
+	r.mu.RLock()
+	var all []*Job
+	for _, j := range r.byID {
+		if j.TenantID != tenantID {
+			continue
+		}
+		if statusFilter != "" && j.Status != statusFilter {
+			continue
+		}
+		if kindFilter != "" && j.Kind != kindFilter {
+			continue
+		}
+		all = append(all, j)
+	}
+	r.mu.RUnlock()
+
+	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
+}
+
+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
+}
@@ -0,0 +1,98 @@
+package jobs
+
+import (
+	"evobgp/internal/observability"
+	"evobgp/internal/store"
+)
+
+const (
+	KindModuleRefresh   = "module_refresh"
+	KindDeployApply     = "deploy_apply"
+	KindRevisionRollback = "revision_rollback"
+	KindBirdReload      = "bird_reload"
+)
+
+// Worker executes queued jobs against an in-memory store (stub for full render/deploy pipeline).
+type Worker struct {
+	Store *store.Memory
+}
+
+// Process is registered as Registry.workerStart.
+func (w *Worker) Process(j *Job) {
+	defer func() {
+		observability.RecordJobTerminal(j.Kind, j.statusLocked())
+	}()
+
+	if w == nil || w.Store == nil {
+		j.MarkRunning()
+		j.Fail("worker not configured")
+		return
+	}
+	j.MarkRunning()
+	if j.IsCancelRequested() {
+		j.MarkCancelled()
+		return
+	}
+
+	switch j.Kind {
+	case KindModuleRefresh:
+		j.Succeed()
+	case KindDeployApply:
+		w.runDeployApply(j)
+	case KindRevisionRollback:
+		w.runRollback(j)
+	case KindBirdReload:
+		j.Succeed()
+	default:
+		j.Fail("unknown job kind")
+	}
+}
+
+func (w *Worker) runDeployApply(j *Job) {
+	rev, _ := j.Meta["revision_id"].(string)
+	spk, hasSpeaker := j.Meta["speaker_id"].(string)
+	if rev == "" {
+		j.Fail("missing revision_id in job meta")
+		return
+	}
+	applyOne := func(speakerID string) error {
+		if err := w.Store.SetLastAppliedRevision(j.TenantID, speakerID, rev); err != nil {
+			return err
+		}
+		// Replica / node pulls use LatestPublishedRevision; keep pointer in sync with successful deploy.
+		if err := w.Store.PublishRevisionForSpeaker(speakerID, rev); err != nil {
+			return err
+		}
+		return nil
+	}
+	if hasSpeaker && spk != "" {
+		if err := applyOne(spk); err != nil {
+			j.Fail(err.Error())
+			return
+		}
+		j.Succeed()
+		return
+	}
+	for _, sp := range w.Store.ListSpeakersForTenant(j.TenantID) {
+		if err := applyOne(sp.ID); err != nil {
+			j.Fail(err.Error())
+			return
+		}
+	}
+	j.Succeed()
+}
+
+func (w *Worker) runRollback(j *Job) {
+	src, _ := j.Meta["source_revision_id"].(string)
+	if src == "" {
+		j.Fail("missing source_revision_id in job meta")
+		return
+	}
+	newID, err := w.Store.CreateRollbackRevision(j.TenantID, src)
+	if err != nil {
+		j.Fail(err.Error())
+		return
+	}
+	j.mergeMeta(map[string]any{"new_revision_id": newID})
+	j.Succeed()
+}