EvoBGP/internal/jobs/worker.go

package jobs

import (
	"context"
	"net/http"
	"os"
	"strings"
	"time"

	"evobgp/internal/birddeploy"
	"evobgp/internal/birdfmt"
	"evobgp/internal/observability"
	"evobgp/internal/pipeline"
	"evobgp/internal/store"
)

const (
	KindModuleRefresh   = "module_refresh"
	KindDeployApply     = "deploy_apply"
	KindRevisionRollback = "revision_rollback"
	KindBirdReload      = "bird_reload"
)

// Worker executes queued jobs against store.Backend (memory or SQL).
type Worker struct {
	Store       store.Backend
	HTTPClient  *http.Client // optional; CDN refresh uses this (default 45s timeout).
}

var defaultWorkerHTTP = &http.Client{Timeout: 45 * time.Second}

func (w *Worker) httpClient() *http.Client {
	if w != nil && w.HTTPClient != nil {
		return w.HTTPClient
	}
	return defaultWorkerHTTP
}

// 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:
		mid, _ := j.Meta["module_id"].(string)
		if strings.TrimSpace(mid) == "" {
			j.Fail("missing module_id in job meta")
			return
		}
		rev, err := pipeline.RefreshModule(context.Background(), w.Store, w.httpClient(), j.TenantID, mid)
		if err != nil {
			j.Fail(err.Error())
			return
		}
		j.mergeMeta(map[string]any{"revision_id": rev})
		j.Succeed()
	case KindDeployApply:
		w.runDeployApply(j)
	case KindRevisionRollback:
		w.runRollback(j)
	case KindBirdReload:
		sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET"))
		if sock == "" {
			j.Succeed()
			return
		}
		ctl := &birdfmt.BirdCtl{
			Socket: sock,
			Birdc:  strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")),
		}
		if err := ctl.Configure(context.Background()); err != nil {
			j.Fail(err.Error())
			return
		}
		j.Succeed()
	default:
		j.Fail("unknown job kind")
	}
}

func (w *Worker) runDeployApply(j *Job) {
	revID, _ := j.Meta["revision_id"].(string)
	spk, hasSpeaker := j.Meta["speaker_id"].(string)
	if revID == "" {
		j.Fail("missing revision_id in job meta")
		return
	}
	activeDir := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_ACTIVE_DIR"))
	if activeDir != "" {
		revObj, err := w.Store.GetRevision(j.TenantID, revID)
		if err != nil {
			j.Fail(err.Error())
			return
		}
		staging := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_STAGING_DIR"))
		if staging == "" {
			staging = os.TempDir() + "/evobgp-bird-staging"
		}
		cfg := birddeploy.Config{
			ActiveDir:  activeDir,
			StagingDir: staging,
			BirdBin:    strings.TrimSpace(os.Getenv("EVOBGP_BIRD_BIN")),
			BirdcBin:   strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")),
			Socket:     strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")),
		}
		ctl := &birdfmt.BirdCtl{Bird: cfg.BirdBin, Birdc: cfg.BirdcBin, Socket: cfg.Socket}
		if err := birddeploy.ApplyRevision(context.Background(), ctl, revObj, cfg); err != nil {
			j.Fail(err.Error())
			return
		}
	}
	applyOne := func(speakerID string) error {
		if err := w.Store.SetLastAppliedRevision(j.TenantID, speakerID, revID); err != nil {
			return err
		}
		// Replica / node pulls use LatestPublishedRevision; keep pointer in sync with successful deploy.
		if err := w.Store.PublishRevisionForSpeaker(speakerID, revID); 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()
}