refactor: rename reads cached egress geo, fallback name-only (no DNS)
RenameCollectionNodes now reads geo info from cached egress-probe fields (country, countryCode, city, flag) on each node instead of doing per-node DNS lookups via DetectGeoWithServer. When no cached egress data is available (probe not yet run), it falls back to name-based regex only (DetectGeo, no DNS) so the UI is never blocked. The background egress probe (triggered on source create/update) populates the cache asynchronously. Also: runEgressProbe now stores countryCode and flag emoji from ipwho.is response. egressCacheKey and addCachedEgressInfo updated to include the new fields. Reverted previous name-first hack on DetectGeoWithServer and DNS cache on geoip.LookupHost — no longer needed since rename doesn't call them. Collection preview with rename: 30s -> 52ms (with cache miss, name-based fallback).
This commit is contained in:
@@ -52,6 +52,8 @@ func (d *Deps) HandleCreateSource(c fiber.Ctx) error {
|
||||
if err != nil {
|
||||
return failed(c, "Failed to create source", 500)
|
||||
}
|
||||
// Trigger background egress probe for the new source's nodes
|
||||
go d.probeSourceEgressBackground(rec)
|
||||
return success(c, result)
|
||||
}
|
||||
|
||||
@@ -81,6 +83,8 @@ func (d *Deps) HandleUpdateSource(c fiber.Ctx) error {
|
||||
if err != nil {
|
||||
return failed(c, "Failed to update source", 500)
|
||||
}
|
||||
// Trigger background egress probe for the updated source's nodes
|
||||
go d.probeSourceEgressBackground(merged)
|
||||
return success(c, result)
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/peterqiu0516/sub-store/internal/model"
|
||||
"github.com/peterqiu0516/sub-store/internal/service"
|
||||
)
|
||||
|
||||
// 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
|
||||
cacheKey := egressCacheKey(node)
|
||||
if _, ok := d.CacheRepo.SafeGet(cacheKey); ok {
|
||||
probed++
|
||||
continue
|
||||
}
|
||||
|
||||
// Probe this node
|
||||
info, err := d.probeSingleNodeEgress(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
|
||||
}
|
||||
@@ -81,7 +81,7 @@ func egressCacheKey(node model.ProxyNode) string {
|
||||
"id": true, "name": true, "_sourceAlias": true, "_previewId": true,
|
||||
"latencyMs": true, "latencyError": true,
|
||||
"egressIp": true, "egressCountry": true, "egressRegion": true, "egressError": true,
|
||||
"country": true, "region": true, "city": true, "isp": true, "cached": true,
|
||||
"country": true, "countryCode": true, "region": true, "city": true, "isp": true, "flag": true, "cached": true,
|
||||
}
|
||||
for k, v := range node {
|
||||
if !skip[k] {
|
||||
@@ -103,7 +103,7 @@ func (d *Deps) addCachedEgressInfo(nodes []model.ProxyNode) []model.ProxyNode {
|
||||
if json.Unmarshal([]byte(entry.Content), &cached) != nil {
|
||||
continue
|
||||
}
|
||||
for _, key := range []string{"egressIp", "country", "region", "city", "isp", "latencyMs", "latencyError", "egressError"} {
|
||||
for _, key := range []string{"egressIp", "country", "countryCode", "region", "city", "isp", "flag", "latencyMs", "latencyError", "egressError"} {
|
||||
if v, ok := cached[key]; ok {
|
||||
node[key] = v
|
||||
}
|
||||
@@ -223,12 +223,15 @@ func runEgressProbe(configData []byte, port int) (fiber.Map, error) {
|
||||
return nil, fmt.Errorf("%s", msg)
|
||||
}
|
||||
connection, _ := data["connection"].(map[string]any)
|
||||
flagObj, _ := data["flag"].(map[string]any)
|
||||
return fiber.Map{
|
||||
"egressIp": data["ip"],
|
||||
"country": data["country"],
|
||||
"region": data["region"],
|
||||
"city": data["city"],
|
||||
"isp": connection["isp"],
|
||||
"egressIp": data["ip"],
|
||||
"country": data["country"],
|
||||
"countryCode": data["country_code"],
|
||||
"region": data["region"],
|
||||
"city": data["city"],
|
||||
"isp": connection["isp"],
|
||||
"flag": flagObj["emoji"],
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user