Files
EvoBGP/internal/httpapi/routes.go
T
Denozordec 7a2b015b12
CI / changes (push) Successful in 11s
CI / openapi (push) Has been skipped
CI / go (push) Successful in 59s
CI / docker-web (deploy/docker/evobgp-web/Dockerfile, , evobgp-web) (push) Has been skipped
CI / docker-bird (push) Has been skipped
CI / docker-web (deploy/docker/evobgp-web/Dockerfile, evobgp-all, evobgp-web-all) (push) Has been skipped
CI / bird2 (push) Successful in 43s
CI / docker-go (evobgp-all, 1, deploy/docker/gobinary/Dockerfile, , evobgp-all) (push) Has been skipped
CI / docker-go (evobgp-api, 1, deploy/docker/gobinary/Dockerfile, , evobgp-api) (push) Has been skipped
CI / docker-go (evobgp-deploy, 0, deploy/docker/gobinary/Dockerfile, , evobgp-deploy) (push) Has been skipped
CI / docker-go (evobgp-ingest, 0, deploy/docker/gobinary/Dockerfile, , evobgp-ingest) (push) Has been skipped
CI / docker-go (evobgp-node, 0, deploy/docker/gobinary/Dockerfile, , evobgp-node) (push) Has been skipped
CI / docker-go (evobgp-render, 0, deploy/docker/gobinary/Dockerfile, , evobgp-render) (push) Has been skipped
CI / docker-go (evobgp-scheduler, 0, deploy/docker/gobinary/Dockerfile, , evobgp-scheduler) (push) Has been skipped
CI / docker-go-prime (push) Successful in 55s
CI / docker-go (deploy/docker/evobgp-agent/Dockerfile, , evobgp-agent) (push) Has been cancelled
feat: add tenant refresh functionality to HTTP API and job processing
Implemented a new endpoint for tenant refresh in the HTTP API, allowing for the refresh of modules associated with a tenant. Enhanced job processing to handle tenant refresh jobs, including logic for managing module IDs and job status updates. Updated the scheduler to enqueue tenant refresh jobs based on module due dates, improving the overall efficiency of module management. Additionally, introduced caching for ASN prefix data to optimize performance during refresh operations.
2026-05-19 10:35:20 +07:00

1279 lines
36 KiB
Go

package httpapi
import (
"context"
"crypto/ed25519"
"encoding/base64"
"encoding/json"
"errors"
"io"
"net/http"
"os"
"sort"
"strconv"
"strings"
"time"
"evobgp/internal/birdfmt"
"evobgp/internal/bundle"
"evobgp/internal/jobs"
"evobgp/internal/observability"
"evobgp/internal/pipeline"
"evobgp/internal/store"
)
// Handler returns the root HTTP handler (system routes public; rest under /v1/ authenticated).
func (s *Server) Handler() http.Handler {
v1 := http.NewServeMux()
s.registerV1(v1)
wrappedV1 := http.StripPrefix("/v1", v1)
s.mux.Handle("GET /metrics", observability.MetricsHandler())
s.mux.HandleFunc("GET /v1/health", s.handleHealth)
s.mux.HandleFunc("GET /v1/ready", s.handleReady)
s.mux.HandleFunc("GET /v1/version", s.handleVersion)
s.mux.Handle("/v1/", s.authMiddleware(wrappedV1))
return s.withCORS(observability.HTTPMiddleware(s.mux))
}
// BundlePublicKeyBase64 returns the Ed25519 public key for verifying bundles (share with evobgp-node).
func (s *Server) BundlePublicKeyBase64() string {
pub := s.bundlePriv.Public().(ed25519.PublicKey)
return base64.StdEncoding.EncodeToString(pub)
}
func (s *Server) registerRoutes() {
// routes attached in Handler()
}
func (s *Server) registerV1(m *http.ServeMux) {
m.HandleFunc("GET /modules", s.handleListModules)
m.HandleFunc("GET /router-lists/catalog", s.handleRouterListsCatalog)
m.HandleFunc("GET /modules/{module_id}", s.handleGetModule)
m.HandleFunc("GET /peers", s.handleListPeers)
m.HandleFunc("GET /speakers", s.handleListSpeakers)
m.HandleFunc("POST /modules/{module_id}/refresh", s.handleModuleRefresh)
m.HandleFunc("POST /tenant/refresh", s.handleTenantRefresh)
m.HandleFunc("GET /revisions", s.handleListRevisions)
m.HandleFunc("GET /revisions/{revision_id}", s.handleGetRevision)
m.HandleFunc("GET /revisions/{revision_id}/preview", s.handleRevisionPreview)
m.HandleFunc("GET /revisions/{revision_id}/diagnostic-log", s.handleRevisionDiagnosticLog)
m.HandleFunc("GET /revisions/{revision_a}/diff/{revision_b}", s.handleRevisionDiff)
m.HandleFunc("POST /revisions/{revision_id}/rollback", s.handleRevisionRollback)
m.HandleFunc("POST /apply", s.handleApply)
m.HandleFunc("POST /speakers/{id}/apply", s.handleSpeakerApply)
m.HandleFunc("POST /bird/reload", s.handleBirdReload)
m.HandleFunc("GET /bird/status", s.handleBirdStatus)
m.HandleFunc("GET /jobs", s.handleListJobs)
m.HandleFunc("GET /jobs/{job_id}", s.handleGetJob)
m.HandleFunc("POST /jobs/{job_id}/cancel", s.handleCancelJob)
m.HandleFunc("GET /speakers/{speaker_id}/revisions/latest", s.handleNodeLatestRevision)
m.HandleFunc("GET /speakers/{speaker_id}/bundle/{revision_id}", s.handleNodeBundle)
m.HandleFunc("POST /nodes/enroll", s.handleNodeEnroll)
s.registerCRUDRoutes(m)
}
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}
func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
checks := map[string]string{"store": "ok", "jobs": "memory"}
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
defer cancel()
if s.pgPool != nil {
if err := s.pgPool.Ping(ctx); err != nil {
checks["postgres"] = err.Error()
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "checks": checks})
return
}
checks["postgres"] = "ok"
} else {
checks["store_backend"] = "memory"
}
writeJSON(w, http.StatusOK, map[string]any{"status": "ready", "checks": checks})
}
func (s *Server) handleVersion(w http.ResponseWriter, r *http.Request) {
sha := strings.TrimSpace(os.Getenv("EVOBGP_GIT_SHA"))
if sha == "" {
sha = "unknown"
}
writeJSON(w, http.StatusOK, map[string]string{"api_version": "0.1.0", "git_sha": sha})
}
func moduleJSON(mod *store.Module) map[string]any {
m := map[string]any{
"id": mod.ID,
"type": mod.Type,
"name": mod.Name,
"enabled": mod.Enabled,
"priority": mod.Priority,
"refresh_interval_sec": mod.RefreshIntervalSec,
"cron_expr": mod.CronExpr,
}
if mod.LastRefreshedAt != nil {
m["last_refreshed_at"] = mod.LastRefreshedAt.UTC().Format(time.RFC3339Nano)
} else {
m["last_refreshed_at"] = nil
}
if mod.DefaultCommunityID != nil {
m["default_community_id"] = *mod.DefaultCommunityID
} else {
m["default_community_id"] = nil
}
if mod.DohProfileID != nil {
m["doh_profile_id"] = *mod.DohProfileID
} else {
m["doh_profile_id"] = nil
}
return m
}
func peerJSON(p *store.BGPPeer) map[string]any {
m := map[string]any{
"id": p.ID,
"name": p.Name,
"neighbor": p.Neighbor,
"remote_asn": p.RemoteASN,
"enabled": p.Enabled,
"session_state": p.SessionState,
}
if p.SpeakerID != nil {
m["bgp_speaker_id"] = *p.SpeakerID
} else {
m["bgp_speaker_id"] = nil
}
return m
}
func speakerJSON(sp *store.Speaker) map[string]any {
m := map[string]any{
"id": sp.ID,
"role": sp.Role,
"endpoint": sp.Endpoint,
}
if sp.LastAppliedRevisionID != nil {
m["last_applied_revision_id"] = *sp.LastAppliedRevisionID
} else {
m["last_applied_revision_id"] = nil
}
return m
}
func (s *Server) handleListModules(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
typeFilter := strings.TrimSpace(r.URL.Query().Get("type"))
enabledRaw := strings.TrimSpace(r.URL.Query().Get("enabled"))
var enabledFilter *bool
if enabledRaw != "" {
v, err := strconv.ParseBool(enabledRaw)
if err != nil {
writeProblem(w, http.StatusBadRequest, "Bad Request", "enabled must be boolean")
return
}
enabledFilter = &v
}
mods := s.store.ListModules(a.TenantID)
items := make([]map[string]any, 0, len(mods))
for _, mod := range mods {
if typeFilter != "" && mod.Type != typeFilter {
continue
}
if enabledFilter != nil && mod.Enabled != *enabledFilter {
continue
}
items = append(items, moduleJSON(mod))
}
writeJSON(w, http.StatusOK, map[string]any{
"items": items, "next_cursor": nil, "has_more": false,
})
}
func (s *Server) handleRouterListsCatalog(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
mods := s.store.ListModules(a.TenantID)
moduleItems := make([]map[string]any, 0, len(mods))
domains := make([]map[string]any, 0)
asns := make([]map[string]any, 0)
ipRanges := make([]map[string]any, 0)
for _, mod := range mods {
switch mod.Type {
case "DOMAINS", "AS_PREFIXES", "IP_RANGES":
moduleItems = append(moduleItems, moduleJSON(mod))
default:
continue
}
switch mod.Type {
case "DOMAINS":
list, err := s.store.ListDomainEntries(a.TenantID, mod.ID)
if err != nil {
writeStoreErr(w, err)
return
}
for _, x := range list {
domains = append(domains, map[string]any{
"module_id": mod.ID,
"entry": domainEntryJSON(x),
})
}
case "AS_PREFIXES":
list, err := s.store.ListASEntries(a.TenantID, mod.ID)
if err != nil {
writeStoreErr(w, err)
return
}
for _, x := range list {
asns = append(asns, map[string]any{
"module_id": mod.ID,
"entry": asEntryJSON(x),
})
}
case "IP_RANGES":
list, err := s.store.ListIPRangeEntries(a.TenantID, mod.ID)
if err != nil {
writeStoreErr(w, err)
return
}
for _, x := range list {
ipRanges = append(ipRanges, map[string]any{
"module_id": mod.ID,
"entry": ipRangeJSON(x),
})
}
}
}
comms, err := s.store.ListCommunities(a.TenantID)
if err != nil {
writeStoreErr(w, err)
return
}
communityItems := make([]map[string]any, 0, len(comms))
for _, c := range comms {
communityItems = append(communityItems, map[string]any{
"id": c.ID,
"community": c.Community,
"title": c.Title,
})
}
writeJSON(w, http.StatusOK, map[string]any{
"modules": map[string]any{
"items": moduleItems,
},
"domains": map[string]any{
"items": domains,
},
"asns": map[string]any{
"items": asns,
},
"ip_ranges": map[string]any{
"items": ipRanges,
},
"communities": map[string]any{
"items": communityItems,
},
})
}
func (s *Server) handleGetModule(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
mod, err := s.store.GetModule(a.TenantID, r.PathValue("module_id"))
if err != nil {
if err == store.ErrNotFound || err == store.ErrTenantScope {
writeProblem(w, http.StatusNotFound, "Not Found", "module not found")
return
}
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
writeJSON(w, http.StatusOK, moduleJSON(mod))
}
func (s *Server) handleListPeers(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
peers := s.store.ListPeers(a.TenantID)
liveStates := s.liveBGPProtocolStates(r.Context())
items := make([]map[string]any, 0, len(peers))
for _, p := range peers {
row := peerJSON(p)
if st, ok := liveStates[peerProtocolNameForID(p.ID)]; ok && strings.TrimSpace(st) != "" {
row["session_state"] = strings.TrimSpace(st)
}
items = append(items, row)
}
writeJSON(w, http.StatusOK, map[string]any{
"items": items, "next_cursor": nil, "has_more": false,
})
}
func (s *Server) liveBGPProtocolStates(ctx context.Context) map[string]string {
sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET"))
if sock == "" {
return map[string]string{}
}
out, err := birdfmt.ShowProtocols(ctx, sock, strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")))
if err != nil {
return map[string]string{}
}
return parseBGPProtocolStates(out)
}
// parseBGPProtocolStates parses `birdc show protocols all` summary rows into protocol_name -> state.
func parseBGPProtocolStates(output string) map[string]string {
out := make(map[string]string)
for _, raw := range strings.Split(output, "\n") {
line := strings.TrimSpace(raw)
if line == "" {
continue
}
low := strings.ToLower(line)
if strings.HasPrefix(low, "bird ") || strings.HasPrefix(low, "name ") || strings.HasPrefix(low, "table ") {
continue
}
fields := strings.Fields(line)
if len(fields) < 4 {
continue
}
if !strings.EqualFold(fields[1], "BGP") {
continue
}
state := extractBGPSessionState(line)
if state == "" {
state = fields[3]
}
out[fields[0]] = state
}
return out
}
func extractBGPSessionState(line string) string {
known := []string{
"Established",
"Idle",
"Connect",
"Active",
"OpenSent",
"OpenConfirm",
}
for _, st := range known {
if strings.Contains(line, st) {
return st
}
}
return ""
}
// peerProtocolNameForID must stay in sync with pipeline peer protocol naming.
func peerProtocolNameForID(peerID string) string {
s := strings.ReplaceAll(strings.TrimSpace(peerID), "-", "")
if len(s) > 16 {
s = s[:16]
}
if s == "" {
s = "x"
}
return "evobgp_p_" + s
}
func (s *Server) handleListSpeakers(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
speakers := s.store.ListSpeakersForTenant(a.TenantID)
items := make([]map[string]any, 0, len(speakers))
for _, sp := range speakers {
items = append(items, speakerJSON(sp))
}
writeJSON(w, http.StatusOK, map[string]any{
"items": items, "next_cursor": nil, "has_more": false,
})
}
func (s *Server) handleModuleRefresh(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "editor") {
return
}
moduleID := r.PathValue("module_id")
mod, err := s.store.GetModule(a.TenantID, moduleID)
if err != nil {
if err == store.ErrNotFound || err == store.ErrTenantScope {
writeProblem(w, http.StatusNotFound, "Not Found", "module not found")
return
}
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
idem := r.Header.Get("Idempotency-Key")
var idemPtr *string
if strings.TrimSpace(idem) != "" {
idem = strings.TrimSpace(idem)
idemPtr = &idem
}
mid := mod.ID
j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindModuleRefresh, idemPtr, &mid, map[string]any{"module_id": moduleID})
if err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
w.Header().Set("Location", "/v1/jobs/"+j.ID)
snap := j.Snapshot()
writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]})
}
func (s *Server) handleTenantRefresh(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "editor") {
return
}
var body struct {
ModuleIDs []string `json:"module_ids"`
}
if r.Body != nil && r.ContentLength != 0 {
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&body); err != nil && err != io.EOF {
writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid JSON body")
return
}
}
moduleIDs := body.ModuleIDs
if len(moduleIDs) == 0 {
now := time.Now().UTC()
for _, mod := range s.store.ListModules(a.TenantID) {
if pipeline.ModuleDueForScheduler(mod, now) {
moduleIDs = append(moduleIDs, mod.ID)
}
}
}
if len(moduleIDs) == 0 {
writeJSON(w, http.StatusNoContent, map[string]any{"message": "no modules due for refresh"})
return
}
idem := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
var idemPtr *string
if idem != "" {
idemPtr = &idem
}
j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindTenantRefresh, idemPtr, nil, map[string]any{
"module_ids": moduleIDs,
"trigger": "api",
})
if err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
w.Header().Set("Location", "/v1/jobs/"+j.ID)
snap := j.Snapshot()
writeJSON(w, http.StatusAccepted, map[string]any{
"job_id": snap["job_id"],
"status": snap["status"],
"module_ids": moduleIDs,
"modules_due": len(moduleIDs),
})
}
func (s *Server) handleListRevisions(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
if limit == 0 {
limit = 50
}
cursor := r.URL.Query().Get("cursor")
moduleID := r.URL.Query().Get("module_id")
items, next, more := s.store.ListRevisions(a.TenantID, moduleID, cursor, limit)
out := make([]map[string]any, 0, len(items))
for _, rev := range items {
out = append(out, revisionJSON(rev))
}
writeJSON(w, http.StatusOK, map[string]any{
"items": out, "next_cursor": strPtrOrNull(next), "has_more": more,
})
}
func revisionJSON(rev *store.Revision) map[string]any {
m := map[string]any{
"id": rev.ID,
"content_hash": rev.ContentHash,
"created_at": rev.CreatedAt.UTC().Format(time.RFC3339Nano),
"materialized_prefix_count": rev.MaterializedPrefixCount,
}
if rev.ModuleID != "" {
m["module_id"] = rev.ModuleID
} else {
m["module_id"] = nil
}
if rev.ParentRevisionID != nil {
m["parent_revision_id"] = *rev.ParentRevisionID
} else {
m["parent_revision_id"] = nil
}
return m
}
func strPtrOrNull(s string) any {
if s == "" {
return nil
}
return s
}
// enqueueModuleRefreshIfEnabled queues module_refresh when the module exists and is enabled (best-effort, no HTTP error).
func (s *Server) enqueueModuleRefreshIfEnabled(tenantID, moduleID, trigger string) {
if s.jobs == nil {
return
}
mod, err := s.store.GetModule(tenantID, moduleID)
if err != nil || !mod.Enabled {
return
}
mid := moduleID
_, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, nil, &mid, map[string]any{
"module_id": moduleID,
"trigger": trigger,
})
}
// enqueuePeerReconcile queues fast peer-only reconcile/render (best-effort, no HTTP error).
func (s *Server) enqueuePeerReconcile(tenantID, trigger string) {
if s.jobs == nil {
return
}
_, _, _ = s.jobs.Enqueue(tenantID, jobs.KindPeerReconcile, nil, nil, map[string]any{
"trigger": trigger,
"job_title": "Обновление BGP пиров",
})
}
func (s *Server) handleGetRevision(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
rev, err := s.store.GetRevision(a.TenantID, r.PathValue("revision_id"))
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
writeJSON(w, http.StatusOK, revisionJSON(rev))
}
func (s *Server) handleRevisionPreview(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
rev, err := s.store.GetRevision(a.TenantID, r.PathValue("revision_id"))
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
acc := r.Header.Get("Accept")
if strings.Contains(acc, "text/plain") && !strings.Contains(acc, "application/json") {
var b strings.Builder
for _, k := range sortedFragmentKeys(rev.PreviewFragments) {
b.WriteString("# --- ")
b.WriteString(k)
b.WriteString(" ---\n")
b.WriteString(rev.PreviewFragments[k])
b.WriteByte('\n')
}
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(b.String()))
return
}
obj := make(map[string]any, len(rev.PreviewFragments)+1)
for k, v := range rev.PreviewFragments {
obj[k] = v
}
writeJSON(w, http.StatusOK, obj)
}
func (s *Server) handleRevisionDiagnosticLog(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
revID := r.PathValue("revision_id")
rev, err := s.store.GetRevision(a.TenantID, revID)
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
moduleType := ""
moduleName := ""
if strings.TrimSpace(rev.ModuleID) != "" {
if mod, modErr := s.store.GetModule(a.TenantID, rev.ModuleID); modErr == nil && mod != nil {
moduleType = strings.TrimSpace(mod.Type)
moduleName = strings.TrimSpace(mod.Name)
}
}
communityLabels := map[string]string{}
if communities, listErr := s.store.ListCommunities(a.TenantID); listErr == nil {
for _, c := range communities {
if c == nil {
continue
}
label := strings.TrimSpace(c.Title)
if label == "" {
label = strings.TrimSpace(c.Community)
}
if label == "" {
label = c.ID
}
communityLabels[c.ID] = label
}
}
type rawRow struct {
Prefix string
Source string
Kind string
SourceName string
SourceDetail string
CommunityID string
CommunityLabel string
}
type summary struct {
Kind string
Source string
SourceDetail string
CommunityID string
CommunityLabel string
Count int
Sample []string
}
cdnSourceURLByID := map[string]string{}
if moduleType == "CDN_CIDRS" && strings.TrimSpace(rev.ModuleID) != "" {
if sources, srcErr := s.store.ListCDNSources(a.TenantID, rev.ModuleID); srcErr == nil {
for _, src := range sources {
if src == nil {
continue
}
id := strings.TrimSpace(src.ID)
url := strings.TrimSpace(src.URL)
if id != "" && url != "" {
cdnSourceURLByID[id] = url
}
}
}
}
rows := make([]rawRow, 0, rev.MaterializedPrefixCount)
byGroup := map[string]*summary{}
cursor := ""
for {
page, next, more := s.store.ListRevisionPrefixes(a.TenantID, revID, cursor, 2000)
for _, p := range page {
kind, sourceName := classifyRevisionSource(p.Source)
sourceDetail := sourceName
if kind == "cdn" {
if url, ok := cdnSourceURLByID[sourceName]; ok && strings.TrimSpace(url) != "" {
sourceDetail = url
}
}
communityID := "none"
if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" {
communityID = strings.TrimSpace(*p.CommunityID)
}
communityLabel := "без community"
if communityID != "none" {
if lbl, ok := communityLabels[communityID]; ok && strings.TrimSpace(lbl) != "" {
communityLabel = lbl
} else {
communityLabel = communityID
}
}
rows = append(rows, rawRow{
Prefix: p.Prefix,
Source: p.Source,
Kind: kind,
SourceName: sourceName,
SourceDetail: sourceDetail,
CommunityID: communityID,
CommunityLabel: communityLabel,
})
groupKey := kind + "|" + sourceName + "|" + communityID
g, ok := byGroup[groupKey]
if !ok {
g = &summary{
Kind: kind,
Source: sourceName,
SourceDetail: sourceDetail,
CommunityID: communityID,
CommunityLabel: communityLabel,
Sample: make([]string, 0, 5),
}
byGroup[groupKey] = g
}
g.Count++
if len(g.Sample) < 5 {
g.Sample = append(g.Sample, p.Prefix)
}
}
if !more || strings.TrimSpace(next) == "" {
break
}
cursor = next
}
summaryKeys := make([]string, 0, len(byGroup))
for k := range byGroup {
summaryKeys = append(summaryKeys, k)
}
sort.Strings(summaryKeys)
sort.Slice(rows, func(i, j int) bool {
if rows[i].Kind != rows[j].Kind {
return rows[i].Kind < rows[j].Kind
}
if rows[i].SourceName != rows[j].SourceName {
return rows[i].SourceName < rows[j].SourceName
}
if rows[i].CommunityID != rows[j].CommunityID {
return rows[i].CommunityID < rows[j].CommunityID
}
return rows[i].Prefix < rows[j].Prefix
})
var b strings.Builder
b.WriteString("EvoBGP revision diagnostic log\n")
b.WriteString("generated_at=" + time.Now().UTC().Format(time.RFC3339Nano) + "\n")
b.WriteString("tenant_id=" + a.TenantID + "\n")
b.WriteString("revision_id=" + rev.ID + "\n")
b.WriteString("revision_created_at=" + rev.CreatedAt.UTC().Format(time.RFC3339Nano) + "\n")
b.WriteString("content_hash=" + rev.ContentHash + "\n")
b.WriteString("materialized_prefix_count=" + strconv.Itoa(rev.MaterializedPrefixCount) + "\n")
if rev.ModuleID != "" {
b.WriteString("module_id=" + rev.ModuleID + "\n")
} else {
b.WriteString("module_id=\n")
}
b.WriteString("module_type=" + moduleType + "\n")
b.WriteString("module_name=" + moduleName + "\n")
b.WriteString("fetched_rows=" + strconv.Itoa(len(rows)) + "\n\n")
b.WriteString("## Aggregation by source and community\n")
for _, k := range summaryKeys {
g := byGroup[k]
b.WriteString("- kind=" + g.Kind +
" source=" + g.Source +
" source_detail=" + g.SourceDetail +
" community_id=" + g.CommunityID +
" community_label=" + g.CommunityLabel +
" count=" + strconv.Itoa(g.Count))
if len(g.Sample) > 0 {
b.WriteString(" sample=" + strings.Join(g.Sample, ","))
}
b.WriteByte('\n')
}
b.WriteString("\n## Raw rows\n")
b.WriteString("prefix\tkind\tsource\tsource_detail\tcommunity_id\tcommunity_label\n")
for _, row := range rows {
b.WriteString(row.Prefix)
b.WriteByte('\t')
b.WriteString(row.Kind)
b.WriteByte('\t')
b.WriteString(row.Source)
b.WriteByte('\t')
b.WriteString(strings.ReplaceAll(row.SourceDetail, "\t", " "))
b.WriteByte('\t')
b.WriteString(row.CommunityID)
b.WriteByte('\t')
b.WriteString(strings.ReplaceAll(row.CommunityLabel, "\t", " "))
b.WriteByte('\n')
}
filename := "revision-" + shortRevisionID(rev.ID) + "-diagnostic.log"
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.Header().Set("Content-Disposition", `attachment; filename="`+filename+`"`)
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(b.String()))
}
func classifyRevisionSource(src string) (kind, sourceName 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 strings.TrimSpace(src) == "" {
return "unknown", "unknown"
}
return "source", strings.TrimSpace(src)
}
}
func shortRevisionID(id string) string {
s := strings.TrimSpace(id)
if len(s) <= 8 {
return s
}
return s[:8]
}
func (s *Server) handleRevisionDiff(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
d, err := s.store.RevisionDiff(a.TenantID, r.PathValue("revision_a"), r.PathValue("revision_b"))
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
writeJSON(w, http.StatusOK, d)
}
func (s *Server) handleRevisionRollback(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "editor") {
return
}
revID := r.PathValue("revision_id")
if _, err := s.store.GetRevision(a.TenantID, revID); err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
idem := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
var idemPtr *string
if idem != "" {
idemPtr = &idem
}
j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindRevisionRollback, idemPtr, nil, map[string]any{
"source_revision_id": revID,
})
if err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
w.Header().Set("Location", "/v1/jobs/"+j.ID)
snap := j.Snapshot()
writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]})
}
func (s *Server) handleApply(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if strings.ToLower(a.Role) != "operator" {
writeProblem(w, http.StatusForbidden, "Forbidden", "operator role required")
return
}
var body struct {
RevisionID string `json:"revision_id"`
Strategy string `json:"strategy"`
DryRun bool `json:"dry_run"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
revID := strings.TrimSpace(body.RevisionID)
if revID == "" {
writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "revision_id required")
return
}
if _, err := s.store.GetRevision(a.TenantID, revID); err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
if body.DryRun {
writeJSON(w, http.StatusOK, map[string]any{"dry_run": true, "revision_id": revID})
return
}
idem := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
var idemPtr *string
if idem != "" {
idemPtr = &idem
}
j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindDeployApply, idemPtr, nil, map[string]any{
"revision_id": revID,
"strategy": body.Strategy,
})
if err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
w.Header().Set("Location", "/v1/jobs/"+j.ID)
snap := j.Snapshot()
writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]})
}
func (s *Server) handleSpeakerApply(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if strings.ToLower(a.Role) != "operator" {
writeProblem(w, http.StatusForbidden, "Forbidden", "operator role required")
return
}
spkID := r.PathValue("id")
if _, err := s.store.GetSpeaker(a.TenantID, spkID); err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "speaker not found")
return
}
var body struct {
RevisionID string `json:"revision_id"`
DryRun bool `json:"dry_run"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
revID := strings.TrimSpace(body.RevisionID)
if revID == "" {
writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "revision_id required")
return
}
if _, err := s.store.GetRevision(a.TenantID, revID); err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
if body.DryRun {
writeJSON(w, http.StatusOK, map[string]any{"dry_run": true})
return
}
idem := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
var idemPtr *string
if idem != "" {
idemPtr = &idem
}
j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindDeployApply, idemPtr, nil, map[string]any{
"revision_id": revID,
"speaker_id": spkID,
})
if err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
w.Header().Set("Location", "/v1/jobs/"+j.ID)
snap := j.Snapshot()
writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]})
}
func (s *Server) handleBirdReload(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if strings.ToLower(a.Role) != "operator" {
writeProblem(w, http.StatusForbidden, "Forbidden", "operator role required")
return
}
idem := strings.TrimSpace(r.Header.Get("Idempotency-Key"))
var idemPtr *string
if idem != "" {
idemPtr = &idem
}
j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindBirdReload, idemPtr, nil, nil)
if err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
w.Header().Set("Location", "/v1/jobs/"+j.ID)
writeJSON(w, http.StatusAccepted, map[string]any{"job_id": j.ID, "status": "queued"})
}
func (s *Server) handleBirdStatus(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
ctx, cancel := context.WithTimeout(r.Context(), 12*time.Second)
defer cancel()
st := birdfmt.InspectLocalBird(ctx)
writeJSON(w, http.StatusOK, st)
}
func (s *Server) handleListJobs(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
status := r.URL.Query().Get("status")
kind := r.URL.Query().Get("kind")
cursor := r.URL.Query().Get("cursor")
list, next, more := s.jobs.List(a.TenantID, status, kind, cursor, limit)
items := make([]map[string]any, 0, len(list))
for _, j := range list {
items = append(items, j.Snapshot())
}
writeJSON(w, http.StatusOK, map[string]any{
"items": items, "next_cursor": strPtrOrNull(next), "has_more": more,
})
}
func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "viewer") {
return
}
j, err := s.jobs.Get(a.TenantID, r.PathValue("job_id"))
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "job not found")
return
}
writeJSON(w, http.StatusOK, j.Snapshot())
}
func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireAtLeast(w, a, "editor") {
return
}
j, err := s.jobs.RequestCancel(a.TenantID, r.PathValue("job_id"))
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "job not found")
return
}
writeJSON(w, http.StatusAccepted, j.Snapshot())
}
func (s *Server) handleNodeLatestRevision(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireNode(w, a) {
return
}
sid := r.PathValue("speaker_id")
sp, err := s.store.GetSpeakerAnyTenant(sid)
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "speaker not found")
return
}
if sp.TenantID != a.TenantID {
writeProblem(w, http.StatusForbidden, "Forbidden", "speaker not in tenant scope")
return
}
rid, at, err := s.store.LatestPublishedRevision(sid)
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "no published revision")
return
}
writeJSON(w, http.StatusOK, map[string]any{
"revision_id": rid, "published_at": at.UTC().Format(time.RFC3339Nano),
})
}
func (s *Server) handleNodeBundle(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireNode(w, a) {
return
}
sid := r.PathValue("speaker_id")
rid := r.PathValue("revision_id")
sp, err := s.store.GetSpeakerAnyTenant(sid)
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "speaker not found")
return
}
if sp.TenantID != a.TenantID {
writeProblem(w, http.StatusForbidden, "Forbidden", "speaker not in tenant scope")
return
}
rev, err := s.store.GetRevision(a.TenantID, rid)
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
return
}
tgz, err := bundle.BuildGzippedTar(rid, sid, rev.PreviewFragments, s.bundlePriv)
if err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
w.Header().Set("Content-Type", "application/gzip")
w.Header().Set("Content-Disposition", `attachment; filename="bundle.tar.gz"`)
w.WriteHeader(http.StatusOK)
_, _ = w.Write(tgz)
}
func (s *Server) handleNodeEnroll(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requireNode(w, a) {
return
}
var body struct {
SpeakerID string `json:"speaker_id"`
PublicKey string `json:"public_key"`
}
dec := json.NewDecoder(r.Body)
if err := dec.Decode(&body); err != nil && !errors.Is(err, io.EOF) {
writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid JSON body")
return
}
sid := strings.TrimSpace(body.SpeakerID)
if sid == "" {
writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "speaker_id is required")
return
}
sp, err := s.store.GetSpeaker(a.TenantID, sid)
if err != nil {
writeProblem(w, http.StatusNotFound, "Not Found", "speaker not found")
return
}
updates := map[string]any{
"node_enrolled_at": time.Now().UTC().Format(time.RFC3339Nano),
}
if pk := strings.TrimSpace(body.PublicKey); pk != "" {
updates["node_public_key"] = pk
}
meta, err := mergeSpeakerMetaJSON(sp.MetaJSON, updates)
if err != nil {
writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "speaker meta_json must be a JSON object (or empty)")
return
}
patch := &store.SpeakerPatch{MetaJSON: &meta}
if _, err := s.store.UpdateSpeaker(a.TenantID, sid, patch); err != nil {
writeProblem(w, http.StatusInternalServerError, "Internal Error", err.Error())
return
}
writeJSON(w, http.StatusOK, map[string]any{
"status": "enrolled",
"speaker_id": sp.ID,
"tenant_id": a.TenantID,
})
}
func mergeSpeakerMetaJSON(existing string, updates map[string]any) (string, error) {
existing = strings.TrimSpace(existing)
var m map[string]any
if existing != "" {
if err := json.Unmarshal([]byte(existing), &m); err != nil {
return "", err
}
if m == nil {
return "", errors.New("meta must be a JSON object")
}
}
if m == nil {
m = make(map[string]any)
}
for k, v := range updates {
m[k] = v
}
b, err := json.Marshal(m)
if err != nil {
return "", err
}
return string(b), nil
}
func sortedFragmentKeys(m map[string]string) []string {
keys := make([]string, 0, len(m))
for k := range m {
keys = append(keys, k)
}
sort.Strings(keys)
return keys
}