1package daemon23import (4 "context"5 "errors"6 "fmt"7 "io"8 "net/http"9 "net/url"10 "sort"11 "strings"12 "sync"13 "time"14)1516type PeerMonitorConfig struct {17 Peers []string18 Interval time.Duration19 Timeout time.Duration20 Publish bool21 PublishFunc func(context.Context, []PeerStatus) error22 Client *http.Client23 Observe func(outcome string, duration time.Duration)24}2526type PeerStatus struct {27 URL string `json:"url"`28 Up bool `json:"up"`29 Latency time.Duration `json:"latency"`30 LastChecked time.Time `json:"last_checked"`31 LastSuccess time.Time `json:"last_success,omitempty"`32 Error string `json:"error,omitempty"`33}3435type PeerMonitor struct {36 interval time.Duration37 timeout time.Duration38 publish bool39 client *http.Client40 observe func(string, time.Duration)41 publishFn func(context.Context, []PeerStatus) error42 mu sync.RWMutex43 peers map[string]PeerStatus44}4546// SetPublisher installs the operator-selected publication sink. It is kept47// separate from probing so a monitor can be constructed before tenants are48// opened during process startup.49func (m *PeerMonitor) SetPublisher(publish func(context.Context, []PeerStatus) error) {50 if m == nil {51 return52 }53 m.mu.Lock()54 m.publishFn = publish55 m.mu.Unlock()56}5758func NewPeerMonitor(cfg PeerMonitorConfig) (*PeerMonitor, error) {59 if cfg.Interval <= 0 {60 cfg.Interval = time.Minute61 }62 if cfg.Timeout <= 0 {63 cfg.Timeout = 5 * time.Second64 }65 peers := make(map[string]PeerStatus, len(cfg.Peers))66 for _, raw := range cfg.Peers {67 u, err := url.Parse(strings.TrimSpace(raw))68 if err != nil || u.Host == "" || u.User != nil || u.Fragment != "" || (u.Scheme != "http" && u.Scheme != "https") {69 return nil, fmt.Errorf("peer monitor: invalid peer URL %q", raw)70 }71 canonical := strings.TrimRight(u.String(), "/")72 peers[canonical] = PeerStatus{URL: canonical}73 }74 return &PeerMonitor{interval: cfg.Interval, timeout: cfg.Timeout, publish: cfg.Publish, client: cfg.Client, observe: cfg.Observe, publishFn: cfg.PublishFunc, peers: peers}, nil75}7677func (m *PeerMonitor) PublicationEnabled() bool {78 if m == nil {79 return false80 }81 m.mu.RLock()82 defer m.mu.RUnlock()83 return m.publish && m.publishFn != nil84}8586func (m *PeerMonitor) Peers() []string {87 if m == nil {88 return nil89 }90 m.mu.RLock()91 defer m.mu.RUnlock()92 result := make([]string, 0, len(m.peers))93 for peer := range m.peers {94 result = append(result, peer)95 }96 sort.Strings(result)97 return result98}99100func (m *PeerMonitor) Snapshot() []PeerStatus {101 if m == nil {102 return nil103 }104 m.mu.RLock()105 defer m.mu.RUnlock()106 result := make([]PeerStatus, 0, len(m.peers))107 for _, status := range m.peers {108 result = append(result, status)109 }110 sort.Slice(result, func(i, j int) bool { return result[i].URL < result[j].URL })111 return result112}113114func (m *PeerMonitor) Probe(ctx context.Context) error {115 if m == nil {116 return errors.New("peer monitor: nil monitor")117 }118 if ctx == nil {119 ctx = context.Background()120 }121 client := m.client122 if client == nil {123 client = &http.Client{}124 }125 var firstErr error126 for _, peer := range m.Peers() {127 started := time.Now()128 probeCtx, cancel := context.WithTimeout(ctx, m.timeout)129 req, err := http.NewRequestWithContext(probeCtx, http.MethodGet, peer, nil)130 if err == nil {131 req.Header.Set("Accept", "application/nostr+json")132 response, requestErr := client.Do(req)133 if requestErr == nil {134 _, _ = io.CopyN(io.Discard, response.Body, 1<<20)135 _ = response.Body.Close()136 if response.StatusCode < 200 || response.StatusCode >= 400 {137 requestErr = fmt.Errorf("peer returned HTTP %d", response.StatusCode)138 }139 }140 err = requestErr141 }142 cancel()143 latency := time.Since(started)144 status := PeerStatus{URL: peer, Up: err == nil, Latency: latency, LastChecked: time.Now()}145 if err == nil {146 m.mu.RLock()147 status.LastSuccess = m.peers[peer].LastSuccess148 m.mu.RUnlock()149 status.LastSuccess = status.LastChecked150 } else {151 status.Error = err.Error()152 m.mu.RLock()153 status.LastSuccess = m.peers[peer].LastSuccess154 m.mu.RUnlock()155 if firstErr == nil {156 firstErr = err157 }158 }159 m.mu.Lock()160 m.peers[peer] = status161 m.mu.Unlock()162 if m.observe != nil {163 outcome := "ok"164 if err != nil {165 outcome = "error"166 }167 m.observe(outcome, latency)168 }169 }170 m.mu.RLock()171 publish := m.publishFn172 m.mu.RUnlock()173 if m.publish && publish != nil {174 if err := publish(ctx, m.Snapshot()); err != nil {175 return fmt.Errorf("peer monitor: publish status: %w", err)176 }177 }178 return firstErr179}180181func (m *PeerMonitor) Run(ctx context.Context) error {182 if m == nil {183 return errors.New("peer monitor: nil monitor")184 }185 if ctx == nil {186 ctx = context.Background()187 }188 _ = m.Probe(ctx)189 ticker := time.NewTicker(m.interval)190 defer ticker.Stop()191 for {192 select {193 case <-ctx.Done():194 return ctx.Err()195 case <-ticker.C:196 _ = m.Probe(ctx)197 }198 }199}