EvoBGP/internal/store/memory_telemetry_test.go

package store

import (
	"context"
	"errors"
	"fmt"
	"testing"
	"time"

	"evobgp/internal/birdfmt"
)

func TestMemoryTelemetryFencingRefreshAndDiscovery(t *testing.T) {
	m := NewMemory()
	m.SeedDemo()
	tenant, _, _, _, _ := m.DemoIDs()
	sp, err := m.CreateSpeaker(tenant, &Speaker{Role: "replica", MetaJSON: `{"agent_domain":"node.test","agent_secret":"secret"}`})
	if err != nil {
		t.Fatal(err)
	}
	ctx := context.Background()
	claims, err := m.ClaimTelemetry(ctx, "owner-a", 2, time.Minute, "")
	if err != nil || len(claims) != 1 || claims[0].SpeakerID != sp.ID {
		t.Fatalf("claims=%+v err=%v", claims, err)
	}
	first := claims[0]
	if snaps, err := m.ListTelemetry(ctx, tenant); err != nil || len(snaps) != 1 || snaps[0].ConfigFingerprint != "" {
		t.Fatalf("uncommitted config visible: %+v %v", snaps, err)
	}
	if other, err := m.ClaimTelemetry(ctx, "owner-b", 1, time.Minute, ""); err != nil || len(other) != 0 {
		t.Fatalf("stole lease: %+v %v", other, err)
	}
	if requested, err := m.RequestTelemetryRefresh(ctx, tenant, time.Minute); err != nil || !requested {
		t.Fatalf("refresh=%v err=%v", requested, err)
	}
	if requested, err := m.RequestTelemetryRefresh(ctx, tenant, time.Minute); err != nil || requested {
		t.Fatalf("not coalesced: %v %v", requested, err)
	}
	at := time.Now().UTC()
	result := TelemetryResult{BGP: TelemetryBGP{TelemetryObservation: TelemetryObservation{Source: "agent_bird_protocols", ObservedAt: at, ReceivedAt: at}, Sessions: []birdfmt.BGPSession{{Name: birdfmt.DynamicPeerProtocolPrefix + "test", Neighbor: "192.0.2.1", NeighborAS: 64501, State: "Established"}}}, Health: TelemetryHealth{TelemetryObservation: TelemetryObservation{Source: "agent_health", ObservedAt: at, ReceivedAt: at}, OK: true}}
	if err := m.CompleteTelemetry(ctx, first, result, at.Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	if err := m.CompleteTelemetry(ctx, first, result, at.Add(time.Hour)); !errors.Is(err, ErrTelemetryLeaseLost) {
		t.Fatalf("reused lease: %v", err)
	}
	snaps, err := m.ListTelemetry(ctx, tenant)
	if err != nil || len(snaps) != 1 || snaps[0].ConfigFingerprint == "" || snaps[0].BGP.LastSuccessAt.IsZero() || snaps[0].NextPollAt.After(time.Now().Add(time.Second)) {
		t.Fatalf("snapshot=%+v err=%v", snaps, err)
	}
	discoveries, err := m.ListPeerDiscoveries(tenant, PeerDiscoveryPending)
	if err != nil || len(discoveries) != 1 {
		t.Fatalf("discovery=%+v err=%v", discoveries, err)
	}
	next, err := m.ClaimTelemetry(ctx, "owner-b", 1, time.Minute, "")
	if err != nil || len(next) != 1 || next[0].FencingToken <= first.FencingToken {
		t.Fatalf("new claim=%+v %v", next, err)
	}
	meta := `{"agent_domain":"new.test","agent_secret":"secret"}`
	if _, err := m.UpdateSpeaker(tenant, sp.ID, &SpeakerPatch{MetaJSON: &meta}); err != nil {
		t.Fatal(err)
	}
	if err := m.CompleteTelemetry(ctx, next[0], result, at.Add(time.Hour)); !errors.Is(err, ErrTelemetryLeaseLost) {
		t.Fatalf("config changed, late result accepted: %v", err)
	}
	m.mu.Lock()
	row := m.telemetry[telemetryKey(tenant, sp.ID)]
	row.leaseUntil = time.Now().Add(-time.Second)
	m.telemetry[telemetryKey(tenant, sp.ID)] = row
	m.mu.Unlock()
	newConfig, err := m.ClaimTelemetry(ctx, "owner-c", 1, time.Minute, "")
	if err != nil || len(newConfig) != 1 {
		t.Fatalf("new config claim=%+v %v", newConfig, err)
	}
	failed := TelemetryResult{BGP: TelemetryBGP{TelemetryObservation: TelemetryObservation{Source: "agent_bird_protocols", Error: "timeout", ReceivedAt: time.Now().UTC()}}, Health: TelemetryHealth{TelemetryObservation: TelemetryObservation{Source: "agent_health", Error: "timeout", ReceivedAt: time.Now().UTC()}}}
	if err := m.CompleteTelemetry(ctx, newConfig[0], failed, time.Now().Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
	snaps, err = m.ListTelemetry(ctx, tenant)
	if err != nil || len(snaps) != 1 || !snaps[0].BGP.LastSuccessAt.IsZero() || !snaps[0].Health.LastSuccessAt.IsZero() {
		t.Fatalf("old endpoint success survived config change: %+v %v", snaps, err)
	}
}

func TestMemoryTelemetryLocalAuthority(t *testing.T) {
	m := NewMemory()
	m.SeedDemo()
	tenant, _, _, _, _ := m.DemoIDs()
	sp, err := m.CreateSpeaker(tenant, &Speaker{Role: "master"})
	if err != nil {
		t.Fatal(err)
	}
	ctx := context.Background()
	first, err := m.ClaimTelemetry(ctx, "owner-a", 1, time.Minute, "site-a")
	if err != nil || len(first) != 1 || first[0].TargetID != TelemetryLocalTargetID(tenant, "site-a") || first[0].SpeakerID != sp.ID {
		t.Fatalf("local claim=%+v %v", first, err)
	}
	second, err := m.ClaimTelemetry(ctx, "owner-b", 1, time.Minute, "site-b")
	if err != nil || len(second) != 1 || second[0].TargetID != TelemetryLocalTargetID(tenant, "site-b") {
		t.Fatalf("other authority=%+v %v", second, err)
	}
}

func TestTelemetryDiscoveryBatchRotatesPastStaticPrefix(t *testing.T) {
	sessions := make([]birdfmt.BGPSession, 300)
	for i := range sessions {
		sessions[i] = birdfmt.BGPSession{Name: fmt.Sprintf("static-%d", i)}
	}
	sessions[299] = birdfmt.BGPSession{Name: birdfmt.DynamicPeerProtocolPrefix + "last", Neighbor: "192.0.2.9"}
	batch, cursor := TelemetryDiscoveryBatch(sessions, 0, 256)
	if len(batch) != 1 || batch[0].Neighbor != "192.0.2.9" || cursor != 0 {
		t.Fatalf("batch=%+v cursor=%d", batch, cursor)
	}
	for i := range sessions {
		sessions[i].Name = birdfmt.DynamicPeerProtocolPrefix + fmt.Sprint(i)
	}
	first, cursor := TelemetryDiscoveryBatch(sessions, 0, 256)
	second, next := TelemetryDiscoveryBatch(sessions, cursor, 256)
	if len(first) != 256 || len(second) != 256 || second[0].Name != sessions[256].Name || next != 212 {
		t.Fatalf("rotation first=%d second=%d cursor=%d next=%d", len(first), len(second), cursor, next)
	}
}

func TestTelemetryFingerprintIgnoresDispatchStatus(t *testing.T) {
	sp := &Speaker{Role: "replica", Endpoint: "https://node.test", MetaJSON: `{"agent_domain":"node.test","agent_secret":"secret"}`}
	before := TelemetrySpeakerFingerprint(sp)
	sp.MetaJSON = `{"agent_domain":"node.test","agent_secret":"secret","last_dispatch_at":"2026-10-26T10:00:00Z","last_dispatch_status":"ok"}`
	if TelemetrySpeakerFingerprint(sp) != before {
		t.Fatal("dispatch status invalidated telemetry config")
	}
	sp.Role = "master"
	before = TelemetrySpeakerFingerprint(sp)
	sp.MetaJSON = `{"agent_domain":"other.test","agent_secret":"changed"}`
	if TelemetrySpeakerFingerprint(sp) != before {
		t.Fatal("agent metadata invalidated local BIRD config")
	}
}

func TestMemoryTelemetryExpiredClaimCannotDiscover(t *testing.T) {
	m := NewMemory()
	m.SeedDemo()
	tenant, _, _, _, _ := m.DemoIDs()
	sp, err := m.CreateSpeaker(tenant, &Speaker{Role: "replica", MetaJSON: `{"agent_domain":"node.test","agent_secret":"secret"}`})
	if err != nil {
		t.Fatal(err)
	}
	ctx := context.Background()
	first, err := m.ClaimTelemetry(ctx, "owner-a", 1, time.Minute, "")
	if err != nil || len(first) != 1 || first[0].SpeakerID != sp.ID {
		t.Fatalf("first=%+v %v", first, err)
	}
	m.mu.Lock()
	row := m.telemetry[telemetryKey(tenant, sp.ID)]
	row.leaseUntil = time.Now().Add(-time.Second)
	m.telemetry[telemetryKey(tenant, sp.ID)] = row
	m.mu.Unlock()
	second, err := m.ClaimTelemetry(ctx, "owner-b", 1, time.Minute, "")
	if err != nil || len(second) != 1 {
		t.Fatalf("second=%+v %v", second, err)
	}
	at := time.Now().UTC()
	result := TelemetryResult{BGP: TelemetryBGP{TelemetryObservation: TelemetryObservation{Source: "agent_bird_protocols", ReceivedAt: at}, Sessions: []birdfmt.BGPSession{{Name: birdfmt.DynamicPeerProtocolPrefix + "late", Neighbor: "192.0.2.13"}}}}
	if err := m.CompleteTelemetry(ctx, first[0], result, at.Add(time.Hour)); !errors.Is(err, ErrTelemetryLeaseLost) {
		t.Fatalf("old claim accepted: %v", err)
	}
	if list, err := m.ListPeerDiscoveries(tenant, ""); err != nil || len(list) != 0 {
		t.Fatalf("stale discovery persisted: %+v %v", list, err)
	}
	if err := m.CompleteTelemetry(ctx, second[0], result, at.Add(time.Hour)); err != nil {
		t.Fatal(err)
	}
}