EvoBGP/internal/jobs/worker_test.go

package jobs

import (
	"context"
	"fmt"
	"strings"
	"testing"
	"time"

	"evobgp/internal/store"
)

func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) {
	t.Setenv("EVOBGP_ASN_RESOLVE", "0")
	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "") // skip bird binary path in deploy_apply
	t.Setenv("EVOBGP_JOB_MAX_CONCURRENT", "8")

	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, modIP, _, _ := m.DemoIDs()
	for _, mod := range m.ListModules(tenant) {
		if mod == nil || mod.ID == modIP {
			continue
		}
		disabled := false
		if _, err := m.UpdateModule(tenant, mod.ID, &store.ModulePatch{Enabled: &disabled}); err != nil {
			t.Fatal(err)
		}
	}

	mod2, err := m.CreateModule(tenant, &store.Module{Type: "IP_RANGES", Name: "extra-ip", Enabled: true, Priority: 30})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := m.CreateIPRangeEntry(tenant, modIP, &store.IPRangeEntry{Prefix: "10.0.0.0/24"}); err != nil {
		t.Fatal(err)
	}
	if _, err := m.CreateIPRangeEntry(tenant, mod2.ID, &store.IPRangeEntry{Prefix: "192.168.0.0/24"}); err != nil {
		t.Fatal(err)
	}

	// Hold workers until both jobs are enqueued so inflightRefresh=2 before either
	// finishModuleRefreshSuccess. Otherwise a fast ingest can finalize+deploy before
	// the second Enqueue — sequential refreshes correctly produce two deploy_apply jobs.
	start := make(chan struct{})
	wk := &Worker{Store: m}
	reg := NewRegistry(func(j *Job) {
		<-start
		wk.Process(j)
	})
	wk.Registry = reg

	mid1 := modIP
	mid2 := mod2.ID
	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid1, map[string]any{"module_id": modIP}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid2, map[string]any{"module_id": mod2.ID}); err != nil {
		t.Fatal(err)
	}
	close(start)

	waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, 2)

	deployJobs, _, _ := reg.List(tenant, "", KindDeployApply, "", 100)
	if len(deployJobs) != 1 {
		t.Fatalf("want exactly one deploy_apply job, got %d", len(deployJobs))
	}

	var deferred, withDeployID int
	refreshJobs, _, _ := reg.List(tenant, "", KindModuleRefresh, "", 100)
	for _, j := range refreshJobs {
		if j == nil {
			continue
		}
		snap := j.Snapshot()
		if st, _ := snap["status"].(string); st != StatusSucceeded {
			continue
		}
		meta, _ := snap["meta"].(map[string]any)
		if v, ok := meta["deploy_apply_deferred"].(bool); ok && v {
			deferred++
		}
		if _, ok := meta["deploy_apply_job_id"].(string); ok {
			withDeployID++
		}
	}
	if deferred != 1 || withDeployID != 1 {
		t.Fatalf("want one deferred and one with deploy_apply_job_id, got deferred=%d deploy_meta=%d", deferred, withDeployID)
	}
}

func waitSucceededModuleRefreshCount(t *testing.T, reg *Registry, tenant string, want int) {
	t.Helper()
	waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, want)
}

func waitSucceededJobsByKindCount(t *testing.T, reg *Registry, tenant, kind string, want int) {
	t.Helper()
	deadline := time.Now().Add(30 * time.Second)
	for time.Now().Before(deadline) {
		jobs, _, _ := reg.List(tenant, StatusSucceeded, kind, "", 100)
		if len(jobs) >= want {
			return
		}
		all, _, _ := reg.List(tenant, "", kind, "", 100)
		for _, j := range all {
			if j == nil {
				continue
			}
			snap := j.Snapshot()
			if st, _ := snap["status"].(string); st != StatusFailed {
				continue
			}
			t.Fatalf("%s job failed: %#v", kind, snap["error"])
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("timeout waiting for %s jobs", kind)
}

func TestPeerReconcile_RendersFromLatestRevisionAndQueuesDeploy(t *testing.T) {
	t.Setenv("EVOBGP_ASN_RESOLVE", "0")
	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "") // skip bird binary path in deploy_apply

	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, modIP, _, _ := m.DemoIDs()
	for _, mod := range m.ListModules(tenant) {
		if mod == nil || mod.ID == modIP {
			continue
		}
		disabled := false
		if _, err := m.UpdateModule(tenant, mod.ID, &store.ModulePatch{Enabled: &disabled}); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := m.CreateIPRangeEntry(tenant, modIP, &store.IPRangeEntry{Prefix: "10.10.0.0/24"}); err != nil {
		t.Fatal(err)
	}
	peer, err := m.CreatePeer(tenant, &store.BGPPeer{
		Name:      "peer-a",
		Neighbor:  "192.0.2.2",
		RemoteASN: 64512,
		Enabled:   true,
	})
	if err != nil {
		t.Fatal(err)
	}

	wk := &Worker{Store: m}
	reg := NewRegistry(wk.Process)
	wk.Registry = reg

	// Create baseline revision from enabled IP module.
	mid := modIP
	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid, map[string]any{"module_id": modIP}); err != nil {
		t.Fatal(err)
	}
	waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, 1)
	beforeRevs, _, _ := m.ListRevisions(tenant, "", "", 200)

	enabled := false
	if _, err := m.UpdatePeer(tenant, peer.ID, &store.PeerPatch{Enabled: &enabled}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := reg.Enqueue(tenant, KindPeerReconcile, nil, nil, map[string]any{"trigger": "peer_patch"}); err != nil {
		t.Fatal(err)
	}
	waitSucceededJobsByKindCount(t, reg, tenant, KindPeerReconcile, 1)

	afterRevs, _, _ := m.ListRevisions(tenant, "", "", 200)
	if len(afterRevs) <= len(beforeRevs) {
		t.Fatalf("expected new revision after peer_reconcile, before=%d after=%d", len(beforeRevs), len(afterRevs))
	}
	peerJobs, _, _ := reg.List(tenant, StatusSucceeded, KindPeerReconcile, "", 10)
	if len(peerJobs) == 0 {
		t.Fatal("expected succeeded peer_reconcile job")
	}
	meta := peerJobs[0].Snapshot()["meta"].(map[string]any)
	if _, ok := meta["revision_id"].(string); !ok {
		t.Fatalf("expected revision_id in peer_reconcile meta, got %#v", meta)
	}
	if title, ok := meta["job_title"].(string); !ok || title != "Обновление BGP пиров" {
		t.Fatalf("expected job_title in peer_reconcile meta, got %#v", meta["job_title"])
	}
	if _, ok := meta["deploy_apply_job_id"].(string); !ok {
		t.Fatalf("expected deploy_apply_job_id in peer_reconcile meta, got %#v", meta)
	}
}

type renderHookStore struct {
	*store.Memory
	onCreate func(prefixes []store.PrefixRow)
}

func (s *renderHookStore) CreateRenderRevision(ctx context.Context, revisionID, tenantID, moduleID string, parent *string, contentHash string, preview map[string]string, prefixes []store.PrefixRow) error {
	if err := s.Memory.CreateRenderRevision(ctx, revisionID, tenantID, moduleID, parent, contentHash, preview, prefixes); err != nil {
		return err
	}
	if s.onCreate != nil {
		copied := append([]store.PrefixRow(nil), prefixes...)
		s.onCreate(copied)
	}
	return nil
}

func newIPRenderFixture(t *testing.T) (tenant, module string, mem *store.Memory) {
	t.Helper()
	t.Setenv("EVOBGP_ASN_RESOLVE", "0")
	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "")
	mem = store.NewMemory()
	mem.SeedDemo()
	var modIP string
	tenant, _, modIP, _, _ = mem.DemoIDs()
	module = modIP
	for _, mod := range mem.ListModules(tenant) {
		if mod == nil || mod.ID == module {
			continue
		}
		disabled := false
		if _, err := mem.UpdateModule(tenant, mod.ID, &store.ModulePatch{Enabled: &disabled}); err != nil {
			t.Fatal(err)
		}
	}
	return tenant, module, mem
}

func installModuleSnapshot(t *testing.T, mem *store.Memory, tenant, module, prefix string) {
	t.Helper()
	err := mem.SetModulePrefixSnapshot(context.Background(), tenant, module, "snap-"+prefix, []store.PrefixRow{{
		Prefix: prefix,
		Source: "ip_range",
	}})
	if err != nil {
		t.Fatalf("snapshot %s: %v", prefix, err)
	}
}

func revisionHasPrefix(t *testing.T, mem *store.Memory, tenant, revID, prefix string) bool {
	t.Helper()
	rows, _, _ := mem.ListRevisionPrefixes(tenant, revID, "", 100)
	for _, row := range rows {
		if row.Prefix == prefix {
			return true
		}
	}
	return false
}

func TestRenderTenantUntilQuietOverlappingIngests(t *testing.T) {
	t.Run("latest snapshot", func(t *testing.T) {
		tenant, module, mem := newIPRenderFixture(t)
		prefixes := []string{"10.8.0.0/24", "10.9.0.0/24", "10.10.0.0/24", "10.11.0.0/24"}
		installModuleSnapshot(t, mem, tenant, module, prefixes[0])
		hook := &renderHookStore{Memory: mem}
		var wk *Worker
		step := 0
		hook.onCreate = func([]store.PrefixRow) {
			step++
			if step >= len(prefixes) {
				return
			}
			installModuleSnapshot(t, mem, tenant, module, prefixes[step])
			gate := wk.tenantRefreshGate(tenant)
			gate.mu.Lock()
			gate.dirty = true
			gate.mu.Unlock()
		}
		wk = &Worker{Store: hook, Registry: NewRegistry(nil)}
		j := &Job{ID: "refresh-overlap", TenantID: tenant, Kind: KindModuleRefresh, Status: StatusRunning}
		wk.finishModuleRefreshSuccess(j, module)
		if j.Status != StatusSucceeded {
			errText := ""
			if j.Error != nil {
				errText = *j.Error
			}
			t.Fatalf("status=%s err=%s", j.Status, errText)
		}
		if step != len(prefixes) {
			t.Fatalf("render commits=%d, want %d so three overlapping ingests are included", step, len(prefixes))
		}
		rev := j.metaString("revision_id")
		if !revisionHasPrefix(t, mem, tenant, rev, prefixes[len(prefixes)-1]) {
			t.Fatalf("deployed revision %s missing latest prefix %s", rev, prefixes[len(prefixes)-1])
		}
		if revisionHasPrefix(t, mem, tenant, rev, prefixes[len(prefixes)-2]) {
			t.Fatalf("deployed revision %s still has stale prefix %s", rev, prefixes[len(prefixes)-2])
		}
		deployID := j.metaString("deploy_apply_job_id")
		deploy, err := wk.Registry.Get(tenant, deployID)
		if err != nil {
			t.Fatal(err)
		}
		if deploy.metaString("revision_id") != rev {
			t.Fatalf("deploy revision=%s, want %s", deploy.metaString("revision_id"), rev)
		}
	})

	t.Run("cancel does not deploy", func(t *testing.T) {
		tenant, module, mem := newIPRenderFixture(t)
		installModuleSnapshot(t, mem, tenant, module, "10.12.0.0/24")
		hook := &renderHookStore{Memory: mem}
		var wk *Worker
		var j *Job
		commits := 0
		hook.onCreate = func([]store.PrefixRow) {
			commits++
			installModuleSnapshot(t, mem, tenant, module, "10.13.0.0/24")
			gate := wk.tenantRefreshGate(tenant)
			gate.mu.Lock()
			gate.dirty = true
			gate.mu.Unlock()
			j.RequestCancel()
		}
		wk = &Worker{Store: hook, Registry: NewRegistry(nil)}
		j = &Job{ID: "refresh-cancel", TenantID: tenant, Kind: KindModuleRefresh, Status: StatusRunning}
		wk.finishModuleRefreshSuccess(j, module)
		if j.Status != StatusFailed {
			t.Fatalf("status=%s, want failed", j.Status)
		}
		if j.Error == nil || !strings.Contains(*j.Error, "newer snapshot") {
			t.Fatalf("error=%v, want superseded render", j.Error)
		}
		if commits != 1 {
			t.Fatalf("render commits=%d, want 1", commits)
		}
		if j.metaString("deploy_apply_job_id") != "" || j.metaString("revision_id") != "" {
			t.Fatalf("cancelled dirty render published meta %#v", j.metaCopy())
		}
		deploys, _, _ := wk.Registry.List(tenant, "", KindDeployApply, "", 10)
		if len(deploys) != 0 {
			t.Fatalf("deploy jobs=%d, want 0", len(deploys))
		}
		revs, _, _ := mem.ListRevisions(tenant, "", "", 50)
		for _, rev := range revs {
			if rev != nil && revisionHasPrefix(t, mem, tenant, rev.ID, "10.13.0.0/24") {
				t.Fatalf("revision %s contains the snapshot that landed during cancel", rev.ID)
			}
		}
	})

	t.Run("cap does not publish", func(t *testing.T) {
		tenant, module, mem := newIPRenderFixture(t)
		installModuleSnapshot(t, mem, tenant, module, "10.20.0.0/24")
		hook := &renderHookStore{Memory: mem}
		var wk *Worker
		commits := 0
		hook.onCreate = func([]store.PrefixRow) {
			commits++
			installModuleSnapshot(t, mem, tenant, module, fmt.Sprintf("10.21.%d.0/24", commits))
			gate := wk.tenantRefreshGate(tenant)
			gate.mu.Lock()
			gate.dirty = true
			gate.mu.Unlock()
		}
		wk = &Worker{Store: hook, Registry: NewRegistry(nil)}
		j := &Job{ID: "refresh-cap", TenantID: tenant, Kind: KindModuleRefresh, Status: StatusRunning}
		wk.finishModuleRefreshSuccess(j, module)
		if j.Status != StatusFailed || j.Error == nil || !strings.Contains(*j.Error, "did not quiesce") {
			errText := ""
			if j.Error != nil {
				errText = *j.Error
			}
			t.Fatalf("status=%s err=%s", j.Status, errText)
		}
		if commits != renderQuietPassLimit {
			t.Fatalf("render commits=%d, want cap %d", commits, renderQuietPassLimit)
		}
		if j.metaString("deploy_apply_job_id") != "" {
			t.Fatal("safety cap published a deploy")
		}
	})
}