EvoBGP/internal/httpapi/firewall_replicate.go

package httpapi

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"strings"
	"time"

	"evobgp/internal/nodedispatch"
	"evobgp/internal/store"
)

func (s *Server) replicateFirewallStateToSpeakers(tenantID string) {
	if !nodedispatch.Enabled() {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
	defer cancel()

	clients, err := s.store.ListApprovedFirewallClientsForReplication(tenantID)
	if err != nil {
		log.Printf("httpapi: firewall replicate clients: %v", err)
		return
	}
	rules, err := s.store.ListAllFirewallRulesForReplication(tenantID)
	if err != nil {
		log.Printf("httpapi: firewall replicate rules: %v", err)
		return
	}
	revs, _, _ := s.store.ListRevisions(tenantID, "", "", 1)
	if len(revs) == 0 {
		return
	}
	revID := revs[0].ID
	prefixesByCommunity, _, err := s.loadPrefixesByCommunity(tenantID, revID)
	if err != nil {
		log.Printf("httpapi: firewall replicate prefixes: %v", err)
		return
	}

	payloadRules := make([]map[string]any, 0, len(rules))
	for _, r := range rules {
		payloadRules = append(payloadRules, map[string]any{
			"client_id":    r.ClientID,
			"priority":     r.Priority,
			"action":       r.Action,
			"community_id": r.CommunityID,
		})
	}
	payloadClients := make([]map[string]any, 0, len(clients))
	for _, c := range clients {
		payloadClients = append(payloadClients, map[string]any{
			"token_hash_hex": c.TokenHashHex,
			"client_id":      c.ClientID,
			"name":           c.Name,
		})
	}
	body := map[string]any{
		"tenant_id":             tenantID,
		"revision_id":           revID,
		"generated_at":          time.Now().UTC().Format(time.RFC3339),
		"clients":               payloadClients,
		"rules":                 payloadRules,
		"prefixes_by_community": prefixesByCommunity,
	}

	speakers := s.store.ListSpeakersForTenant(tenantID)
	for _, sp := range speakers {
		meta := store.ParseSpeakerMeta(sp.MetaJSON)
		if !store.SpeakerNeedsRemoteDispatch(sp.Role, meta) || !meta.FirewallFailover {
			continue
		}
		domain := strings.TrimSpace(meta.AgentDomain)
		if domain == "" {
			continue
		}
		url := "https://" + strings.TrimSuffix(domain, "/") + "/v1/agent/firewall-replicate"
		status, errMsg := postFirewallReplicate(ctx, url, meta.AgentSecret, body)
		patch := store.SpeakerMeta{
			LastFirewallReplicateAt:     time.Now().UTC().Format(time.RFC3339Nano),
			LastFirewallReplicateStatus: status,
			LastFirewallReplicateError:  errMsg,
		}
		merged := store.MergeSpeakerMetaJSON(sp.MetaJSON, patch)
		mp := merged
		if _, err := s.store.UpdateSpeaker(tenantID, sp.ID, &store.SpeakerPatch{MetaJSON: &mp}); err != nil {
			log.Printf("httpapi: firewall replicate meta update %s: %v", sp.ID, err)
		}
	}
}

func postFirewallReplicate(ctx context.Context, url, secret string, body map[string]any) (status, errMsg string) {
	b, err := json.Marshal(body)
	if err != nil {
		return "error", err.Error()
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(b))
	if err != nil {
		return "error", err.Error()
	}
	req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(secret))
	req.Header.Set("Content-Type", "application/json")
	client := &http.Client{Timeout: 30 * time.Second}
	resp, err := client.Do(req)
	if err != nil {
		return "error", err.Error()
	}
	defer func() { _ = resp.Body.Close() }()
	_, _ = io.Copy(io.Discard, resp.Body)
	if resp.StatusCode >= 200 && resp.StatusCode < 300 {
		return "ok", ""
	}
	return "error", resp.Status
}