package store import ( "context" "crypto/sha256" "encoding/hex" "errors" "strconv" "strings" "time" "evobgp/internal/birdfmt" ) // ErrTelemetryLeaseLost means a collector result belongs to an expired or replaced lease. var ErrTelemetryLeaseLost = errors.New("store: telemetry lease lost") // TelemetryLocalTargetID identifies the synthetic local BIRD target of a tenant. func TelemetryLocalTargetID(tenantID, authority string) string { return "cp-local/" + tenantID + "/" + authority } // TelemetryObservation is the most recent attempt and last successful reading of one source. type TelemetryObservation struct { ObservedAt time.Time `json:"observed_at"` ReceivedAt time.Time `json:"received_at"` LastSuccessAt time.Time `json:"last_success_at"` Source string `json:"source"` Error string `json:"error,omitempty"` } // TelemetryBGP is one BIRD protocols measurement. type TelemetryBGP struct { TelemetryObservation Sessions []birdfmt.BGPSession `json:"sessions,omitempty"` } // TelemetryHealth is one legacy agent health measurement. type TelemetryHealth struct { TelemetryObservation OK bool `json:"ok"` LastSyncAt string `json:"last_sync_at,omitempty"` LastAppliedRevisionID string `json:"last_applied_revision_id,omitempty"` } // TelemetrySnapshot contains independent BGP and agent health readings for one target. type TelemetrySnapshot struct { TenantID, TargetID, SpeakerID, Kind string LocalAuthority string ConfigFingerprint string BGP TelemetryBGP Health TelemetryHealth NextPollAt time.Time FailureCount int DiscoveryCursor int } // TelemetrySpeakerFingerprint binds a snapshot to the configuration it observed. func TelemetrySpeakerFingerprint(sp *Speaker) string { if sp == nil { return "" } if strings.EqualFold(strings.TrimSpace(sp.Role), "master") { sum := sha256.Sum256([]byte("master\x00" + sp.Endpoint)) return hex.EncodeToString(sum[:]) } meta := ParseSpeakerMeta(sp.MetaJSON) sum := sha256.Sum256([]byte(sp.Role + "\x00" + sp.Endpoint + "\x00" + meta.AgentDomain + "\x00" + meta.AgentSecret + "\x00" + strconv.Itoa(meta.AgentPort))) return hex.EncodeToString(sum[:]) } // TelemetryClaim fences a bounded collector attempt. Speaker is the current config. type TelemetryClaim struct { TelemetrySnapshot Speaker *Speaker OwnerID string FencingToken int64 } // TelemetryResult is the atomic pair of results of one collector attempt. type TelemetryResult struct { BGP TelemetryBGP Health TelemetryHealth } // TelemetryDiscoveryBatch rotates over long protocol lists without starving later entries. func TelemetryDiscoveryBatch(sessions []birdfmt.BGPSession, cursor, limit int) ([]birdfmt.BGPSession, int) { if len(sessions) == 0 || limit <= 0 { return nil, 0 } if cursor < 0 || cursor >= len(sessions) { cursor = 0 } out := make([]birdfmt.BGPSession, 0, limit) for visited := 0; visited < len(sessions); visited++ { idx := (cursor + visited) % len(sessions) if birdfmt.IsDynamicDiscoverySession(sessions[idx].Name) { out = append(out, sessions[idx]) if len(out) == limit { return out, (idx + 1) % len(sessions) } } } return out, 0 } // TelemetryStore coordinates polling and read-only snapshots across API replicas. type TelemetryStore interface { ClaimTelemetry(context.Context, string, int, time.Duration, string) ([]TelemetryClaim, error) CompleteTelemetry(context.Context, TelemetryClaim, TelemetryResult, time.Time) error ListTelemetry(context.Context, string) ([]TelemetrySnapshot, error) RequestTelemetryRefresh(context.Context, string, time.Duration) (bool, error) }