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 }