package handler import ( "context" "encoding/json" "log/slog" "time" "github.com/peterqiu0516/sub-store/internal/model" "github.com/peterqiu0516/sub-store/internal/service" ) // probeNodeEgressFn is the probe seam: a package-level variable so tests can // stub out the mihomo-based egress probe. Default implementation probes via // a local mihomo instance. var probeNodeEgressFn = (*Deps).probeSingleNodeEgress // probeSourceEgressBackground fetches the source's nodes and probes each // node's egress info asynchronously. Results are written to the cache so // that subsequent preview/collection requests can read them without // triggering DNS or GeoIP lookups. func (d *Deps) probeSourceEgressBackground(rec model.SourceRecord) { defer func() { if r := recover(); r != nil { slog.Warn("background egress probe panicked", "source", rec.ID, "error", r) } }() settings, _ := d.SettingsRepo.Get() result, err := service.BuildSubscriptionResult(context.Background(), service.BuildOptions{ Source: &rec, Sources: []model.SourceRecord{rec}, Target: "json", Settings: settings, CacheRepo: d.CacheRepo, ProxyURL: d.Cfg.Fetcher.ProxyURL, }) if err != nil { slog.Warn("background egress probe: failed to build source", "source", rec.ID, "error", err) return } var payload struct { Proxies []model.ProxyNode `json:"proxies"` } if err := json.Unmarshal([]byte(result.Body), &payload); err != nil { slog.Warn("background egress probe: failed to parse nodes", "source", rec.ID, "error", err) return } nodes := payload.Proxies slog.Info("background egress probe started", "source", rec.ID, "nodes", len(nodes)) probed := 0 for i, node := range nodes { if node == nil { continue } // Skip if already cached (fresh entries only — SafeGet treats expired // entries as misses, and error results are cached with the full TTL too) cacheKey := service.EgressCacheKey(node) if _, ok := d.CacheRepo.SafeGet(cacheKey); ok { probed++ continue } // Probe this node info, err := probeNodeEgressFn(d, node) if err != nil { slog.Debug("background egress probe: node failed", "source", rec.ID, "node", node["name"], "error", err) continue } probed++ // Cache the result if data, err := json.Marshal(info); err == nil { ttl := int(d.Cfg.Fetcher.EgressCacheTTL.Seconds()) if ttl <= 0 { ttl = 86400 } d.CacheRepo.SafePut(cacheKey, string(data), nil, ttl) } // Log progress every 10 nodes if (i+1)%10 == 0 { slog.Info("background egress probe progress", "source", rec.ID, "done", i+1, "total", len(nodes)) } } slog.Info("background egress probe completed", "source", rec.ID, "probed", probed, "total", len(nodes)) } // probeSingleNodeEgress probes a single node's egress info without going // through the HTTP handler. It reuses the same probe logic as // HandleEgressInfo. func (d *Deps) probeSingleNodeEgress(node model.ProxyNode) (map[string]any, error) { if v := getStringValue(node["server"]); v == "" { node["server"] = getStringValue(node["address"]) } if v := getStringValue(node["name"]); v == "" { node["name"] = "PROXY" } latencyMs, latencyErr := probeServerPortLatency(node, 5*time.Second) port, err := freeLocalPort() if err != nil { return nil, err } configData, err := buildEgressProbeConfig(node, port) if err != nil { return nil, err } info, err := runEgressProbe(configData, port) if err != nil { info = map[string]any{"egressError": err.Error()} } if latencyMs >= 0 { info["latencyMs"] = latencyMs } if latencyErr != "" { info["latencyError"] = latencyErr } return info, nil } // StartEgressRefresher runs a background goroutine that periodically // re-probes egress info for all enabled sources' nodes whose cache entries // are missing or expired (HH-773). The first pass runs immediately; fresh // cache hits — including error results within TTL — are skipped by // probeSourceEgressBackground. Cancel the context to stop. func StartEgressRefresher(ctx context.Context, d *Deps, interval time.Duration) { go func() { d.refreshAllSourceEgress(ctx) ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ticker.C: d.refreshAllSourceEgress(ctx) case <-ctx.Done(): return } } }() } // refreshAllSourceEgress rebuilds the node list for every enabled source and // re-probes nodes with missing/expired egress cache entries. func (d *Deps) refreshAllSourceEgress(ctx context.Context) { defer func() { if r := recover(); r != nil { slog.Warn("egress refresher panicked", "error", r) } }() sources, err := d.SourceRepo.List() if err != nil { slog.Warn("egress refresher: failed to list sources", "error", err) return } for _, rec := range sources { if !rec.Enabled { continue } select { case <-ctx.Done(): return default: } d.probeSourceEgressBackground(rec) } }