sub-store/internal/handler/egress_background.go

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)
	}
}