Files
EvoBGP/internal/httpapi/routes.go
T
Denozordec 927e27640a
quality / commitlint (push) Skipped
quality / changes (push) Successful in 8s
quality / docker-check (push) Skipped
quality / openapi (push) Successful in 26s
quality / web (push) Successful in 1m27s
quality / go (push) Successful in 1m18s
quality / bird2 (push) Successful in 16s
CD / quality (push) Successful in 3m43s
CD / publish (push) Successful in 3m11s
feat(docs): update speaker installation instructions and logging details
- Enhanced the speaker installation documentation to clarify the use of TCP port 179 and the logging commands for monitoring BIRD and evobgp-agent.
- Updated the speaker form dialog to include additional information about MikroTik connections and logging commands.
- Modified the BIRD configuration to include logging to stderr for better visibility during operations.
- Adjusted the Docker Compose configuration to ensure proper network settings and sysctl configurations for BGP functionality.
2026-08-21 16:11:28 +07:00

1102 lines
33 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/reports"
"evobgp/internal/store"
"evobgp/internal/version"
)
// 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.ProtectMetrics(observability.MetricsHandler()))
s.mux.HandleFunc("GET /version", s.handleVersion)
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.HandleFunc("GET /v1/auth/config", s.withAuthRateLimit(s.handleAuthConfigPublic))
// Firewall subsystem moved to the standalone EvoFirewall service; see docs/firewall.md.
// Registered on the public mux so it wins over the "/v1/" subtree below regardless of auth.
s.mux.HandleFunc("/v1/firewall/", s.handleFirewallGone)
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 /lookup", s.handleLookup)
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("GET /bundle/signing-public-key", s.handleBundleSigningPublicKey)
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/prune-estimate", s.handleRevisionPruneEstimate)
m.HandleFunc("POST /revisions/prune", s.handleRevisionPrune)
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("GET /jobs/{job_id}/report", s.handleGetJobReport)
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)
s.registerAuditRoutes(m)
s.registerPostgresMonitoringRoutes(m)
s.registerPostgresMaintenanceRoutes(m)
s.registerMaintenanceRoutes(m)
s.registerRuntimeLogsRoutes(m)
}
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}
// handleAuthConfigPublic exposes portal-auth wiring so the UI can decide whether to redirect to the login portal.
// Registered on the public mux (no auth middleware): safe to call without a bearer token.
func (s *Server) handleAuthConfigPublic(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{
"required": s.authRequired,
"portal_url": s.authPortalURL,
})
}
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 err := s.store.Ping(ctx); err != nil {
checks["store"] = "unavailable"
if s.pgPool != nil {
checks["postgres"] = "unavailable"
}
writeJSON(w, http.StatusServiceUnavailable, map[string]any{"status": "not_ready", "checks": checks})
return
}
if s.pgPool != nil {
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) {
ver, sha, buildTime := version.Info()
resp := map[string]string{
"version": ver,
"api_version": ver,
"git_sha": sha,
}
if buildTime != "" {
resp["build_time"] = buildTime
}
writeJSON(w, http.StatusOK, resp)
}
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
}
ids := mod.EffectiveDohProfileIDs()
if len(ids) > 0 {
m["doh_profile_ids"] = ids
} else {
m["doh_profile_ids"] = []string{}
}
m["doh_resolver_policy"] = store.NormalizeDohResolverPolicy(mod.DohResolverPolicy)
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 {
return speakerJSONFromStore(nil, sp)
}
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.requirePerm(w, a, "bgp:modules:read") {
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
}
filtered := make([]*store.Module, 0)
all := s.store.ListModules(a.TenantID)
all = store.FilterOwned(all, func(m *store.Module) string { return m.CreatedByUserID }, a.Kind, a.IsAdmin, a.UserID)
for _, mod := range all {
if typeFilter != "" && mod.Type != typeFilter {
continue
}
if enabledFilter != nil && mod.Enabled != *enabledFilter {
continue
}
filtered = append(filtered, mod)
}
page, next, more := store.PaginateOffset(filtered, r.URL.Query().Get("cursor"), parseListLimit(r))
items := make([]map[string]any, 0, len(page))
for _, mod := range page {
items = append(items, moduleJSON(mod))
}
writeJSON(w, http.StatusOK, map[string]any{
"items": items, "next_cursor": strPtrOrNull(next), "has_more": more,
})
}
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.requirePerm(w, a, "bgp:directories:read") {
return
}
cat, err := reports.BuildRouterListsCatalog(s.store, a.TenantID)
if err != nil {
writeStoreErr(w, err)
return
}
writeJSON(w, http.StatusOK, map[string]any{
"modules": map[string]any{"items": cat.Modules},
"domains": map[string]any{"items": cat.Domains},
"asns": map[string]any{"items": cat.ASNs},
"ip_ranges": map[string]any{"items": cat.IPRanges},
"communities": map[string]any{"items": cat.Communities},
})
}
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.requirePerm(w, a, "bgp:modules:read") {
return
}
mod, err := s.store.GetModule(a.TenantID, r.PathValue("module_id"))
if err == nil && !store.CanAccessOwned(a.Kind, a.IsAdmin, a.UserID, mod.CreatedByUserID) {
writeProblem(w, http.StatusNotFound, "Not Found", "module not found")
return
}
if err != nil {
if err == store.ErrNotFound || err == store.ErrTenantScope {
writeProblem(w, http.StatusNotFound, "Not Found", "module not found")
return
}
writeInternalError(w, "internal", err)
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.requirePerm(w, a, "bgp:network:read") {
return
}
allPeers := s.store.ListPeers(a.TenantID)
allPeers = store.FilterOwned(allPeers, func(p *store.BGPPeer) string { return p.CreatedByUserID }, a.Kind, a.IsAdmin, a.UserID)
page, next, more := store.PaginateOffset(allPeers, r.URL.Query().Get("cursor"), parseListLimit(r))
fresh := r != nil && strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("live")), "1")
ctx, cancel := context.WithTimeout(r.Context(), 12*time.Second)
defer cancel()
liveViews := s.collectSpeakerBGPLive(ctx, a.TenantID, fresh)
if fresh {
s.syncPeerDiscoveriesFromLive(a.TenantID, liveViews)
}
items := make([]map[string]any, 0, len(page))
for _, p := range page {
row := peerJSON(p)
applyPeerLiveFields(row, p, liveViews)
items = append(items, row)
}
writeJSON(w, http.StatusOK, map[string]any{
"items": items,
"next_cursor": strPtrOrNull(next),
"has_more": more,
"live_speaker_poll": liveSpeakerPollJSON(liveViews),
})
}
func (s *Server) liveBGPProtocolStates(r *http.Request) map[string]string {
if r != nil && strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("live")), "1") {
return s.liveBGPProtocolStatesFresh(r.Context())
}
if cached, ok := observability.CachedBirdProtocolStates(90 * time.Second); ok {
return cached
}
return s.liveBGPProtocolStatesFresh(r.Context())
}
func (s *Server) liveBGPProtocolStatesFresh(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{}
}
states := birdfmt.ParseBGPProtocolStates(out)
observability.SetBirdProtocolStates(states)
return states
}
// 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 forwards to birdfmt for tests and legacy callers.
func peerProtocolNameForID(peerID string) string {
return birdfmt.PeerProtocolName(peerID)
}
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.requirePerm(w, a, "bgp:network:read") {
return
}
speakers := s.store.ListSpeakersForTenant(a.TenantID)
fresh := r != nil && strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("live")), "1")
var liveByID map[string]map[string]any
if fresh {
ctx, cancel := context.WithTimeout(r.Context(), 12*time.Second)
defer cancel()
liveByID = s.collectSpeakerLiveStatus(ctx, a.TenantID, true, speakers)
}
items := make([]map[string]any, 0, len(speakers))
for _, sp := range speakers {
row := speakerJSONFromStore(s.store, sp)
if liveByID != nil {
if live, ok := liveByID[sp.ID]; ok {
row["live"] = live
}
}
items = append(items, row)
}
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.requirePerm(w, a, "bgp:modules:write") {
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
}
writeInternalError(w, "internal", err)
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 {
writeInternalError(w, "internal", err)
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.requirePerm(w, a, "bgp:modules:write") {
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 {
writeInternalError(w, "internal", err)
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.requirePerm(w, a, "bgp:operations:read") {
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
}
// 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
key := "module_refresh:" + moduleID
_, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, &key, &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.requirePerm(w, a, "bgp:operations:read") {
return
}
rev, err := s.store.GetRevisionSummary(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.requirePerm(w, a, "bgp:operations:read") {
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
}
if expanded := pipeline.BuildExpandedBirdPreview(rev.PreviewFragments); expanded != "" {
obj[pipeline.AuxBirdFullExpandedKey()] = expanded
}
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.requirePerm(w, a, "bgp:operations:read") {
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
}
logOut, err := reports.BuildRevisionDiagnosticLog(s.store, a.TenantID, rev)
if err != nil {
writeInternalError(w, "revision diagnostic log", err)
return
}
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.Header().Set("Content-Disposition", `attachment; filename="`+logOut.Filename+`"`)
w.WriteHeader(http.StatusOK)
_, _ = w.Write(logOut.Body)
}
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.requirePerm(w, a, "bgp:operations:read") {
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.requirePerm(w, a, "bgp:operations:admin") {
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 {
writeInternalError(w, "internal", err)
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 !s.requirePerm(w, a, "bgp:operations:admin") {
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 {
writeInternalError(w, "internal", err)
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 !s.requirePerm(w, a, "bgp:operations:admin") {
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 {
writeInternalError(w, "internal", err)
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 !s.requirePerm(w, a, "bgp:operations:admin") {
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 {
writeInternalError(w, "internal", err)
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.requirePerm(w, a, "bgp:monitoring:read") {
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.requirePerm(w, a, "bgp:operations:read") {
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.requirePerm(w, a, "bgp:operations:read") {
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) handleGetJobReport(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok {
writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
return
}
if !s.requirePerm(w, a, "bgp:operations:read") {
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
}
snap := j.Snapshot()
meta, _ := snap["meta"].(map[string]any)
out := map[string]any{
"job_id": snap["job_id"],
"kind": snap["kind"],
"status": snap["status"],
"meta": meta,
"error": snap["error"],
"created_at": snap["created_at"],
}
if meta != nil {
if v, ok := meta["log_entries"]; ok {
out["log_entries"] = v
}
if v, ok := meta["log_total"]; ok {
out["log_total"] = v
}
if v, ok := meta["revision_id"]; ok {
out["revision_id"] = v
}
}
writeJSON(w, http.StatusOK, out)
}
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.requirePerm(w, a, "bgp:operations:admin") {
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
}
frags := rev.PreviewFragments
overlaid, err := pipeline.OverlayFragmentsForSpeaker(s.store, a.TenantID, sid, rid, frags)
if err != nil {
writeInternalError(w, "bundle overlay", err)
return
}
frags = overlaid
tgz, err := bundle.BuildGzippedTar(rid, sid, frags, s.bundlePriv)
if err != nil {
writeInternalError(w, "internal", err)
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 {
writeInternalError(w, "internal", err)
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
}