fix(notifications): throttle failed-login alerts where they are published

Every failed login was published on the event bus, and the e-mail,
Telegram and Discord notifiers exempt login events from their cooldown
so that successful logins are never swallowed. An unauthenticated login
flood therefore filled each notifier's queue and crowded out every
other alert, a successful login, an Xray crash or a node going down.

UserLoginNotify now passes a failed attempt only for the first failure
of its source address in a minute, at most ten addresses per minute;
successful logins are always published.

internal/web/service/tgbot/tgbot_report.go

@@ -208,7 +208,40 @@ func (t *Tgbot) prepareServerUsageInfo() string {
 	return info
 }
 
-// UserLoginNotify publishes a login event to the event bus.
+const (
+	loginFailAlertWindow = 1 * time.Minute
+	// Distinct source addresses alerted per window; the address is the caller's
+	// to vary, so this ceiling keeps a broad flood below the notifier queue size.
+	loginFailAlertMaxSources = 10
+)
+
+var loginFailAlerts = &loginFailAlertGate{}
+
+// loginFailAlertGate throttles failed-login alerts before they reach the bus: the
+// notifiers exempt login events from their cooldown, so a flood crowds out every other alert.
+type loginFailAlertGate struct {
+	mu          sync.Mutex
+	windowStart time.Time
+	sources     map[string]struct{}
+}
+
+// allow passes the first failure of each source address in a window.
+func (g *loginFailAlertGate) allow(ip string, now time.Time) bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if now.Sub(g.windowStart) >= loginFailAlertWindow {
+		g.windowStart = now
+		g.sources = make(map[string]struct{}, loginFailAlertMaxSources)
+	}
+	if _, seen := g.sources[ip]; seen || len(g.sources) >= loginFailAlertMaxSources {
+		return false
+	}
+	g.sources[ip] = struct{}{}
+	return true
+}
+
+// UserLoginNotify publishes a login event to the event bus. Successful logins
+// are always published; failed ones are throttled by loginFailAlerts.
 func (t *Tgbot) UserLoginNotify(attempt LoginAttempt) {
 	if attempt.Username == "" || attempt.IP == "" || attempt.Time == "" {
 		logger.Warning("UserLoginNotify failed, invalid info!")
@@ -222,6 +255,8 @@ func (t *Tgbot) UserLoginNotify(attempt LoginAttempt) {
 	status := "fail"
 	if attempt.Status == LoginSuccess {
 		status = "success"
+	} else if !loginFailAlerts.allow(attempt.IP, time.Now()) {
+		return
 	}
 
 	EventBus.Publish(eventbus.Event{

internal/web/service/tgbot/tgbot_reports_usage_test.go

@@ -2,10 +2,13 @@ package tgbot
 
 import (
 	"encoding/json"
+	"fmt"
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/mhsanaei/3x-ui/v3/internal/database"
 	"github.com/mhsanaei/3x-ui/v3/internal/database/dbtest"
@@ -365,3 +368,72 @@ func TestLoginAlertKeepsUsernameOnOneLine(t *testing.T) {
 		}
 	}
 }
+
+// The notifiers exempt login events from their cooldown, so failed logins are throttled where
+// they are published, or an unauthenticated flood crowds every other alert out of the queues.
+func TestLoginFailAlertGate(t *testing.T) {
+	g := &loginFailAlertGate{}
+	now := time.Now()
+
+	if !g.allow("192.0.2.1", now) {
+		t.Fatal("first failure of a source must be alerted")
+	}
+	if g.allow("192.0.2.1", now.Add(loginFailAlertWindow-time.Second)) {
+		t.Fatal("a source must be alerted once per window")
+	}
+
+	passed := 1
+	for i := range 1000 {
+		if g.allow(fmt.Sprintf("2001:db8::%x", i), now.Add(time.Second)) {
+			passed++
+		}
+	}
+	if passed != loginFailAlertMaxSources {
+		t.Fatalf("%d sources alerted in one window, want %d", passed, loginFailAlertMaxSources)
+	}
+	if len(g.sources) > loginFailAlertMaxSources {
+		t.Fatalf("gate tracks %d sources, want at most %d", len(g.sources), loginFailAlertMaxSources)
+	}
+
+	if !g.allow("192.0.2.1", now.Add(loginFailAlertWindow)) {
+		t.Fatal("a new window must alert again")
+	}
+}
+
+func TestUserLoginNotifyThrottlesOnlyFailures(t *testing.T) {
+	bus := eventbus.New(0)
+	defer bus.Stop()
+	prevBus, prevGate := EventBus, loginFailAlerts
+	EventBus, loginFailAlerts = bus, &loginFailAlertGate{}
+	defer func() { EventBus, loginFailAlerts = prevBus, prevGate }()
+
+	var failures, successes atomic.Int32
+	bus.Subscribe("counter", func(e eventbus.Event) {
+		if e.Data.(*eventbus.LoginEventData).Status == "success" {
+			successes.Add(1)
+		} else {
+			failures.Add(1)
+		}
+	})
+
+	bot := &Tgbot{}
+	for range 50 {
+		bot.UserLoginNotify(LoginAttempt{Username: "admin", IP: "192.0.2.1", Time: "now", Status: LoginFail, Reason: "too many failed attempts"})
+	}
+	bot.UserLoginNotify(LoginAttempt{Username: "admin", IP: "192.0.2.2", Time: "now", Status: LoginFail, Reason: "invalid credentials"})
+	for range 3 {
+		bot.UserLoginNotify(LoginAttempt{Username: "admin", IP: "192.0.2.1", Time: "now", Status: LoginSuccess})
+	}
+
+	// The bus delivers in publication order, so the last success arrives after every failure.
+	deadline := time.Now().Add(5 * time.Second)
+	for successes.Load() < 3 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if got := successes.Load(); got != 3 {
+		t.Fatalf("successful logins reported %d times, want 3", got)
+	}
+	if got := failures.Load(); got != 2 {
+		t.Fatalf("failed logins from two sources reported %d times, want 2", got)
+	}
+}