diff options
| -rw-r--r-- | sub/subController.go | 4 | ||||
| -rw-r--r-- | web/service/inbound.go | 12 | ||||
| -rw-r--r-- | web/service/setting.go | 40 | ||||
| -rw-r--r-- | web/service/tgbot.go | 119 |
4 files changed, 119 insertions, 56 deletions
diff --git a/sub/subController.go b/sub/subController.go index e7ced42b..79ea755d 100644 --- a/sub/subController.go +++ b/sub/subController.go @@ -143,11 +143,11 @@ func (a *SUBController) subs(c *gin.Context) { // Add headers header := fmt.Sprintf("upload=%d; download=%d; total=%d; expire=%d", traffic.Up, traffic.Down, traffic.Total, traffic.ExpiryTime/1000) - profileUrl := a.subProfileUrl + profileUrl := a.subProfileUrl if profileUrl == "" { profileUrl = fmt.Sprintf("%s://%s%s", scheme, hostWithPort, c.Request.RequestURI) } - a.ApplyCommonHeaders(c, header, a.updateInterval, a.subTitle, a.subSupportUrl, profileUrl, a.subAnnounce, a.subEnableRouting, a.subRoutingRules) + a.ApplyCommonHeaders(c, header, a.updateInterval, a.subTitle, a.subSupportUrl, profileUrl, a.subAnnounce, a.subEnableRouting, a.subRoutingRules) if a.subEncrypt { c.String(200, base64.StdEncoding.EncodeToString([]byte(result))) diff --git a/web/service/inbound.go b/web/service/inbound.go index ec51bc27..101c79d9 100644 --- a/web/service/inbound.go +++ b/web/service/inbound.go @@ -2141,25 +2141,25 @@ func (s *InboundService) GetInboundClientIps(clientEmail string) (string, error) if err != nil { return "", err } - + if InboundClientIps.Ips == "" { return "", nil } - + // Try to parse as new format (with timestamps) type IPWithTimestamp struct { IP string `json:"ip"` Timestamp int64 `json:"timestamp"` } - + var ipsWithTime []IPWithTimestamp err = json.Unmarshal([]byte(InboundClientIps.Ips), &ipsWithTime) - + // If successfully parsed as new format, return with timestamps if err == nil && len(ipsWithTime) > 0 { return InboundClientIps.Ips, nil } - + // Otherwise, assume it's old format (simple string array) // Try to parse as simple array and convert to new format var oldIps []string @@ -2176,7 +2176,7 @@ func (s *InboundService) GetInboundClientIps(clientEmail string) (string, error) result, _ := json.Marshal(newIpsWithTime) return string(result), nil } - + // Return as-is if parsing fails return InboundClientIps.Ips, nil } diff --git a/web/service/setting.go b/web/service/setting.go index 1e95c2df..33012c39 100644 --- a/web/service/setting.go +++ b/web/service/setting.go @@ -5,7 +5,7 @@ import ( "encoding/json" "errors" "fmt" - "net" + "net" "reflect" "strconv" "strings" @@ -719,25 +719,25 @@ func (s *SettingService) GetDefaultXrayConfig() (any, error) { } func extractHostname(host string) string { - h, _, err := net.SplitHostPort(host) - // Err is not nil means host does not contain port - if err != nil { - h = host - } - - ip := net.ParseIP(h) - // If it's not an IP, return as is - if ip == nil { - return h - } - - // If it's an IPv4, return as is - if ip.To4() != nil { - return h - } - - // IPv6 needs bracketing - return "[" + h + "]" + h, _, err := net.SplitHostPort(host) + // Err is not nil means host does not contain port + if err != nil { + h = host + } + + ip := net.ParseIP(h) + // If it's not an IP, return as is + if ip == nil { + return h + } + + // If it's an IPv4, return as is + if ip.To4() != nil { + return h + } + + // IPv6 needs bracketing + return "[" + h + "]" } func (s *SettingService) GetDefaultSettings(host string) (any, error) { diff --git a/web/service/tgbot.go b/web/service/tgbot.go index 15f5dbc2..3ff80b40 100644 --- a/web/service/tgbot.go +++ b/web/service/tgbot.go @@ -272,41 +272,78 @@ func (t *Tgbot) Start(i18nFS embed.FS) error { return nil } -// NewBot creates a new Telegram bot instance with optional proxy and API server settings. -func (t *Tgbot) NewBot(token string, proxyUrl string, apiServerUrl string) (*telego.Bot, error) { - if proxyUrl == "" && apiServerUrl == "" { - return telego.NewBot(token) +// createRobustFastHTTPClient creates a fasthttp.Client with proper connection handling +func (t *Tgbot) createRobustFastHTTPClient(proxyUrl string) *fasthttp.Client { + client := &fasthttp.Client{ + // Connection timeouts + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + MaxIdleConnDuration: 60 * time.Second, + MaxConnDuration: 0, // unlimited, but controlled by MaxIdleConnDuration + MaxIdemponentCallAttempts: 3, + ReadBufferSize: 4096, + WriteBufferSize: 4096, + MaxConnsPerHost: 100, + MaxConnWaitTimeout: 10 * time.Second, + DisableHeaderNamesNormalizing: false, + DisablePathNormalizing: false, + // Retry on connection errors + RetryIf: func(request *fasthttp.Request) bool { + // Retry on connection errors for GET requests + return string(request.Header.Method()) == "GET" || string(request.Header.Method()) == "POST" + }, + } + + // Set proxy if provided + if proxyUrl != "" { + client.Dial = fasthttpproxy.FasthttpSocksDialer(proxyUrl) } + return client +} + +// NewBot creates a new Telegram bot instance with optional proxy and API server settings. +func (t *Tgbot) NewBot(token string, proxyUrl string, apiServerUrl string) (*telego.Bot, error) { + // Validate proxy URL if provided if proxyUrl != "" { if !strings.HasPrefix(proxyUrl, "socks5://") { - logger.Warning("Invalid socks5 URL, using default") - return telego.NewBot(token) + logger.Warning("Invalid socks5 URL, ignoring proxy") + proxyUrl = "" // Clear invalid proxy + } else { + _, err := url.Parse(proxyUrl) + if err != nil { + logger.Warningf("Can't parse proxy URL, ignoring proxy: %v", err) + proxyUrl = "" + } } + } - _, err := url.Parse(proxyUrl) - if err != nil { - logger.Warningf("Can't parse proxy URL, using default instance for tgbot: %v", err) - return telego.NewBot(token) + // Validate API server URL if provided + if apiServerUrl != "" { + if !strings.HasPrefix(apiServerUrl, "http") { + logger.Warning("Invalid http(s) URL for API server, using default") + apiServerUrl = "" + } else { + _, err := url.Parse(apiServerUrl) + if err != nil { + logger.Warningf("Can't parse API server URL, using default: %v", err) + apiServerUrl = "" + } } - - return telego.NewBot(token, telego.WithFastHTTPClient(&fasthttp.Client{ - Dial: fasthttpproxy.FasthttpSocksDialer(proxyUrl), - })) } - if !strings.HasPrefix(apiServerUrl, "http") { - logger.Warning("Invalid http(s) URL, using default") - return telego.NewBot(token) - } + // Create robust fasthttp client + client := t.createRobustFastHTTPClient(proxyUrl) - _, err := url.Parse(apiServerUrl) - if err != nil { - logger.Warningf("Can't parse API server URL, using default instance for tgbot: %v", err) - return telego.NewBot(token) + // Build bot options + var options []telego.BotOption + options = append(options, telego.WithFastHTTPClient(client)) + + if apiServerUrl != "" { + options = append(options, telego.WithAPIServer(apiServerUrl)) } - return telego.NewBot(token, telego.WithAPIServer(apiServerUrl)) + return telego.NewBot(token, options...) } // IsRunning checks if the Telegram bot is currently running. @@ -390,7 +427,7 @@ func (t *Tgbot) decodeQuery(query string) (string, error) { // OnReceive starts the message receiving loop for the Telegram bot. func (t *Tgbot) OnReceive() { params := telego.GetUpdatesParams{ - Timeout: 30, // Increased timeout to reduce API calls + Timeout: 20, // Reduced timeout to detect connection issues faster } // Strict singleton: never start a second long-polling loop. tgBotMutex.Lock() @@ -408,7 +445,7 @@ func (t *Tgbot) OnReceive() { botWG.Add(1) tgBotMutex.Unlock() - // Get updates channel using the context. + // Get updates channel using the context with shorter timeout for better error recovery updates, _ := bot.UpdatesViaLongPolling(ctx, ¶ms) go func() { defer botWG.Done() @@ -2247,10 +2284,36 @@ func (t *Tgbot) SendMsgToTgbot(chatId int64, msg string, replyMarkup ...telego.R if len(replyMarkup) > 0 && n == (len(allMessages)-1) { params.ReplyMarkup = replyMarkup[0] } - _, err := bot.SendMessage(context.Background(), ¶ms) - if err != nil { - logger.Warning("Error sending telegram message :", err) + + // Retry logic with exponential backoff for connection errors + maxRetries := 3 + for attempt := range maxRetries { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + _, err := bot.SendMessage(ctx, ¶ms) + cancel() + + if err == nil { + break // Success + } + + // Check if error is a connection error + errStr := err.Error() + isConnectionError := strings.Contains(errStr, "connection") || + strings.Contains(errStr, "timeout") || + strings.Contains(errStr, "closed") + + if isConnectionError && attempt < maxRetries-1 { + // Exponential backoff: 1s, 2s, 4s + backoff := time.Duration(1<<uint(attempt)) * time.Second + logger.Warningf("Connection error sending telegram message (attempt %d/%d), retrying in %v: %v", + attempt+1, maxRetries, backoff, err) + time.Sleep(backoff) + } else { + logger.Warning("Error sending telegram message:", err) + break + } } + // Reduced delay to improve performance (only needed for rate limiting) if n < len(allMessages)-1 { // Only delay between messages, not after the last one time.Sleep(100 * time.Millisecond) |