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) } }