diff --git a/cmd/server.go b/cmd/server.go index 51e1d1f..2722d9c 100644 --- a/cmd/server.go +++ b/cmd/server.go @@ -122,7 +122,13 @@ func startFiber(cfg *config.Config, db *sqlx.DB) error { app.Use(middleware.DownloadHostIsolation(cfg.Auth.DownloadHosts)) // Register routes - handler.RegisterRoutes(app, cfg, db) + deps := handler.RegisterRoutes(app, cfg, db) + + // Start egress geo refresher — HH-773: hourly low-frequency background + // re-probe of nodes with missing/expired egress cache entries. + egressCtx, egressCancel := context.WithCancel(context.Background()) + defer egressCancel() + handler.StartEgressRefresher(egressCtx, deps, time.Hour) // Health check app.Get("/health", func(c fiber.Ctx) error { @@ -154,6 +160,7 @@ func startFiber(cfg *config.Config, db *sqlx.DB) error { logrus.Info("Shutting down server...") cacheCancel() + egressCancel() shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) defer shutdownCancel() diff --git a/internal/filter/collection_rename.go b/internal/filter/collection_rename.go index 081e0bf..ba81f8d 100644 --- a/internal/filter/collection_rename.go +++ b/internal/filter/collection_rename.go @@ -66,7 +66,8 @@ func geoFromNode(proxy model.ProxyNode) util.GeoInfo { // suffix) and numbered sequentially within each group, starting at 1. // // Nodes whose geographic info cannot be determined (no cached data and name -// regex fails) are dropped from the output. +// regex fails) are kept in the output under their original name (alias +// prefixed), so they still participate in numbering and sorting. // // opts controls which fields are included in the output. A nil opts means all // fields are included (full default format). @@ -99,7 +100,19 @@ func RenameCollectionNodes(proxies []model.ProxyNode, opts *model.RenameOptions) geo = util.DetectGeo(ToString(proxy["name"])) } if geo.CountryName == "" { - // Geo detection failed — skip this node + // Geo detection failed — keep the node under its original name + // (alias-prefixed) so it still participates in numbering and sorting. + var parts []string + if o.Alias && alias != "" { + parts = append(parts, "["+alias+"]") + } + if name := ToString(proxy["name"]); name != "" { + parts = append(parts, name) + } + infos = append(infos, nodeInfo{ + proxy: proxy, + baseName: strings.Join(parts, " "), + }) continue } diff --git a/internal/filter/collection_rename_test.go b/internal/filter/collection_rename_test.go index 66ea250..e500e5f 100644 --- a/internal/filter/collection_rename_test.go +++ b/internal/filter/collection_rename_test.go @@ -132,10 +132,21 @@ func TestRenameCollectionNodes_NoAlias(t *testing.T) { func TestRenameCollectionNodes_UnknownCountry(t *testing.T) { proxies := []model.ProxyNode{ {"name": "Unknown Node", "type": "ss", "_sourceAlias": "A"}, + {"name": "Bare Node", "type": "ss"}, } result := RenameCollectionNodes(proxies, nil) - if len(result) != 0 { - t.Fatalf("expected 0 nodes (no cached geo info, should be filtered), got %d", len(result)) + if len(result) != 2 { + t.Fatalf("expected 2 nodes (geo-unknown nodes must be kept), got %d", len(result)) + } + // Unknown geo: keep the original name, alias-prefixed when present + if name, _ := result[0]["name"].(string); name != "[A] Unknown Node" { + t.Errorf("node[0] name = %q, want %q", name, "[A] Unknown Node") + } + if name, _ := result[1]["name"].(string); name != "Bare Node" { + t.Errorf("node[1] name = %q, want %q", name, "Bare Node") + } + if _, ok := result[0]["_sourceAlias"]; ok { + t.Error("_sourceAlias should be stripped from rename output") } } diff --git a/internal/handler/egress_background.go b/internal/handler/egress_background.go index ed442d0..12f77b9 100644 --- a/internal/handler/egress_background.go +++ b/internal/handler/egress_background.go @@ -10,6 +10,11 @@ import ( "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 @@ -24,12 +29,12 @@ func (d *Deps) probeSourceEgressBackground(rec model.SourceRecord) { settings, _ := d.SettingsRepo.Get() result, err := service.BuildSubscriptionResult(context.Background(), service.BuildOptions{ - Source: &rec, - Sources: []model.SourceRecord{rec}, - Target: "json", - Settings: settings, + Source: &rec, + Sources: []model.SourceRecord{rec}, + Target: "json", + Settings: settings, CacheRepo: d.CacheRepo, - ProxyURL: d.Cfg.Fetcher.ProxyURL, + ProxyURL: d.Cfg.Fetcher.ProxyURL, }) if err != nil { slog.Warn("background egress probe: failed to build source", "source", rec.ID, "error", err) @@ -52,15 +57,16 @@ func (d *Deps) probeSourceEgressBackground(rec model.SourceRecord) { if node == nil { continue } - // Skip if already cached - cacheKey := egressCacheKey(node) + // 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 := d.probeSingleNodeEgress(node) + info, err := probeNodeEgressFn(d, node) if err != nil { slog.Debug("background egress probe: node failed", "source", rec.ID, "node", node["name"], "error", err) @@ -119,3 +125,50 @@ func (d *Deps) probeSingleNodeEgress(node model.ProxyNode) (map[string]any, erro } 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) + } +} diff --git a/internal/handler/egress_background_test.go b/internal/handler/egress_background_test.go new file mode 100644 index 0000000..df6eed4 --- /dev/null +++ b/internal/handler/egress_background_test.go @@ -0,0 +1,148 @@ +package handler + +import ( + "context" + "testing" + "time" + + "github.com/peterqiu0516/sub-store/internal/model" + "github.com/peterqiu0516/sub-store/internal/proxy" + "github.com/peterqiu0516/sub-store/internal/service" +) + +// egressKeyForNode parses a single proxy URI and returns its egress cache key +// (via the service-layer key shared with the pipeline). +func egressKeyForNode(t *testing.T, uri string) string { + t.Helper() + nodes := proxy.ParseProxies(uri) + if len(nodes) != 1 { + t.Fatalf("parse %q: got %d nodes", uri, len(nodes)) + } + return service.EgressCacheKey(nodes[0]) +} + +// stubProbeNodeEgress replaces the mihomo-based probe seam for the duration +// of the test and records every probed node name. +func stubProbeNodeEgress(t *testing.T) *[]string { + t.Helper() + var probed []string + old := probeNodeEgressFn + probeNodeEgressFn = func(d *Deps, node model.ProxyNode) (map[string]any, error) { + probed = append(probed, node["name"].(string)) + return map[string]any{"egressIp": "2.2.2.2", "country": "United States", "countryCode": "US", "flag": "🇺🇸"}, nil + } + t.Cleanup(func() { probeNodeEgressFn = old }) + return &probed +} + +func TestRefreshAllSourceEgress_ProbesMissingSkipsCached(t *testing.T) { + deps := newTestDeps(t) + content := "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#cached-node\nss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.5:81#missing-node" + if _, err := deps.SourceRepo.Upsert(model.SourceRecord{ + ID: "s1", Name: "s1", Type: "local", Content: content, Enabled: true, + Filters: []model.FilterRule{}, Meta: map[string]any{}, + }); err != nil { + t.Fatalf("upsert source: %v", err) + } + // Disabled sources are never refreshed. + if _, err := deps.SourceRepo.Upsert(model.SourceRecord{ + ID: "s2", Name: "s2", Type: "local", Content: "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.6:82#disabled-node", Enabled: false, + Filters: []model.FilterRule{}, Meta: map[string]any{}, + }); err != nil { + t.Fatalf("upsert disabled source: %v", err) + } + + // Fresh cache entry (geo result) for the first node only. + deps.CacheRepo.SafePut(egressKeyForNode(t, "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#cached-node"), + `{"egressIp":"1.1.1.1","country":"Japan","countryCode":"JP","flag":"🇯🇵"}`, nil, 300) + + probed := stubProbeNodeEgress(t) + deps.refreshAllSourceEgress(context.Background()) + + if len(*probed) != 1 || (*probed)[0] != "missing-node" { + t.Fatalf("probed = %v, want exactly [missing-node] (cached and disabled skipped)", *probed) + } + // The fresh probe result must now be cached. + if _, ok := deps.CacheRepo.SafeGet(egressKeyForNode(t, "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.5:81#missing-node")); !ok { + t.Fatal("probe result should be cached after refresh") + } +} + +func TestRefreshAllSourceEgress_SkipsErrorResultsWithinTTL(t *testing.T) { + deps := newTestDeps(t) + if _, err := deps.SourceRepo.Upsert(model.SourceRecord{ + ID: "s1", Name: "s1", Type: "local", Content: "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#err-node", Enabled: true, + Filters: []model.FilterRule{}, Meta: map[string]any{}, + }); err != nil { + t.Fatalf("upsert source: %v", err) + } + deps.CacheRepo.SafePut(egressKeyForNode(t, "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#err-node"), + `{"egressError":"dial tcp: timeout"}`, nil, 300) + + probed := stubProbeNodeEgress(t) + deps.refreshAllSourceEgress(context.Background()) + + if len(*probed) != 0 { + t.Fatalf("cached error result within TTL must be skipped, probed = %v", *probed) + } +} + +func TestRefreshAllSourceEgress_ReprobesExpiredEntries(t *testing.T) { + deps := newTestDeps(t) + if _, err := deps.SourceRepo.Upsert(model.SourceRecord{ + ID: "s1", Name: "s1", Type: "local", Content: "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#stale-node", Enabled: true, + Filters: []model.FilterRule{}, Meta: map[string]any{}, + }); err != nil { + t.Fatalf("upsert source: %v", err) + } + deps.CacheRepo.SafePut(egressKeyForNode(t, "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#stale-node"), + `{"egressError":"old failure"}`, nil, 300) + // Expire every entry: SafeGet must treat them as misses. + if _, err := deps.DB.Exec("UPDATE source_cache SET cached_at = cached_at - 10000"); err != nil { + t.Fatalf("expire cache entries: %v", err) + } + + probed := stubProbeNodeEgress(t) + deps.refreshAllSourceEgress(context.Background()) + + if len(*probed) != 1 { + t.Fatalf("expired entry must be re-probed, probed = %v", *probed) + } + if _, ok := deps.CacheRepo.SafeGet(egressKeyForNode(t, "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#stale-node")); !ok { + t.Fatal("fresh result should be re-cached after re-probe") + } +} + +// StartEgressRefresher must run the first refresh immediately and stop when +// the context is cancelled. +func TestStartEgressRefresher_ImmediateRunAndCancel(t *testing.T) { + deps := newTestDeps(t) + if _, err := deps.SourceRepo.Upsert(model.SourceRecord{ + ID: "s1", Name: "s1", Type: "local", Content: "ss://YWVzLTI1Ni1nY206cGFzcw==@1.2.3.4:80#n1", Enabled: true, + Filters: []model.FilterRule{}, Meta: map[string]any{}, + }); err != nil { + t.Fatalf("upsert source: %v", err) + } + + done := make(chan string, 8) + old := probeNodeEgressFn + probeNodeEgressFn = func(d *Deps, node model.ProxyNode) (map[string]any, error) { + done <- node["name"].(string) + return map[string]any{"country": "United States"}, nil + } + t.Cleanup(func() { probeNodeEgressFn = old }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + StartEgressRefresher(ctx, deps, time.Hour) + + select { + case name := <-done: + if name != "n1" { + t.Fatalf("probed %q, want n1", name) + } + case <-time.After(5 * time.Second): + t.Fatal("refresher did not run its first pass immediately") + } + cancel() +} diff --git a/internal/handler/egress_info.go b/internal/handler/egress_info.go index 3347d2c..423b9a8 100644 --- a/internal/handler/egress_info.go +++ b/internal/handler/egress_info.go @@ -3,7 +3,6 @@ package handler import ( "bytes" "context" - "crypto/sha256" "encoding/json" "fmt" "io" @@ -20,6 +19,7 @@ import ( "gopkg.in/yaml.v3" "github.com/peterqiu0516/sub-store/internal/model" + "github.com/peterqiu0516/sub-store/internal/service" "github.com/peterqiu0516/sub-store/internal/util" ) @@ -35,7 +35,7 @@ func (d *Deps) HandleEgressInfo(c fiber.Ctx) error { node["name"] = "PROXY" } - cacheKey := egressCacheKey(node) + cacheKey := service.EgressCacheKey(node) if entry, ok := d.CacheRepo.SafeGet(cacheKey); ok { var cached map[string]any if json.Unmarshal([]byte(entry.Content), &cached) == nil { @@ -75,27 +75,9 @@ func (d *Deps) HandleEgressInfo(c fiber.Ctx) error { return success(c, info) } -func egressCacheKey(node model.ProxyNode) string { - clean := model.ProxyNode{} - skip := map[string]bool{ - "id": true, "name": true, "_sourceAlias": true, "_previewId": true, - "latencyMs": true, "latencyError": true, - "egressIp": true, "egressCountry": true, "egressRegion": true, "egressError": true, - "country": true, "countryCode": true, "region": true, "city": true, "isp": true, "flag": true, "cached": true, - } - for k, v := range node { - if !skip[k] { - clean[k] = v - } - } - data, _ := json.Marshal(clean) - sum := sha256.Sum256(data) - return fmt.Sprintf("egress:%x", sum) -} - func (d *Deps) addCachedEgressInfo(nodes []model.ProxyNode) []model.ProxyNode { for _, node := range nodes { - entry, ok := d.CacheRepo.SafeGet(egressCacheKey(node)) + entry, ok := d.CacheRepo.SafeGet(service.EgressCacheKey(node)) if !ok { continue } diff --git a/internal/handler/handler_test.go b/internal/handler/handler_test.go index 87fa4cf..fa2e497 100644 --- a/internal/handler/handler_test.go +++ b/internal/handler/handler_test.go @@ -14,11 +14,13 @@ import ( "github.com/gofiber/fiber/v3" "github.com/jmoiron/sqlx" + "gopkg.in/yaml.v3" _ "modernc.org/sqlite" "github.com/peterqiu0516/sub-store/internal/config" "github.com/peterqiu0516/sub-store/internal/database" "github.com/peterqiu0516/sub-store/internal/model" + "github.com/peterqiu0516/sub-store/internal/service" "github.com/peterqiu0516/sub-store/internal/template" ) @@ -2127,25 +2129,42 @@ func TestBuildEgressProbeConfig(t *testing.T) { if err != nil { t.Fatalf("buildEgressProbeConfig returned error: %v", err) } + // The probe config is a minimal mihomo (Clash Meta) YAML. var doc map[string]any - if err := json.Unmarshal(data, &doc); err != nil { - t.Fatalf("invalid config JSON: %v", err) + if err := yaml.Unmarshal(data, &doc); err != nil { + t.Fatalf("invalid config YAML: %v", err) } - route := doc["route"].(map[string]any) - if route["final"] != "PROXY" { - t.Fatalf("route.final = %v, want PROXY", route["final"]) + if port, _ := doc["mixed-port"].(int); port != 19090 { + t.Fatalf("mixed-port = %v, want 19090", doc["mixed-port"]) } - inbound := doc["inbounds"].([]any)[0].(map[string]any) - if inbound["listen"] != "127.0.0.1" || inbound["listen_port"].(float64) != 19090 { - t.Fatalf("unexpected inbound: %v", inbound) + proxies, _ := doc["proxies"].([]any) + if len(proxies) != 1 { + t.Fatalf("proxies = %v, want 1 entry", doc["proxies"]) + } + proxyMap, _ := proxies[0].(map[string]any) + if proxyMap["name"] != "PROXY" || proxyMap["server"] != "127.0.0.1" || proxyMap["cipher"] != "aes-256-gcm" { + t.Fatalf("unexpected probe proxy: %v", proxyMap) + } + groups, _ := doc["proxy-groups"].([]any) + group, _ := groups[0].(map[string]any) + if group["type"] != "select" { + t.Fatalf("proxy-group type = %v, want select", group["type"]) } } -func TestHandleEgressInfoUnsupported(t *testing.T) { +func TestHandleEgressInfoCacheHit(t *testing.T) { deps := newTestDeps(t) app := newApp(deps) - code, _ := doRequest(t, app, "POST", "/api/utils/egress-info", `{"type":"unknown","server":"1.2.3.4","port":443}`, nil) - assertStatus(t, "EgressInfo unsupported", code, 400) + // Any node type is probed now; a cached result must short-circuit before + // any probing so the response is deterministic without network access. + deps.CacheRepo.SafePut(service.EgressCacheKey(model.ProxyNode{"type": "ss", "server": "1.2.3.4", "port": 8388}), + `{"egressIp":"1.1.1.1","country":"Japan","latencyMs":12}`, nil, 300) + code, out := doRequest(t, app, "POST", "/api/utils/egress-info", `{"type":"ss","server":"1.2.3.4","port":8388}`, nil) + assertStatus(t, "EgressInfo cache hit", code, 200) + data, _ := out["data"].(map[string]any) + if data["egressIp"] != "1.1.1.1" || data["cached"] != true { + t.Fatalf("unexpected cached egress response: %v", data) + } } func TestProbeServerPortLatency(t *testing.T) { @@ -2180,7 +2199,7 @@ func TestAddCachedEgressInfo(t *testing.T) { "server": "127.0.0.1", "port": 8388, } - deps.CacheRepo.SafePut(egressCacheKey(node), `{"egressIp":"1.1.1.1","country":"Japan","region":"Tokyo","latencyMs":12}`, nil, 300) + deps.CacheRepo.SafePut(service.EgressCacheKey(node), `{"egressIp":"1.1.1.1","country":"Japan","region":"Tokyo","latencyMs":12}`, nil, 300) nodes := deps.addCachedEgressInfo([]model.ProxyNode{node}) if nodes[0]["egressIp"] != "1.1.1.1" || nodes[0]["latencyMs"].(float64) != 12 { t.Fatalf("cached egress not merged: %v", nodes[0]) diff --git a/internal/handler/routes.go b/internal/handler/routes.go index 2242dbe..fde482d 100644 --- a/internal/handler/routes.go +++ b/internal/handler/routes.go @@ -34,8 +34,9 @@ func NewDeps(cfg *config.Config, db *sqlx.DB) *Deps { } } -// RegisterRoutes registers all API and download routes. -func RegisterRoutes(app *fiber.App, cfg *config.Config, db *sqlx.DB) { +// RegisterRoutes wires all routes and returns the Deps used, so callers +// (e.g. server startup) can launch background jobs on the same dependencies. +func RegisterRoutes(app *fiber.App, cfg *config.Config, db *sqlx.DB) *Deps { deps := NewDeps(cfg, db) // Admin API group — requires admin token @@ -103,6 +104,7 @@ func RegisterRoutes(app *fiber.App, cfg *config.Config, db *sqlx.DB) { // Public download routes — no admin token required, uses download token app.Get("/sources/:name/:token", deps.HandleDownloadSource) app.Get("/collections/:name/:token", deps.HandleDownloadCollection) + return deps } // success sends a success JSON response. diff --git a/internal/service/egress_cache.go b/internal/service/egress_cache.go new file mode 100644 index 0000000..a84c69e --- /dev/null +++ b/internal/service/egress_cache.go @@ -0,0 +1,83 @@ +package service + +import ( + "crypto/sha256" + "encoding/json" + "fmt" + + "github.com/peterqiu0516/sub-store/internal/database" + "github.com/peterqiu0516/sub-store/internal/model" +) + +// egressCacheKeySkip lists node fields excluded from the egress cache key so +// the key stays stable across pipeline stages (parse → filter → alias tag → +// rename) and matches the key the background probe wrote. +var egressCacheKeySkip = map[string]bool{ + "id": true, "name": true, "_sourceAlias": true, "_previewId": true, + "latencyMs": true, "latencyError": true, + "egressIp": true, "egressCountry": true, "egressRegion": true, "egressError": true, + "country": true, "countryCode": true, "region": true, "city": true, "isp": true, "flag": true, "cached": true, +} + +// EgressCacheKey returns the source_cache key for a node's egress probe +// result. Pushed down from the handler (HH-773) so the download pipeline +// reads the exact entries the probe wrote. +func EgressCacheKey(node model.ProxyNode) string { + clean := model.ProxyNode{} + for k, v := range node { + if !egressCacheKeySkip[k] { + clean[k] = v + } + } + data, _ := json.Marshal(clean) + sum := sha256.Sum256(data) + return fmt.Sprintf("egress:%x", sum) +} + +// egressGeoFields are the cached egress-probe geo fields merged into nodes +// for collection-level renaming. +var egressGeoFields = []string{"country", "countryCode", "region", "city", "flag"} + +// MergeCachedEgressGeo merges cached geo fields into nodes so collection +// renaming can use probed geo data. Pure cache reads — no network I/O, no +// probing inside the download pipeline. geoFromNode in the filter package +// consumes the merged fields. +func MergeCachedEgressGeo(repo *database.CacheRepo, nodes []model.ProxyNode) []model.ProxyNode { + if repo == nil { + return nodes + } + for _, node := range nodes { + if node == nil { + continue + } + entry, ok := repo.SafeGet(EgressCacheKey(node)) + if !ok { + continue + } + var cached map[string]any + if json.Unmarshal([]byte(entry.Content), &cached) != nil { + continue + } + for _, field := range egressGeoFields { + if v, ok := cached[field]; ok && v != nil { + node[field] = v + } + } + } + return nodes +} + +// StripEgressGeoFields removes the merged geo fields after renaming so they +// do not leak into rendered subscription output. No parser or filter sets +// these fields natively (they exist on nodes only via the egress merge), so +// stripping is safe. +func StripEgressGeoFields(nodes []model.ProxyNode) { + for _, node := range nodes { + if node == nil { + continue + } + for _, field := range egressGeoFields { + delete(node, field) + } + } +} diff --git a/internal/service/egress_cache_test.go b/internal/service/egress_cache_test.go new file mode 100644 index 0000000..5e9d621 --- /dev/null +++ b/internal/service/egress_cache_test.go @@ -0,0 +1,172 @@ +package service + +import ( + "context" + "encoding/json" + "path/filepath" + "strings" + "testing" + + "github.com/jmoiron/sqlx" + _ "modernc.org/sqlite" + + "github.com/peterqiu0516/sub-store/internal/database" + "github.com/peterqiu0516/sub-store/internal/model" +) + +func newTestCacheRepo(t *testing.T) *database.CacheRepo { + t.Helper() + dir := t.TempDir() + db, err := sqlx.Open("sqlite", filepath.Join(dir, "test.db")+"?_pragma=journal_mode(WAL)&_pragma=foreign_keys(on)") + if err != nil { + t.Fatalf("failed to open db: %v", err) + } + t.Cleanup(func() { db.Close() }) + if err := database.RunMigrations(db); err != nil { + t.Fatalf("failed to run migrations: %v", err) + } + return database.NewCacheRepo(db) +} + +const usGeoJson = `{"egressIp":"67.215.229.50","country":"United States","countryCode":"US","region":"California","city":"Los Angeles","flag":"🇺🇸"}` + +// realityVlessContent mirrors the production self-built source: two nodes +// whose names carry geo hints, one (racknerd-la) that only the egress cache +// can resolve. +var realityVlessContent = strings.Join([]string{ + "vless://00000000-0000-0000-0000-000000000001@8.220.220.41:443?encryption=none&flow=xtls-rprx-vision&security=reality&sni=dl.google.com&fp=chrome&pbk=public-key-1&sid=short-id-1&spx=%2F&type=tcp&headerType=none#ali-seoul", + "vless://00000000-0000-0000-0000-000000000002@8.216.16.28:443?encryption=none&flow=xtls-rprx-vision&security=reality&sni=www.yahoo.co.jp&fp=chrome&pbk=public-key-2&sid=short-id-2&spx=%2F&type=tcp&headerType=none#Ali-Tokyo", + "vless://00000000-0000-0000-0000-000000000003@67.215.229.50:443?encryption=none&flow=xtls-rprx-vision&security=reality&sni=www.cloudflare.com&fp=chrome&pbk=public-key-3&sid=short-id-3&spx=%2F&type=tcp&headerType=none#racknerd-la", +}, "\n") + +func TestEgressCacheKey_IgnoresVolatileFields(t *testing.T) { + base := model.ProxyNode{"type": "vless", "server": "1.2.3.4", "port": 443, "uuid": "u"} + decorated := model.ProxyNode{} + for k, v := range base { + decorated[k] = v + } + for k, v := range map[string]any{ + "name": "renamed", "id": "p1", "_sourceAlias": "自建", "_previewId": "x", + "latencyMs": 12.0, "country": "US", "countryCode": "US", "city": "LA", + "region": "CA", "flag": "🇺🇸", "cached": true, "egressError": "e", "egressIp": "1.1.1.1", + } { + decorated[k] = v + } + if EgressCacheKey(base) != EgressCacheKey(decorated) { + t.Error("cache key must ignore volatile/geo decoration fields") + } + + // int vs float64 port marshal identically — key is stable across + // pipeline stages and JSON round-trips. + asFloat := model.ProxyNode{"type": "vless", "server": "1.2.3.4", "port": float64(443), "uuid": "u"} + if EgressCacheKey(base) != EgressCacheKey(asFloat) { + t.Error("cache key must be stable across int/float64 port representations") + } + + changed := model.ProxyNode{"type": "vless", "server": "1.2.3.4", "port": 8443, "uuid": "u"} + if EgressCacheKey(base) == EgressCacheKey(changed) { + t.Error("cache key must change when connection fields change") + } +} + +func TestMergeCachedEgressGeo(t *testing.T) { + repo := newTestCacheRepo(t) + node := model.ProxyNode{"type": "vless", "name": "n1", "server": "1.2.3.4", "port": 443, "uuid": "u"} + repo.SafePut(EgressCacheKey(node), usGeoJson, nil, 300) + + merged := MergeCachedEgressGeo(repo, []model.ProxyNode{node}) + if merged[0]["country"] != "United States" || merged[0]["countryCode"] != "US" || merged[0]["flag"] != "🇺🇸" { + t.Fatalf("geo fields not merged: %v", merged[0]) + } + if _, ok := merged[0]["cached"]; ok { + t.Error("pipeline merge must not set the cached marker") + } + + // Cache miss leaves the node untouched. + miss := model.ProxyNode{"type": "vless", "name": "n2", "server": "5.6.7.8", "port": 443, "uuid": "u"} + out := MergeCachedEgressGeo(repo, []model.ProxyNode{miss}) + if _, ok := out[0]["country"]; ok { + t.Error("cache miss must not add geo fields") + } +} + +func TestBuildSubscriptionResult_CollectionRenameMergesEgressCache(t *testing.T) { + repo := newTestCacheRepo(t) + src := model.SourceRecord{ID: "self", Name: "self", Alias: "自建", Type: "local", Content: realityVlessContent, Enabled: true} + collection := &model.CollectionRecord{ + ID: "daily", + Name: "daily", + SourceIds: []string{"self"}, + RenameEnabled: true, + // Production "daily" options: flag + alias + country + index, city off. + RenameOptions: &model.RenameOptions{Flag: true, Alias: true, Country: true, City: false, Index: true}, + Enabled: true, + } + + // Step 1: build without rename to obtain the parsed nodes and seed the + // egress cache for racknerd-la (the node name regex cannot resolve). + plain := &model.CollectionRecord{ID: "daily", Name: "daily", SourceIds: []string{"self"}, Enabled: true} + base, err := BuildSubscriptionResult(context.Background(), BuildOptions{Collection: plain, Sources: []model.SourceRecord{src}, Target: "json"}) + if err != nil { + t.Fatalf("base build failed: %v", err) + } + var payload struct { + Proxies []model.ProxyNode `json:"proxies"` + } + if err := json.Unmarshal([]byte(base.Body), &payload); err != nil { + t.Fatalf("parse base body: %v", err) + } + seeded := 0 + for _, n := range payload.Proxies { + if n["name"] == "racknerd-la" { + repo.SafePut(EgressCacheKey(n), usGeoJson, nil, 300) + seeded++ + } + } + if seeded != 1 { + t.Fatalf("expected to seed exactly one racknerd-la node, seeded %d", seeded) + } + + // Step 2: renamed collection build must read the cache (no probing) and + // keep all three nodes. + result, err := BuildSubscriptionResult(context.Background(), BuildOptions{ + Collection: collection, Sources: []model.SourceRecord{src}, Target: "json", CacheRepo: repo, + }) + if err != nil { + t.Fatalf("rename build failed: %v", err) + } + if result.Nodes != 3 { + t.Fatalf("expected 3 nodes, got %d: %s", result.Nodes, result.Body) + } + if !strings.Contains(result.Body, "🇺🇸 [自建] 美国") { + t.Errorf("racknerd-la should render as %q via cached geo, body: %s", "🇺🇸 [自建] 美国", result.Body) + } + if !strings.Contains(result.Body, "🇰🇷 [自建] 韩国") { + t.Errorf("ali-seoul should keep name-based geo, body: %s", result.Body) + } + // Merged geo fields must not leak into the subscription output. + if strings.Contains(result.Body, `"countryCode"`) || strings.Contains(result.Body, `"egressIp"`) { + t.Errorf("merged cache fields leaked into output: %s", result.Body) + } +} + +// Without a cache repo the rename pipeline still works via name fallback and +// geo-unknown nodes are kept under their original (alias-prefixed) names. +func TestBuildSubscriptionResult_RenameKeepsGeoUnknownNodes(t *testing.T) { + src := model.SourceRecord{ID: "self", Name: "self", Alias: "自建", Type: "local", Content: realityVlessContent, Enabled: true} + collection := &model.CollectionRecord{ + ID: "daily", Name: "daily", SourceIds: []string{"self"}, RenameEnabled: true, + RenameOptions: &model.RenameOptions{Flag: true, Alias: true, Country: true, City: false, Index: true}, + Enabled: true, + } + result, err := BuildSubscriptionResult(context.Background(), BuildOptions{Collection: collection, Sources: []model.SourceRecord{src}, Target: "json"}) + if err != nil { + t.Fatalf("build failed: %v", err) + } + if result.Nodes != 3 { + t.Fatalf("expected all 3 nodes kept, got %d: %s", result.Nodes, result.Body) + } + if !strings.Contains(result.Body, "[自建] racknerd-la") { + t.Errorf("geo-unknown node should fall back to alias + original name, body: %s", result.Body) + } +} diff --git a/internal/service/subscription.go b/internal/service/subscription.go index 2cfbb8e..092cb3e 100644 --- a/internal/service/subscription.go +++ b/internal/service/subscription.go @@ -149,9 +149,13 @@ func loadProxyNodes(ctx context.Context, opts BuildOptions) ([]model.ProxyNode, originalNodes += count } - // Apply collection-level rename if enabled + // Apply collection-level rename if enabled. Cached egress geo fields are + // merged in first (pure cache reads, HH-773) and stripped after renaming so + // they don't leak into rendered output. if opts.Collection != nil && opts.Collection.RenameEnabled { + allProxies = MergeCachedEgressGeo(opts.CacheRepo, allProxies) allProxies = filter.RenameCollectionNodes(allProxies, opts.Collection.RenameOptions) + StripEgressGeoFields(allProxies) } // Ensure unique names