HH-683: propagate subscription fetch failures (#1)

internal/handler/download_tools.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/gofiber/fiber/v3"
+	"github.com/sirupsen/logrus"
 
 	"github.com/peterqiu0516/sub-store/internal/middleware"
 	"github.com/peterqiu0516/sub-store/internal/model"
@@ -55,7 +56,8 @@ func (d *Deps) HandleDownloadCollection(c fiber.Ctx) error {
 		ProxyURL:         d.Cfg.Fetcher.ProxyURL,
 	})
 	if err != nil {
-		return failed(c, err.Error(), 500)
+		logrus.WithError(err).Error("Failed to build subscription")
+		return failed(c, "Failed to build subscription", 500)
 	}
 	return d.sendDownloadResponse(c, result, target)
 }
@@ -86,7 +88,8 @@ func (d *Deps) HandleDownloadSource(c fiber.Ctx) error {
 		ProxyURL:         d.Cfg.Fetcher.ProxyURL,
 	})
 	if err != nil {
-		return failed(c, err.Error(), 500)
+		logrus.WithError(err).Error("Failed to build subscription")
+		return failed(c, "Failed to build subscription", 500)
 	}
 	return d.sendDownloadResponse(c, result, target)
 }

internal/handler/handler_test.go

@@ -1836,6 +1836,37 @@ func TestHandleDownloadSourceDisabled(t *testing.T) {
 	assertStatus(t, "DownloadSource disabled", code, 404)
 }
 
+func TestHandleDownloadBuildErrorDoesNotLeakUpstreamURL(t *testing.T) {
+	for _, kind := range []string{"source", "collection"} {
+		t.Run(kind, func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
+			upstreamURL := server.URL + "/subscription?token=upstream-secret"
+			server.Close()
+
+			deps := newTestDeps(t)
+			app := newApp(deps)
+			deps.SourceRepo.Upsert(model.SourceRecord{ID: "remote", Name: "Remote", Type: "remote", URL: upstreamURL, Enabled: true})
+
+			path := "/sources/remote/dl-tok?target=json"
+			if kind == "collection" {
+				deps.CollectionRepo.Upsert(model.CollectionRecord{ID: "remote", Name: "Remote", SourceIds: []string{"remote"}, Enabled: true})
+				path = "/collections/remote/dl-tok?target=json"
+			}
+
+			code, body := doRequest(t, app, "GET", path, "", nil)
+			assertStatus(t, kind, code, http.StatusInternalServerError)
+			errorBody, ok := body["error"].(map[string]any)
+			if !ok || errorBody["message"] != "Failed to build subscription" {
+				t.Fatalf("response = %v, want fixed generic error", body)
+			}
+			response, _ := json.Marshal(body)
+			if strings.Contains(string(response), "upstream-secret") || strings.Contains(string(response), upstreamURL) {
+				t.Fatalf("response leaked upstream URL: %s", response)
+			}
+		})
+	}
+}
+
 // ---------------------------------------------------------------------------
 // Preview handlers
 // ---------------------------------------------------------------------------

internal/service/subscription.go

@@ -2,9 +2,9 @@ package service
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
-	"math"
 	"net/http"
 	"net/url"
 	"sort"
@@ -87,7 +87,10 @@ func loadProxyNodes(ctx context.Context, opts BuildOptions) ([]model.ProxyNode,
 		i := i
 		sub := sub
 		tasks[i] = func() ([]model.ProxyNode, error) {
-			raw, meta := loadSubscriptionRaw(ctx, sub, opts)
+			raw, meta, err := loadSubscriptionRaw(ctx, sub, opts)
+			if err != nil {
+				return nil, err
+			}
 			metadataByIndex[i] = meta
 			nodes := proxy.ParseProxies(raw)
 			originalCounts[i] = len(nodes)
@@ -193,14 +196,14 @@ func getSources(opts BuildOptions) []model.SourceRecord {
 	return result
 }
 
-func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts BuildOptions) (string, model.SubscriptionResponseMetadata) {
+func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts BuildOptions) (string, model.SubscriptionResponseMetadata, error) {
 	if sub.Type == "local" || sub.Content != "" {
-		return sub.Content + sub.URL, metadataFromSource(sub)
+		return sub.Content, metadataFromSource(sub), nil
 	}
 
 	urls := splitSourceUrls(sub.URL)
 	if len(urls) == 0 {
-		return "", metadataFromSource(sub)
+		return "", metadataFromSource(sub), nil
 	}
 	if len(urls) > util.MaxRemoteSourceUrls {
 		urls = urls[:util.MaxRemoteSourceUrls]
@@ -215,7 +218,10 @@ func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts Build
 		}
 	}
 
-	results, _ := RunWithConcurrencyT(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings))
+	results, err := RunWithConcurrency(tasks, getConcurrency(opts.Settings), getConcurrencyWait(opts.Settings))
+	if err != nil {
+		return "", model.SubscriptionResponseMetadata{}, err
+	}
 
 	var contents []string
 	var metadata model.SubscriptionResponseMetadata
@@ -226,7 +232,7 @@ func loadSubscriptionRaw(ctx context.Context, sub model.SourceRecord, opts Build
 		}
 	}
 
-	return strings.Join(contents, "\n"), metadata
+	return strings.Join(contents, "\n"), metadata, nil
 }
 
 type fetchResult struct {
@@ -491,6 +497,7 @@ func RunWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait
 		concurrency = len(tasks)
 	}
 	results := make([]T, len(tasks))
+	errs := make([]error, len(tasks))
 	cursor := 0
 	var mu sync.Mutex
 	var wg sync.WaitGroup
@@ -512,8 +519,8 @@ func RunWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait
 				}
 				val, err := tasks[idx]()
 				if err != nil {
-					results[idx] = val // zero value
-					// In non-settled mode, we could cancel, but let's keep it simple
+					results[idx] = val
+					errs[idx] = err
 					logrus.WithError(err).Warn("task failed")
 					continue
 				}
@@ -522,12 +529,7 @@ func RunWithConcurrency[T any](tasks []func() (T, error), concurrency int, wait
 		}()
 	}
 	wg.Wait()
-	return results, nil
-}
-
-// RunWithConcurrencyT is a variant for tasks returning a value + metadata.
-func RunWithConcurrencyT[T any](tasks []func() (T, error), concurrency int, wait time.Duration) ([]T, error) {
-	return RunWithConcurrency(tasks, concurrency, wait)
+	return results, errors.Join(errs...)
 }
 
 // RunSettledWithConcurrency runs tasks with allSettled semantics.
@@ -618,9 +620,6 @@ func (aw *AsyncWriter) WaitWithTimeout(timeout time.Duration) {
 	}
 }
 
-// Prevent unused import
-var _ = math.MaxInt32
-
 // buildHTTPClient creates an *http.Client with optional proxy support.
 // If proxyURL is empty, a standard client is returned.
 func buildHTTPClient(timeout time.Duration, proxyURL string) *http.Client {

internal/service/subscription_test.go

@@ -2,7 +2,11 @@ package service
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -38,6 +42,50 @@ func TestBuildSubscriptionResult_LocalSource(t *testing.T) {
 	}
 }
 
+func TestBuildSubscriptionResult_RealityVlessSubscription(t *testing.T) {
+	content := 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")
+	source := &model.SourceRecord{ID: "reality", Type: "local", Content: content, Enabled: true}
+
+	result, err := BuildSubscriptionResult(context.Background(), BuildOptions{Source: source, Target: "json"})
+	if err != nil {
+		t.Fatalf("build failed: %v", err)
+	}
+	if result.OriginalNodes != 3 || result.Nodes != 3 {
+		t.Fatalf("nodes = %d/%d, want 3/3", result.OriginalNodes, result.Nodes)
+	}
+	var output struct {
+		Proxies []model.ProxyNode `json:"proxies"`
+	}
+	if err := json.Unmarshal([]byte(result.Body), &output); err != nil {
+		t.Fatalf("decode output: %v", err)
+	}
+	for _, node := range output.Proxies {
+		if node["type"] != "vless" || node["flow"] != "xtls-rprx-vision" || node["network"] != "tcp" {
+			t.Fatalf("VLESS fields lost: %#v", node)
+		}
+		if reality, ok := node["reality-opts"].(map[string]any); !ok || reality["public-key"] == "" || reality["short-id"] == "" {
+			t.Fatalf("Reality fields lost: %#v", node["reality-opts"])
+		}
+	}
+}
+
+func TestBuildSubscriptionResult_RemoteFetchError(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		http.Error(w, "blocked", http.StatusForbidden)
+	}))
+	defer server.Close()
+	source := &model.SourceRecord{ID: "blocked", Name: "blocked", Type: "remote", URL: server.URL, Enabled: true}
+
+	_, err := BuildSubscriptionResult(context.Background(), BuildOptions{Source: source, Target: "json"})
+	if err == nil || !strings.Contains(err.Error(), "Remote source blocked failed: 403") {
+		t.Fatalf("error = %v, want upstream 403", err)
+	}
+}
+
 func TestBuildSubscriptionResult_NoEnabledSources(t *testing.T) {
 	source := &model.SourceRecord{
 		ID:      "test-src",
@@ -520,18 +568,19 @@ func TestRunWithConcurrency(t *testing.T) {
 }
 
 func TestRunWithConcurrency_WithErrors(t *testing.T) {
+	taskErr := errors.New("task error")
 	tasks := make([]func() (int, error), 5)
 	for i := range tasks {
 		i := i
 		if i == 2 {
-			tasks[i] = func() (int, error) { return 0, errors.New("task error") }
+			tasks[i] = func() (int, error) { return 0, taskErr }
 		} else {
 			tasks[i] = func() (int, error) { return i, nil }
 		}
 	}
 	results, err := RunWithConcurrency(tasks, 2, 0)
-	if err != nil {
-		t.Fatalf("unexpected error: %v", err)
+	if !errors.Is(err, taskErr) {
+		t.Fatalf("error = %v, want task error", err)
 	}
 	if len(results) != 5 {
 		t.Fatalf("expected 5 results, got %d", len(results))
@@ -567,21 +616,6 @@ func TestRunWithConcurrency_ConcurrencyOne(t *testing.T) {
 	}
 }
 
-func TestRunWithConcurrencyT(t *testing.T) {
-	tasks := make([]func() (string, error), 3)
-	for i := range tasks {
-		i := i
-		tasks[i] = func() (string, error) { return string(rune('a' + i)), nil }
-	}
-	results, err := RunWithConcurrencyT(tasks, 2, 0)
-	if err != nil {
-		t.Fatalf("unexpected error: %v", err)
-	}
-	if len(results) != 3 {
-		t.Errorf("expected 3 results, got %d", len(results))
-	}
-}
-
 func TestRunSettledWithConcurrency(t *testing.T) {
 	tasks := make([]func() (int, error), 5)
 	for i := range tasks {
@@ -668,7 +702,10 @@ func TestLoadSubscriptionRaw_LocalSource(t *testing.T) {
 		Type:    "local",
 		Content: "ss://pass@host:80#Node1",
 	}
-	raw, meta := loadSubscriptionRaw(context.Background(), sub, BuildOptions{})
+	raw, meta, err := loadSubscriptionRaw(context.Background(), sub, BuildOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
 	if raw != "ss://pass@host:80#Node1" {
 		t.Errorf("expected content, got %s", raw)
 	}
@@ -682,7 +719,10 @@ func TestLoadSubscriptionRaw_RemoteNoUrls(t *testing.T) {
 		Type: "remote",
 		URL:  "",
 	}
-	raw, _ := loadSubscriptionRaw(context.Background(), sub, BuildOptions{})
+	raw, _, err := loadSubscriptionRaw(context.Background(), sub, BuildOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
 	if raw != "" {
 		t.Errorf("expected empty, got %s", raw)
 	}
@@ -695,9 +735,12 @@ func TestLoadSubscriptionRaw_ContentPresent(t *testing.T) {
 		Content: "ss://pass@host:80#Node1",
 		URL:     "https://example.com/sub",
 	}
-	raw, _ := loadSubscriptionRaw(context.Background(), sub, BuildOptions{})
-	if raw != "ss://pass@host:80#Node1https://example.com/sub" {
-		t.Errorf("expected content+url, got %s", raw)
+	raw, _, err := loadSubscriptionRaw(context.Background(), sub, BuildOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if raw != "ss://pass@host:80#Node1" {
+		t.Errorf("expected content only, got %s", raw)
 	}
 }