fix: improve telegram channel connectivity

internal/service/notifier.go

@@ -3,10 +3,10 @@
 // NotifierService dispatches structured messages to one or more channels
 // configured in the system settings table:
 //
-//   notify.telegram.bot_token + notify.telegram.chat_id
-//   notify.bark.server + notify.bark.key
-//   notify.wechat.sendkey
-//   notify.webhook.url + notify.webhook.method
+//	notify.telegram.bot_token + notify.telegram.chat_id
+//	notify.bark.server + notify.bark.key
+//	notify.wechat.sendkey
+//	notify.webhook.url + notify.webhook.method
 //
 // Notifications are triggered by the subscription poller, the download
 // poller, the scan / scrape completions, and any future event worth
@@ -38,7 +38,7 @@ func NewNotifierService(log *zap.Logger, repo *repository.Container) *NotifierSe
 	return &NotifierService{
 		log:    log,
 		repo:   repo,
-		client: &http.Client{Timeout: 10 * time.Second},
+		client: NewExternalHTTPClient(10 * time.Second),
 	}
 }
 
@@ -63,17 +63,20 @@ func (n *NotifierService) sendTelegram(ctx context.Context, title, body string)
 		return
 	}
 	text := fmt.Sprintf("<b>%s</b>\n\n%s", escapeHTML(title), escapeHTML(body))
-	u := fmt.Sprintf("https://api.telegram.org/bot%s/sendMessage", token)
 	form := url.Values{}
 	form.Set("chat_id", chatID)
 	form.Set("text", text)
 	form.Set("parse_mode", "HTML")
-	resp, err := n.client.PostForm(u, form)
-	if err != nil {
-		n.log.Debug("telegram notify failed", zap.Error(err))
-		return
+	cfg := map[string]string{"bot_token": token}
+	if apiBase := n.get(ctx, "notify.telegram.api_base_url"); apiBase != "" {
+		cfg["api_base_url"] = apiBase
+	}
+	if proxyURL := n.get(ctx, "notify.telegram.proxy_url"); proxyURL != "" {
+		cfg["proxy_url"] = proxyURL
+	}
+	if err := telegramPostForm(ctx, cfg, "sendMessage", form, 15*time.Second); err != nil {
+		n.log.Debug("telegram notify failed", zap.Error(err))
 	}
-	defer resp.Body.Close()
 }
 
 func (n *NotifierService) sendBark(ctx context.Context, title, body string) {

internal/service/notify_channels.go

@@ -34,7 +34,7 @@ func NewNotifyChannelService(log *zap.Logger, repo *repository.Container) *Notif
 	return &NotifyChannelService{
 		log:    log,
 		repo:   repo,
-		client: &http.Client{Timeout: 10 * time.Second},
+		client: NewExternalHTTPClient(10 * time.Second),
 	}
 }
 
@@ -89,6 +89,7 @@ func (s *NotifyChannelService) List(ctx context.Context) ([]channelView, error)
 
 // Create persists a new channel.
 func (s *NotifyChannelService) Create(ctx context.Context, in ChannelInput) (*channelView, error) {
+	normalizeChannelInput(&in)
 	if err := validateChannel(in); err != nil {
 		return nil, err
 	}
@@ -113,6 +114,7 @@ func (s *NotifyChannelService) Create(ctx context.Context, in ChannelInput) (*ch
 
 // Update applies a partial patch to an existing channel.
 func (s *NotifyChannelService) Update(ctx context.Context, id string, in ChannelInput) (*channelView, error) {
+	normalizeChannelInput(&in)
 	if err := validateChannel(in); err != nil {
 		return nil, err
 	}
@@ -218,20 +220,18 @@ func (s *NotifyChannelService) dispatchOne(ctx context.Context, n model.NotifyCh
 
 	switch n.Type {
 	case "telegram":
-		token := str(cfg["bot_token"])
-		chat := str(cfg["chat_id"])
+		telegramCfg := telegramStringConfigFromAny(cfg)
+		token := telegramCfg["bot_token"]
+		chat := telegramCfg["chat_id"]
 		if token == "" || chat == "" {
 			return errors.New("telegram missing bot_token / chat_id")
 		}
 		text := fmt.Sprintf("<b>%s</b>\n\n%s", escapeHTML(title), escapeHTML(body))
-		u := fmt.Sprintf("https://api.telegram.org/bot%s/sendMessage", token)
 		form := url.Values{}
 		form.Set("chat_id", chat)
 		form.Set("text", text)
 		form.Set("parse_mode", "HTML")
-		req, _ := http.NewRequestWithContext(ctx, http.MethodPost, u, strings.NewReader(form.Encode()))
-		req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-		return s.do(req)
+		return telegramPostForm(ctx, telegramCfg, "sendMessage", form, 15*time.Second)
 
 	case "bark":
 		key := str(cfg["device_key"])
@@ -332,12 +332,28 @@ func validateChannel(in ChannelInput) error {
 			return errors.New("telegram admin_user_ids required")
 		}
 		if str(cfg["group_chat_id"]) == "" && str(cfg["channel_chat_id"]) == "" && str(cfg["command_chat_id"]) == "" {
-			return errors.New("telegram group_chat_id or channel_chat_id required")
+			return errors.New("telegram group_chat_id or channel_chat_id required: 请填写绑定群组 ID 或绑定频道 ID；如果通知 Chat ID 是群组/频道负数 ID，也可以直接填在 Chat ID")
 		}
 	}
 	return nil
 }
 
+func normalizeChannelInput(in *ChannelInput) {
+	if in == nil || in.Type != "telegram" {
+		return
+	}
+	if in.Config == nil {
+		in.Config = map[string]any{}
+	}
+	chatID := str(in.Config["chat_id"])
+	if chatID == "" || !strings.HasPrefix(chatID, "-") {
+		return
+	}
+	if str(in.Config["group_chat_id"]) == "" && str(in.Config["channel_chat_id"]) == "" && str(in.Config["command_chat_id"]) == "" {
+		in.Config["group_chat_id"] = chatID
+	}
+}
+
 // str safely extracts a string from an interface{} loaded from JSON.
 func str(v any) string {
 	if v == nil {

internal/service/notify_telegram.go

@@ -2,12 +2,8 @@
 package service
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
 	"fmt"
-	"io"
-	"net/http"
 	"strings"
 	"time"
 )
@@ -35,30 +31,7 @@ func (p *TelegramProvider) Send(ctx context.Context, cfg map[string]string, even
 		"text":       text,
 		"parse_mode": parseMode,
 	}
-	body, err := json.Marshal(payload)
-	if err != nil {
-		return err
-	}
-
-	apiURL := fmt.Sprintf("https://api.telegram.org/bot%s/sendMessage", botToken)
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(body))
-	if err != nil {
-		return err
-	}
-	req.Header.Set("Content-Type", "application/json")
-
-	client := &http.Client{Timeout: 15 * time.Second}
-	resp, err := client.Do(req)
-	if err != nil {
-		return err
-	}
-	defer resp.Body.Close()
-
-	respBody, _ := io.ReadAll(resp.Body)
-	if resp.StatusCode >= 400 {
-		return fmt.Errorf("telegram api error %d: %s", resp.StatusCode, string(respBody))
-	}
-	return nil
+	return telegramPostJSON(ctx, cfg, "sendMessage", payload, 15*time.Second)
 }
 
 // ValidateConfig 验证 Telegram 配置。

internal/service/telegram_api.go

@@ -0,0 +1,126 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"regexp"
+	"strings"
+	"time"
+)
+
+const defaultTelegramAPIBaseURL = "https://api.telegram.org"
+
+var telegramTokenPattern = regexp.MustCompile(`bot[0-9]+:[^/\s"'?]+`)
+
+func telegramAPIBaseURL(cfg map[string]string) string {
+	base := strings.TrimSpace(cfg["api_base_url"])
+	if base == "" {
+		base = strings.TrimSpace(os.Getenv("MEDIASTATION_TELEGRAM_API_BASE_URL"))
+	}
+	if base == "" {
+		base = defaultTelegramAPIBaseURL
+	}
+	return strings.TrimRight(base, "/")
+}
+
+func telegramMethodURL(cfg map[string]string, botToken, method string) (string, error) {
+	botToken = strings.TrimSpace(botToken)
+	method = strings.TrimSpace(method)
+	if botToken == "" {
+		return "", errors.New("telegram bot_token required")
+	}
+	if method == "" {
+		return "", errors.New("telegram method required")
+	}
+	base := telegramAPIBaseURL(cfg)
+	if _, err := url.ParseRequestURI(base); err != nil {
+		return "", fmt.Errorf("telegram api_base_url invalid")
+	}
+	return fmt.Sprintf("%s/bot%s/%s", base, botToken, method), nil
+}
+
+func telegramHTTPClient(timeout time.Duration, cfg map[string]string) *http.Client {
+	transport := NewExternalTransport()
+	proxyRaw := strings.TrimSpace(cfg["proxy_url"])
+	if proxyRaw == "" {
+		proxyRaw = strings.TrimSpace(os.Getenv("MEDIASTATION_TELEGRAM_PROXY_URL"))
+	}
+	if proxyRaw != "" {
+		if proxyURL, err := normalizeProxyURL(proxyRaw, "http"); err == nil {
+			transport.Proxy = http.ProxyURL(proxyURL)
+		}
+	}
+	return &http.Client{Timeout: timeout, Transport: transport}
+}
+
+func telegramPostForm(ctx context.Context, cfg map[string]string, method string, form url.Values, timeout time.Duration) error {
+	apiURL, err := telegramMethodURL(cfg, cfg["bot_token"], method)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, strings.NewReader(form.Encode()))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	return telegramDo(telegramHTTPClient(timeout, cfg), req)
+}
+
+func telegramPostJSON(ctx context.Context, cfg map[string]string, method string, payload any, timeout time.Duration) error {
+	apiURL, err := telegramMethodURL(cfg, cfg["bot_token"], method)
+	if err != nil {
+		return err
+	}
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, strings.NewReader(string(body)))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	return telegramDo(telegramHTTPClient(timeout, cfg), req)
+}
+
+func telegramDo(client *http.Client, req *http.Request) error {
+	resp, err := client.Do(req)
+	if err != nil {
+		return sanitizeTelegramError(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode >= 400 {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		return fmt.Errorf("telegram api error %d: %s", resp.StatusCode, sanitizeTelegramText(string(body)))
+	}
+	return nil
+}
+
+func telegramStringConfigFromAny(cfg map[string]any) map[string]string {
+	out := make(map[string]string, len(cfg))
+	for key, value := range cfg {
+		out[key] = str(value)
+	}
+	return out
+}
+
+func sanitizeTelegramError(err error) error {
+	if err == nil {
+		return nil
+	}
+	msg := sanitizeTelegramText(err.Error())
+	if strings.Contains(msg, "Client.Timeout exceeded") || strings.Contains(msg, "context deadline exceeded") {
+		return errors.New("telegram request timeout: 请检查 NAS/Docker 到 Telegram API 的代理、反代或网络连通性")
+	}
+	return errors.New(msg)
+}
+
+func sanitizeTelegramText(text string) string {
+	return telegramTokenPattern.ReplaceAllString(text, "bot<redacted>")
+}

internal/service/telegram_api_test.go

@@ -0,0 +1,34 @@
+package service
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestTelegramMethodURLUsesCustomAPIBase(t *testing.T) {
+	got, err := telegramMethodURL(map[string]string{
+		"api_base_url": "https://tg.example.com/",
+	}, "123456:ABC-def", "sendMessage")
+	if err != nil {
+		t.Fatalf("telegramMethodURL returned error: %v", err)
+	}
+	want := "https://tg.example.com/bot123456:ABC-def/sendMessage"
+	if got != want {
+		t.Fatalf("got %q, want %q", got, want)
+	}
+}
+
+func TestSanitizeTelegramErrorRedactsBotToken(t *testing.T) {
+	err := sanitizeTelegramError(errors.New(`Post "https://api.telegram.org/bot123456:SECRET/sendMessage": context deadline exceeded`))
+	if err == nil {
+		t.Fatal("expected error")
+	}
+	msg := err.Error()
+	if strings.Contains(msg, "SECRET") || strings.Contains(msg, "123456:") {
+		t.Fatalf("telegram token leaked in error: %s", msg)
+	}
+	if !strings.Contains(msg, "timeout") {
+		t.Fatalf("expected timeout hint, got: %s", msg)
+	}
+}

internal/service/telegram_bot.go

@@ -5,7 +5,6 @@
 package service
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -478,7 +477,7 @@ func (s *TelegramBotService) StartPolling(ctx context.Context) {
 		s.pollingCancel[botToken] = cancel
 		s.pollingMu.Unlock()
 
-		go s.pollLoop(pollCtx, botToken)
+		go s.pollLoop(pollCtx, cfg)
 		s.log.Info("started telegram polling", zap.String("channel", ch.Name))
 	}
 }
@@ -495,10 +494,14 @@ func (s *TelegramBotService) StopPolling() {
 }
 
 // pollLoop 对单个 Bot Token 执行长轮询。
-func (s *TelegramBotService) pollLoop(ctx context.Context, botToken string) {
+func (s *TelegramBotService) pollLoop(ctx context.Context, cfg map[string]string) {
 	var offset int64 = 0
-	pollURL := fmt.Sprintf("https://api.telegram.org/bot%s/getUpdates", botToken)
-	client := &http.Client{Timeout: 45 * time.Second}
+	pollURL, err := telegramMethodURL(cfg, cfg["bot_token"], "getUpdates")
+	if err != nil {
+		s.log.Warn("telegram polling config invalid", zap.Error(err))
+		return
+	}
+	client := telegramHTTPClient(45*time.Second, cfg)
 
 	for {
 		select {
@@ -511,7 +514,7 @@ func (s *TelegramBotService) pollLoop(ctx context.Context, botToken string) {
 			"offset":  offset,
 			"timeout": 30,
 		})
-		req, err := http.NewRequestWithContext(ctx, http.MethodPost, pollURL, bytes.NewReader(reqBody))
+		req, err := http.NewRequestWithContext(ctx, http.MethodPost, pollURL, strings.NewReader(string(reqBody)))
 		if err != nil {
 			time.Sleep(5 * time.Second)
 			continue
@@ -520,6 +523,7 @@ func (s *TelegramBotService) pollLoop(ctx context.Context, botToken string) {
 
 		resp, err := client.Do(req)
 		if err != nil {
+			s.log.Debug("telegram polling failed", zap.Error(sanitizeTelegramError(err)))
 			time.Sleep(5 * time.Second)
 			continue
 		}
@@ -554,18 +558,15 @@ func (s *TelegramBotService) pollLoop(ctx context.Context, botToken string) {
 
 // reply 通过 Telegram Bot API 发送回复消息。
 func (s *TelegramBotService) reply(ctx context.Context, channel *model.NotifyChannel, chatID int, reply telegramCommandReply) error {
-	botToken := ""
+	cfg := map[string]string{}
 	if channel != nil {
 		configStr := channel.Config
 		if s.crypto != nil && configStr != "" {
 			configStr = s.crypto.Decrypt(configStr)
 		}
-		var cfg map[string]string
-		if err := json.Unmarshal([]byte(configStr), &cfg); err == nil {
-			botToken = cfg["bot_token"]
-		}
+		_ = json.Unmarshal([]byte(configStr), &cfg)
 	}
-	if botToken == "" {
+	if strings.TrimSpace(cfg["bot_token"]) == "" {
 		return fmt.Errorf("bot_token not configured")
 	}
 
@@ -588,27 +589,7 @@ func (s *TelegramBotService) reply(ctx context.Context, channel *model.NotifyCha
 		}
 		payload["reply_markup"] = map[string]interface{}{"inline_keyboard": keyboard}
 	}
-	body, _ := json.Marshal(payload)
-
-	apiURL := fmt.Sprintf("https://api.telegram.org/bot%s/sendMessage", botToken)
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, bytes.NewReader(body))
-	if err != nil {
-		return err
-	}
-	req.Header.Set("Content-Type", "application/json")
-
-	client := &http.Client{Timeout: 15 * time.Second}
-	resp, err := client.Do(req)
-	if err != nil {
-		return err
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode >= 400 {
-		respBody, _ := io.ReadAll(resp.Body)
-		return fmt.Errorf("telegram api error %d: %s", resp.StatusCode, string(respBody))
-	}
-	return nil
+	return telegramPostJSON(ctx, cfg, "sendMessage", payload, 15*time.Second)
 }
 
 // findChannelByChatID 根据 chat_id 查找已配置的通知渠道。
@@ -756,24 +737,27 @@ func (s *TelegramBotService) telegramUserCanBind(ctx context.Context, channel *m
 }
 
 func (s *TelegramBotService) telegramUserIsChatMember(ctx context.Context, channel *model.NotifyChannel, chatID string, telegramUserID int) bool {
-	token := strings.TrimSpace(s.telegramChannelConfig(channel)["bot_token"])
-	if token == "" || chatID == "" || telegramUserID == 0 {
+	cfg := s.telegramChannelConfig(channel)
+	if strings.TrimSpace(cfg["bot_token"]) == "" || chatID == "" || telegramUserID == 0 {
 		return false
 	}
-	payload, _ := json.Marshal(map[string]interface{}{
+	payload := map[string]interface{}{
 		"chat_id": chatID,
 		"user_id": telegramUserID,
-	})
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
-		fmt.Sprintf("https://api.telegram.org/bot%s/getChatMember", token),
-		bytes.NewReader(payload))
+	}
+	apiURL, err := telegramMethodURL(cfg, cfg["bot_token"], "getChatMember")
+	if err != nil {
+		return false
+	}
+	body, _ := json.Marshal(payload)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, strings.NewReader(string(body)))
 	if err != nil {
 		return false
 	}
 	req.Header.Set("Content-Type", "application/json")
-	resp, err := (&http.Client{Timeout: 8 * time.Second}).Do(req)
+	resp, err := telegramHTTPClient(8*time.Second, cfg).Do(req)
 	if err != nil {
-		s.log.Warn("telegram getChatMember failed", zap.String("chat_id", chatID), zap.Int("telegram_user_id", telegramUserID), zap.Error(err))
+		s.log.Warn("telegram getChatMember failed", zap.String("chat_id", chatID), zap.Int("telegram_user_id", telegramUserID), zap.Error(sanitizeTelegramError(err)))
 		return false
 	}
 	defer resp.Body.Close()
@@ -884,33 +868,40 @@ func (s *TelegramBotService) SetWebhook(ctx context.Context, botToken, webhookUR
 		"url":             webhookURL,
 		"allowed_updates": []string{"message"},
 	})
-	req, _ := http.NewRequestWithContext(ctx, http.MethodPost,
-		fmt.Sprintf("https://api.telegram.org/bot%s/setWebhook", botToken),
-		bytes.NewReader(payload))
-	req.Header.Set("Content-Type", "application/json")
-
-	client := &http.Client{Timeout: 15 * time.Second}
-	resp, err := client.Do(req)
+	cfg := map[string]string{"bot_token": botToken}
+	apiURL, err := telegramMethodURL(cfg, botToken, "setWebhook")
 	if err != nil {
 		return err
 	}
+	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, apiURL, strings.NewReader(string(payload)))
+	req.Header.Set("Content-Type", "application/json")
+
+	client := telegramHTTPClient(15*time.Second, cfg)
+	resp, err := client.Do(req)
+	if err != nil {
+		return sanitizeTelegramError(err)
+	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode >= 400 {
 		body, _ := io.ReadAll(resp.Body)
-		return fmt.Errorf("setWebhook failed: %s", string(body))
+		return fmt.Errorf("setWebhook failed: %s", sanitizeTelegramText(string(body)))
 	}
 	return nil
 }
 
 // GetWebhookInfo 获取 Webhook 配置信息。
 func (s *TelegramBotService) GetWebhookInfo(ctx context.Context, botToken string) (map[string]interface{}, error) {
-	resp, err := (&http.Client{Timeout: 10 * time.Second}).Get(
-		fmt.Sprintf("https://api.telegram.org/bot%s/getWebhookInfo", botToken),
-	)
+	cfg := map[string]string{"bot_token": botToken}
+	apiURL, err := telegramMethodURL(cfg, botToken, "getWebhookInfo")
 	if err != nil {
 		return nil, err
 	}
+	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil)
+	resp, err := telegramHTTPClient(10*time.Second, cfg).Do(req)
+	if err != nil {
+		return nil, sanitizeTelegramError(err)
+	}
 	defer resp.Body.Close()
 
 	var result map[string]interface{}