EvoBGP/internal/jobs/worker.go

package jobs

import (
	"context"
	"fmt"
	"net/http"
	"net/netip"
	"os"
	"sort"
	"strings"
	"sync"
	"time"

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

// mergeBirdPostApplyMeta attaches a birdc snapshot after deploy/reload (best-effort).
func mergeBirdPostApplyMeta(j *Job) {
	if strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")) == "" {
		j.mergeMeta(map[string]any{"bird_post_apply_check": "skipped_no_birdc_socket"})
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
	defer cancel()
	st := birdfmt.InspectLocalBird(ctx)
	inner := map[string]any{
		"bgp_established":    st.BGPEstablished,
		"bgp_sessions_total": st.BGPSessionsTotal,
	}
	if st.Error != "" {
		inner["ok"] = false
		inner["error"] = st.Error
	} else {
		inner["ok"] = true
	}
	j.mergeMeta(map[string]any{"bird_post_apply": inner})
}

const (
	KindModuleRefresh    = "module_refresh"
	KindTenantRefresh    = "tenant_refresh"
	KindPeerReconcile    = "peer_reconcile"
	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).
	// Registry is set after BootstrapWorkers creates the job queue; used to chain deploy_apply after refresh/rollback.
	Registry *Registry
	// refreshGate serializes deploy_apply gating after module_refresh per tenant (see finishModuleRefreshSuccess).
	refreshGate sync.Map // map[string]*sync.Mutex
}

type revisionLogEntry struct {
	Kind           string   `json:"kind"`
	Source         string   `json:"source"`
	Community      string   `json:"community"`
	CommunityLabel string   `json:"community_label"`
	PrefixCount    int      `json:"prefix_count"`
	Sample         []string `json:"sample,omitempty"`
	Message        string   `json:"message"`
}

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
		}
		if err := pipeline.RefreshModuleIngest(context.Background(), w.Store, w.httpClient(), j.TenantID, mid); err != nil {
			j.Fail(err.Error())
			return
		}
		w.finishModuleRefreshSuccess(j, mid)
	case KindTenantRefresh:
		w.runTenantRefresh(j)
	case KindPeerReconcile:
		w.runPeerReconcile(j)
	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
		}
		mergeBirdPostApplyMeta(j)
		j.Succeed()
	default:
		j.Fail("unknown job kind")
	}
}

func (w *Worker) runPeerReconcile(j *Job) {
	if w == nil || w.Store == nil {
		j.Fail("worker not configured")
		return
	}
	const peerJobTitle = "Обновление BGP пиров"
	j.mergeMeta(map[string]any{"job_title": peerJobTitle})

	var revID string
	latest, _, _ := w.Store.ListRevisions(j.TenantID, "", "", 1)
	triggerModuleID, err := w.peerTriggerModuleID(j.TenantID, latest)
	if err != nil {
		j.Fail(err.Error())
		return
	}
	if len(latest) == 0 {
		// First run fallback: render full tenant state once if no baseline revision exists yet.
		rid, err := pipeline.RenderTenantRevision(context.Background(), w.Store, w.httpClient(), j.TenantID, triggerModuleID)
		if err != nil {
			j.Fail(err.Error())
			return
		}
		revID = rid
	} else {
		baseRevID := latest[0].ID
		rows := make([]store.PrefixRow, 0, 1024)
		cursor := ""
		for {
			page, next, more := w.Store.ListRevisionPrefixes(j.TenantID, baseRevID, cursor, 2000)
			rows = append(rows, page...)
			if !more || strings.TrimSpace(next) == "" {
				break
			}
			cursor = next
		}
		rid, err := pipeline.RenderTenantRevisionFromPrefixes(context.Background(), w.Store, w.httpClient(), j.TenantID, triggerModuleID, rows)
		if err != nil {
			j.Fail(err.Error())
			return
		}
		revID = rid
	}
	j.mergeMeta(map[string]any{"revision_id": revID})
	if entries, total, err := w.buildRevisionLogEntries(j.TenantID, revID); err == nil {
		j.mergeMeta(map[string]any{
			"log_entries":   entries,
			"log_total":     total,
			"log_generated": time.Now().UTC().Format(time.RFC3339Nano),
		})
	} else {
		j.mergeMeta(map[string]any{"log_build_error": err.Error()})
	}
	j.Succeed()
	w.enqueueDeployAllSpeakers(j, j.TenantID, revID)
}

func (w *Worker) peerTriggerModuleID(tenantID string, latest []*store.Revision) (string, error) {
	if len(latest) > 0 {
		if mid := strings.TrimSpace(latest[0].ModuleID); mid != "" {
			return mid, nil
		}
	}
	for _, mod := range w.Store.ListModules(tenantID) {
		if mod == nil || !mod.Enabled {
			continue
		}
		if strings.TrimSpace(mod.ID) != "" {
			return mod.ID, nil
		}
	}
	for _, mod := range w.Store.ListModules(tenantID) {
		if mod == nil {
			continue
		}
		if strings.TrimSpace(mod.ID) != "" {
			return mod.ID, nil
		}
	}
	return "", fmt.Errorf("missing module_id for peer reconcile")
}

func (w *Worker) tenantRefreshMu(tenantID string) *sync.Mutex {
	v, _ := w.refreshGate.LoadOrStore(tenantID, &sync.Mutex{})
	return v.(*sync.Mutex)
}

// finishModuleRefreshSuccess marks the refresh job and, for the last active refresh in tenant,
// creates one aggregate revision and enqueues a single deploy_apply.
func (w *Worker) runTenantRefresh(j *Job) {
	moduleIDs := moduleIDsFromJobMeta(j.Meta)
	if len(moduleIDs) == 0 {
		j.Fail("missing module_ids in job meta")
		return
	}
	if err := pipeline.RefreshTenantModules(context.Background(), w.Store, w.httpClient(), j.TenantID, moduleIDs); err != nil {
		j.Fail(err.Error())
		return
	}
	trigger, err := pipeline.PickTenantRefreshTriggerModule(w.Store, j.TenantID, moduleIDs)
	if err != nil {
		j.Fail(err.Error())
		return
	}
	j.mergeMeta(map[string]any{"module_ids": moduleIDs, "modules_refreshed": len(moduleIDs)})
	w.finishModuleRefreshSuccess(j, trigger)
}

func moduleIDsFromJobMeta(meta map[string]any) []string {
	if meta == nil {
		return nil
	}
	raw, ok := meta["module_ids"]
	if !ok {
		return nil
	}
	switch v := raw.(type) {
	case []string:
		return v
	case []any:
		var out []string
		for _, x := range v {
			if s, ok := x.(string); ok && strings.TrimSpace(s) != "" {
				out = append(out, strings.TrimSpace(s))
			}
		}
		return out
	default:
		return nil
	}
}

func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
	if w == nil || w.Store == nil {
		j.Succeed()
		return
	}
	mu := w.tenantRefreshMu(j.TenantID)
	mu.Lock()
	defer mu.Unlock()
	deferDeploy := false
	if w.Registry != nil {
		deferDeploy = w.Registry.CountOtherActiveRefresh(j.TenantID, j.ID) > 0
	}
	if deferDeploy {
		j.mergeMeta(map[string]any{
			"deploy_apply_deferred":     true,
			"deploy_apply_defer_reason": "parallel_module_refresh",
		})
		j.Succeed()
		return
	}

	rev, err := pipeline.RenderTenantRevision(context.Background(), w.Store, w.httpClient(), j.TenantID, triggerModuleID)
	if err != nil {
		j.Fail(err.Error())
		return
	}
	j.mergeMeta(map[string]any{"revision_id": rev})
	if entries, total, err := w.buildRevisionLogEntries(j.TenantID, rev); err == nil {
		j.mergeMeta(map[string]any{
			"log_entries":   entries,
			"log_total":     total,
			"log_generated": time.Now().UTC().Format(time.RFC3339Nano),
		})
	} else {
		j.mergeMeta(map[string]any{"log_build_error": err.Error()})
	}
	j.Succeed()
	w.enqueueDeployAllSpeakers(j, j.TenantID, rev)
}

// enqueueDeployAllSpeakers queues the same work as POST /v1/apply (all speakers, no speaker_id).
func (w *Worker) enqueueDeployAllSpeakers(j *Job, tenantID, revID string) {
	if w == nil || w.Registry == nil {
		return
	}
	revID = strings.TrimSpace(revID)
	if revID == "" {
		return
	}
	applyJob, _, err := w.Registry.Enqueue(tenantID, KindDeployApply, nil, nil, map[string]any{
		"revision_id": revID,
	})
	if err != nil {
		j.mergeMeta(map[string]any{"deploy_apply_enqueue_error": err.Error()})
		return
	}
	if applyJob != nil {
		j.mergeMeta(map[string]any{"deploy_apply_job_id": applyJob.ID})
	}
}

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
		}
	}
	applied := make([]string, 0, 8)
	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
		}
		applied = append(applied, speakerID)
		return nil
	}
	if hasSpeaker && spk != "" {
		if err := applyOne(spk); err != nil {
			j.Fail(err.Error())
			return
		}
		mergeBirdPostApplyMeta(j)
		j.Succeed()
		return
	}
	for _, sp := range w.Store.ListSpeakersForTenant(j.TenantID) {
		if err := applyOne(sp.ID); err != nil {
			j.Fail(err.Error())
			return
		}
	}
	j.mergeMeta(map[string]any{
		"apply_summary": map[string]any{
			"revision_id":    revID,
			"speakers_count": len(applied),
			"speaker_ids":    applied,
			"message":        fmt.Sprintf("Ревизия %s применена на %d спикерах", shortID(revID), len(applied)),
		},
	})
	mergeBirdPostApplyMeta(j)
	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.mergeMeta(map[string]any{
		"rollback_summary": map[string]any{
			"source_revision_id": src,
			"new_revision_id":    newID,
			"message":            fmt.Sprintf("Откат %s → %s", shortID(src), shortID(newID)),
		},
	})
	w.enqueueDeployAllSpeakers(j, j.TenantID, newID)
	j.Succeed()
}

// buildCommunityLabelMap maps community UUID -> human-readable title (or BGP community string).
func buildCommunityLabelMap(st store.Backend, tenantID string) map[string]string {
	out := make(map[string]string)
	if st == nil {
		return out
	}
	list, err := st.ListCommunities(tenantID)
	if err != nil || list == nil {
		return out
	}
	for _, c := range list {
		if c == nil {
			continue
		}
		label := strings.TrimSpace(c.Title)
		if label == "" {
			label = strings.TrimSpace(c.Community)
		}
		if label == "" {
			label = c.ID
		}
		out[c.ID] = label
	}
	return out
}

func resolveCommunityLabel(commID string, byID map[string]string) string {
	if commID == "" || commID == "none" {
		return "без community"
	}
	if lbl, ok := byID[commID]; ok && strings.TrimSpace(lbl) != "" {
		return strings.TrimSpace(lbl)
	}
	return commID
}

func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]any, int, error) {
	if w == nil || w.Store == nil {
		return nil, 0, fmt.Errorf("store not configured")
	}
	commLabels := buildCommunityLabelMap(w.Store, tenantID)
	type agg struct {
		kind      string
		source    string
		community string
		count     int
		sample    []string
	}
	groups := map[string]*agg{}
	total := 0
	cursor := ""
	for {
		rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000)
		for _, p := range rows {
			total++
			src := strings.TrimSpace(p.Source)
			comm := "none"
			if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" {
				comm = strings.TrimSpace(*p.CommunityID)
			}
			kind, sourceName := classifySource(src)
			k := kind + "|" + sourceName + "|" + comm
			g, ok := groups[k]
			if !ok {
				g = &agg{kind: kind, source: sourceName, community: comm}
				groups[k] = g
			}
			g.count++
			if len(g.sample) < 5 {
				g.sample = append(g.sample, p.Prefix)
			}
		}
		if !more {
			break
		}
		cursor = next
		if strings.TrimSpace(cursor) == "" {
			break
		}
	}
	keys := make([]string, 0, len(groups))
	for k := range groups {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	out := make([]map[string]any, 0, len(keys))
	for _, k := range keys {
		g := groups[k]
		cl := resolveCommunityLabel(g.community, commLabels)
		msg := humanLogMessage(g.kind, g.source, g.count, cl, g.sample)
		out = append(out, map[string]any{
			"kind":            g.kind,
			"source":          g.source,
			"community":       g.community,
			"community_label": cl,
			"prefix_count":    g.count,
			"sample":          g.sample,
			"message":         msg,
		})
	}
	return out, total, nil
}

func classifySource(src string) (kind, name string) {
	switch {
	case strings.HasPrefix(src, "as:"):
		return "asn", strings.TrimPrefix(src, "as:")
	case strings.HasPrefix(src, "domain:"):
		return "domain", strings.TrimPrefix(src, "domain:")
	case strings.HasPrefix(src, "cdn:"):
		return "cdn", strings.TrimPrefix(src, "cdn:")
	case src == "ip_range":
		return "ip_range", "manual_ranges"
	default:
		if src == "" {
			return "unknown", "unknown"
		}
		return "source", src
	}
}

// humanLogMessage builds a Russian log line; communityLabel is already resolved (title or BGP value).
func humanLogMessage(kind, source string, count int, communityLabel string, sample []string) string {
	switch kind {
	case "asn":
		return fmt.Sprintf("AS%s: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel)
	case "domain":
		ips := strings.Join(prettyDomainSample(sample), " ")
		if ips == "" {
			ips = "—"
		}
		return fmt.Sprintf("%s: IP (%s) → добавлено в сообщество «%s»", source, ips, communityLabel)
	case "cdn":
		return fmt.Sprintf("CDN «%s»: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel)
	case "ip_range":
		return fmt.Sprintf("Статические диапазоны: добавлено %d префиксов в сообщество «%s»", count, communityLabel)
	default:
		return fmt.Sprintf("%s: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel)
	}
}

func prettyDomainSample(sample []string) []string {
	out := make([]string, 0, len(sample))
	for _, s := range sample {
		p, err := netip.ParsePrefix(strings.TrimSpace(s))
		if err != nil {
			out = append(out, s)
			continue
		}
		if (p.Addr().Is4() && p.Bits() == 32) || (p.Addr().Is6() && p.Bits() == 128) {
			out = append(out, p.Addr().String())
			continue
		}
		out = append(out, s)
	}
	return out
}

func shortID(id string) string {
	s := strings.TrimSpace(id)
	if len(s) <= 8 {
		return s
	}
	return s[:8]
}