package nodedispatch import ( "bytes" "context" "crypto/tls" "encoding/json" "fmt" "io" "net/http" "os" "strings" "time" "evobgp/internal/store" ) // Result is one speaker dispatch outcome for job meta. type Result struct { SpeakerID string `json:"speaker_id"` Endpoint string `json:"endpoint,omitempty"` Status string `json:"status"` AppliedRevisionID string `json:"applied_revision_id,omitempty"` Error string `json:"error,omitempty"` } // Options configures Panel→Node HTTP dispatch. type Options struct { HTTPClient *http.Client Timeout time.Duration MaxRetries int InsecureTLS bool RevisionID string } func (o Options) client() *http.Client { if o.HTTPClient != nil { return o.HTTPClient } timeout := o.Timeout if timeout <= 0 { timeout = 30 * time.Second } tr := http.DefaultTransport.(*http.Transport).Clone() if o.InsecureTLS || strings.TrimSpace(os.Getenv("EVOBGP_NODE_DISPATCH_INSECURE_TLS")) == "1" { tr.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec // dev/lab only via env } return &http.Client{Timeout: timeout, Transport: tr} } func (o Options) retries() int { if o.MaxRetries > 0 { return o.MaxRetries } return 3 } // Enabled reports whether remote dispatch is turned on (EVOBGP_NODE_DISPATCH_ENABLED=1). func Enabled() bool { return strings.TrimSpace(os.Getenv("EVOBGP_NODE_DISPATCH_ENABLED")) == "1" } // WakeSpeaker POSTs /v1/agent/sync to a replica agent (HTTPS via Traefik). func WakeSpeaker(ctx context.Context, sp *store.Speaker, opts Options) Result { res := Result{SpeakerID: sp.ID} if sp == nil { res.Status = "error" res.Error = "nil speaker" return res } meta := store.ParseSpeakerMeta(sp.MetaJSON) url := store.AgentSyncURL(meta) if url == "" { res.Status = "skipped" res.Error = "agent_domain or agent_secret not configured" return res } res.Endpoint = url secret := strings.TrimSpace(meta.AgentSecret) if secret == "" { res.Status = "skipped" res.Error = "agent_secret missing" return res } body := map[string]string{} if rid := strings.TrimSpace(opts.RevisionID); rid != "" { body["revision_id"] = rid } raw, _ := json.Marshal(body) var lastErr error client := opts.client() for attempt := 0; attempt < opts.retries(); attempt++ { if attempt > 0 { select { case <-ctx.Done(): res.Status = "error" res.Error = ctx.Err().Error() return res case <-time.After(time.Duration(attempt) * 2 * time.Second): } } req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(raw)) if err != nil { lastErr = err continue } req.Header.Set("Authorization", "Bearer "+secret) req.Header.Set("Content-Type", "application/json") resp, err := client.Do(req) if err != nil { lastErr = err continue } b, _ := io.ReadAll(resp.Body) _ = resp.Body.Close() if resp.StatusCode >= 200 && resp.StatusCode < 300 { var out struct { AppliedRevisionID string `json:"applied_revision_id"` } _ = json.Unmarshal(b, &out) res.Status = "ok" res.AppliedRevisionID = strings.TrimSpace(out.AppliedRevisionID) if res.AppliedRevisionID == "" { res.AppliedRevisionID = strings.TrimSpace(opts.RevisionID) } return res } lastErr = fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) } res.Status = "error" if lastErr != nil { res.Error = lastErr.Error() } return res } // WakeReplicas dispatches sync to all tenant speakers that need remote wake-up. func WakeReplicas(ctx context.Context, st store.Backend, tenantID, revisionID string, opts Options) []Result { if st == nil { return nil } opts.RevisionID = revisionID var out []Result for _, sp := range st.ListSpeakersForTenant(tenantID) { if sp == nil { continue } meta := store.ParseSpeakerMeta(sp.MetaJSON) if !store.SpeakerNeedsRemoteDispatch(sp.Role, meta) { continue } out = append(out, WakeSpeaker(ctx, sp, opts)) } return out } // CheckHealth GETs /v1/agent/health for UI Connected/Offline status. func CheckHealth(ctx context.Context, sp *store.Speaker, opts Options) (ok bool, detail string) { if sp == nil { return false, "nil speaker" } meta := store.ParseSpeakerMeta(sp.MetaJSON) url := store.AgentHealthURL(meta) if url == "" { return false, "agent_domain not configured" } secret := strings.TrimSpace(meta.AgentSecret) if secret == "" { return false, "agent_secret missing" } req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return false, err.Error() } req.Header.Set("Authorization", "Bearer "+secret) resp, err := opts.client().Do(req) if err != nil { return false, err.Error() } defer func() { _ = resp.Body.Close() }() if resp.StatusCode >= 200 && resp.StatusCode < 300 { return true, "connected" } b, _ := io.ReadAll(resp.Body) return false, fmt.Sprintf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) }