pangolin/server/internal/alert/notifier.go

// Package alert implements the unified alert exit channel for Pangolin's
// scheduler subsystem (task 15G).
//
// # Event types
//
// Seven fixed event types cover all failure modes detected by 15D (detect),
// 15E (orchestrate), and 15F (probe):
//
//   EventTypeBlockConfirmed   — node confirmed GFW-censored after ≥6 cycles
//   EventTypeReplenishFailed  — replacement provisioning exhausted all retries
//   EventTypeWatermarkLow     — pool node-weight drops below 70 % capacity
//   EventTypeBreakerTripped   — circuit breaker blocked a replacement attempt
//   EventTypeProbeAgentLost   — third-party synthetic probe failing ≥3 times
//   EventTypeHeartbeatMissing — first-party probe silent for >90 s
//   EventTypeFault            — node-level outage (domestic + overseas both fail)
//
// # Deduplication
//
// Non-critical events are throttled: the same (Type, NodeID) pair fires at
// most once per 10-minute window (Redis SETNX + TTL).  Critical events always
// fire — they are never suppressed.
//
// # Alert channel safety
//
// Notify must never return a blocking error.  TG API failures are retried up
// to maxRetries times, then fall back to the slog-based LogNotifier.  The
// calling goroutine (scheduler tick) is never stalled by alert-channel faults.
//
// # Identity isolation
//
// The TG bot token and chat ID are injected via environment variables only.
// The bot belongs to a dedicated anonymous operator account; see
// infra/identity-isolation.md and red line 06 §2.
//
// # Privacy
//
// Event messages use internal node IDs (opaque), never public domain names.
// This prevents domain leakage if the alert message is forwarded.
package alert

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"
)

// ─────────────────────────────────────────────────────────────────────────────
// Event types
// ─────────────────────────────────────────────────────────────────────────────

// EventType identifies the class of alert event.
type EventType string

const (
	// EventTypeBlockConfirmed fires when detect (15D) transitions a node from
	// blocked_suspect to blocked_confirmed, confirming GFW censorship after
	// ≥6 consecutive failing probe cycles.
	// Runbook anchor: #block-confirmed
	EventTypeBlockConfirmed EventType = "block_confirmed"

	// EventTypeReplenishFailed fires when orchestrate (15E) exhausts all
	// MaxAttempts replacement attempts for a blocked node.
	// Runbook anchor: #replenish-failed
	EventTypeReplenishFailed EventType = "replenish_failed"

	// EventTypeWatermarkLow fires when the effective capacity of a node pool
	// (combined routing weight of live nodes) drops below 70 % of full capacity.
	// Runbook anchor: #watermark-low
	EventTypeWatermarkLow EventType = "watermark_low"

	// EventTypeBreakerTripped fires when the circuit breaker (15F) blocks a
	// replacement attempt due to a burst of confirmed blocks in the same pool.
	// Runbook anchor: #breaker-tripped
	EventTypeBreakerTripped EventType = "breaker_tripped"

	// EventTypeProbeAgentLost fires when the third-party (Aliyun) synthetic
	// probe agent encounters ≥3 consecutive API failures.
	// Runbook anchor: #probe-agent-lost
	EventTypeProbeAgentLost EventType = "probe_agent_lost"

	// EventTypeHeartbeatMissing fires when a first-party probe agent has not
	// delivered any heartbeat for more than 90 seconds.
	// Runbook anchor: #heartbeat-missing
	EventTypeHeartbeatMissing EventType = "heartbeat_missing"

	// EventTypeFault fires when both domestic AND overseas probes are failing
	// for a node, indicating a node-level outage rather than GFW censorship.
	// Runbook anchor: #node-fault
	EventTypeFault EventType = "fault"
)

// Severity indicates the urgency of an event.
type Severity string

const (
	SeverityCritical Severity = "critical" // requires immediate human action
	SeverityWarning  Severity = "warning"  // review within the hour
	SeverityInfo     Severity = "info"     // informational only
)

// typeMeta bundles display properties for one event type.
type typeMeta struct {
	Severity      Severity
	TitleZH       string // human-readable Chinese label
	Emoji         string // prefixed to the TG message
	RunbookAnchor string // HTML id anchor in docs/runbook-scheduler.md
}

// typeMetaMap is indexed by EventType.
var typeMetaMap = map[EventType]typeMeta{
	EventTypeBlockConfirmed:   {SeverityWarning, "判封确认", "⚠️", "#block-confirmed"},
	EventTypeReplenishFailed:  {SeverityCritical, "补新连续失败≥3", "🔴", "#replenish-failed"},
	EventTypeWatermarkLow:     {SeverityWarning, "水位<70%", "⚠️", "#watermark-low"},
	EventTypeBreakerTripped:   {SeverityCritical, "熔断触发", "🔴", "#breaker-tripped"},
	EventTypeProbeAgentLost:   {SeverityWarning, "探针失联", "⚠️", "#probe-agent-lost"},
	EventTypeHeartbeatMissing: {SeverityWarning, "心跳缺失>90s", "⚠️", "#heartbeat-missing"},
	EventTypeFault:            {SeverityCritical, "故障态", "🔴", "#node-fault"},
}

// ─────────────────────────────────────────────────────────────────────────────
// Event
// ─────────────────────────────────────────────────────────────────────────────

// Event carries all data required to render one alert notification.
type Event struct {
	// Type is one of the seven EventType constants.
	Type EventType

	// Severity controls dedup and display urgency.
	// Derived from Type via NewEvent; may be overridden by the caller.
	Severity Severity

	// NodeID is the internal opaque node or probe agent identifier.
	// MUST be an internal ID — never a public domain name (privacy).
	NodeID string

	// Pool optionally identifies the operational pool (e.g. "free/hkg").
	Pool string

	// Detail holds supplementary key-value data for the notification.
	// Values must not contain personal user data.
	Detail map[string]string

	// RunbookAnchor is the #fragment anchor in docs/runbook-scheduler.md.
	// Overrides the type default when non-empty.
	RunbookAnchor string
}

// NewEvent constructs an Event with type-derived Severity and RunbookAnchor.
// Callers may override any field after construction.
func NewEvent(t EventType, nodeID string, detail map[string]string) Event {
	m := typeMetaMap[t]
	return Event{
		Type:          t,
		Severity:      m.Severity,
		NodeID:        nodeID,
		Detail:        detail,
		RunbookAnchor: m.RunbookAnchor,
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Notifier interface
// ─────────────────────────────────────────────────────────────────────────────

// Notifier is the unified alert outlet for scheduler events.
// Implementations must be safe for concurrent use and must never return a
// blocking error — alert-channel failures must not halt the scheduler.
type Notifier interface {
	Notify(ctx context.Context, event Event) error
}

// ─────────────────────────────────────────────────────────────────────────────
// LogNotifier
// ─────────────────────────────────────────────────────────────────────────────

// LogNotifier writes every event to the structured logger.
// It is used when TG_BOT_TOKEN is not configured, or as the fallback when the
// TGNotifier exhausts its retries.
type LogNotifier struct{}

// Notify implements Notifier.
func (LogNotifier) Notify(_ context.Context, e Event) error {
	m := typeMetaMap[e.Type]
	slog.Warn("alert: scheduler event",
		"type", string(e.Type),
		"title", m.TitleZH,
		"severity", string(e.Severity),
		"node_id", e.NodeID,
		"pool", e.Pool,
		"detail", e.Detail,
		"runbook", e.RunbookAnchor,
	)
	return nil
}

// ─────────────────────────────────────────────────────────────────────────────
// TGNotifier
// ─────────────────────────────────────────────────────────────────────────────

const (
	// dedupTTL is the dedup suppression window for non-critical events.
	dedupTTL = 10 * time.Minute

	// dedupKeyPrefix is the Redis key namespace for dedup tokens.
	// Full key: alert:dedup:{type}:{nodeID}
	dedupKeyPrefix = "alert:dedup:"

	// maxRetries is the number of additional TG send attempts after the first
	// failure.  Exhausting retries falls back to LogNotifier silently.
	maxRetries = 2

	// tgSendMessageURL is the Telegram Bot API pattern for sendMessage.
	tgSendMessageURL = "https://api.telegram.org/bot%s/sendMessage"
)

// TGConfig holds the runtime configuration for TGNotifier.
// All sensitive values come from environment variables; see cmd/ bootstrap.
type TGConfig struct {
	// BotToken is the Telegram bot token (env: TG_BOT_TOKEN).
	// The bot must belong to a dedicated anonymous operator account (red line §2).
	BotToken string

	// ChatID is the target Telegram group or channel ID (env: TG_ALERT_CHAT_ID).
	ChatID string

	// RunbookBaseURL is prepended to the anchor to form the full runbook URL.
	// Leave empty to embed only the #anchor fragment.
	RunbookBaseURL string

	// HTTPTimeout overrides the per-call HTTP timeout (default 10 s).
	HTTPTimeout time.Duration

	// BaseURL overrides the TG API base URL for testing.
	// Leave empty in production.
	BaseURL string
}

// TGNotifier sends events to a Telegram group via the Bot API.
//
// Deduplication:
//   - Non-critical events: one message per (Type, NodeID) per 10-minute window
//     (Redis SETNX + TTL).  The second identical message within the window is
//     silently dropped.
//   - Critical events: always delivered, no dedup.
//
// Retry + fallback:
//   - On TG API failure, retries up to maxRetries (2) times before delegating
//     to the fallback Notifier.  Notify always returns nil.
type TGNotifier struct {
	cfg        TGConfig
	rdb        *redis.Client
	httpClient *http.Client
	fallback   Notifier
	apiURL     string // base send URL, pre-formatted with token
}

// NewTGNotifier constructs a TGNotifier.
// rdb is used for the dedup SETNX gate.
// fallback is used when TG API calls are exhausted; nil defaults to LogNotifier.
func NewTGNotifier(cfg TGConfig, rdb *redis.Client, fallback Notifier) *TGNotifier {
	if fallback == nil {
		fallback = LogNotifier{}
	}
	httpTimeout := 10 * time.Second
	if cfg.HTTPTimeout > 0 {
		httpTimeout = cfg.HTTPTimeout
	}
	baseURL := fmt.Sprintf(tgSendMessageURL, cfg.BotToken)
	if cfg.BaseURL != "" {
		// Test hook: override the entire API base URL (token appended separately).
		baseURL = cfg.BaseURL
	}
	return &TGNotifier{
		cfg:        cfg,
		rdb:        rdb,
		httpClient: &http.Client{Timeout: httpTimeout},
		fallback:   fallback,
		apiURL:     baseURL,
	}
}

// Notify implements Notifier.
func (n *TGNotifier) Notify(ctx context.Context, e Event) error {
	// Critical events bypass the dedup gate entirely.
	if e.Severity != SeverityCritical {
		allowed, err := n.dedupGate(ctx, e)
		if err != nil {
			// Redis error → allow through (fail-open) to avoid losing alerts.
			slog.Warn("alert: dedup Redis error; allowing through", "error", err)
		}
		if !allowed {
			return nil // duplicate within the 10-min window; silently dropped
		}
	}

	text := n.renderMessage(e)

	var lastErr error
	for attempt := 0; attempt <= maxRetries; attempt++ {
		if err := n.callTG(ctx, text); err != nil {
			lastErr = err
			slog.Warn("alert: TG send failed",
				"attempt", attempt+1, "of", maxRetries+1,
				"type", string(e.Type), "node_id", e.NodeID,
				"error", err,
			)
			continue
		}
		return nil // delivered
	}

	// Retries exhausted: degrade to log.  Never return an error.
	slog.Error("alert: TG delivery exhausted retries; falling back to log",
		"type", string(e.Type), "node_id", e.NodeID, "last_error", lastErr)
	_ = n.fallback.Notify(ctx, e)
	return nil
}

// dedupGate returns (true, nil) if the event should be sent (key newly set),
// or (false, nil) if it is a duplicate within the 10-min window.
func (n *TGNotifier) dedupGate(ctx context.Context, e Event) (bool, error) {
	key := dedupKeyPrefix + string(e.Type) + ":" + e.NodeID
	set, err := n.rdb.SetNX(ctx, key, "1", dedupTTL).Result()
	if err != nil {
		return true, fmt.Errorf("alert: dedup SETNX: %w", err)
	}
	return set, nil
}

// renderMessage builds the Telegram HTML message body.
// Node IDs are used as-is (internal opaque IDs, not domain names).
func (n *TGNotifier) renderMessage(e Event) string {
	m := typeMetaMap[e.Type]

	var sb strings.Builder

	// Line 1: emoji + bold event title + node id
	sb.WriteString(m.Emoji)
	sb.WriteString(" <b>")
	sb.WriteString(htmlEscape(m.TitleZH))
	sb.WriteString("</b> — 节点 <code>")
	sb.WriteString(htmlEscape(e.NodeID))
	sb.WriteString("</code>")
	if e.Pool != "" {
		sb.WriteString(" 池 <code>")
		sb.WriteString(htmlEscape(e.Pool))
		sb.WriteString("</code>")
	}
	sb.WriteString("\n")

	// Line 2: severity label
	sb.WriteString("严重度: ")
	sb.WriteString(string(e.Severity))
	sb.WriteString("\n")

	// Detail pairs (sorted-ish; map iteration is random, determinism not required)
	for k, v := range e.Detail {
		sb.WriteString(htmlEscape(k))
		sb.WriteString(": ")
		sb.WriteString(htmlEscape(v))
		sb.WriteString("\n")
	}

	// Runbook link
	anchor := e.RunbookAnchor
	if anchor == "" {
		anchor = m.RunbookAnchor
	}
	link := anchor
	if n.cfg.RunbookBaseURL != "" {
		link = n.cfg.RunbookBaseURL + anchor
	}
	sb.WriteString("📖 <a href=\"")
	sb.WriteString(link)
	sb.WriteString("\">处置手册</a>")

	return sb.String()
}

// callTG posts one sendMessage request to the Telegram Bot API.
func (n *TGNotifier) callTG(ctx context.Context, text string) error {
	payload, err := json.Marshal(map[string]string{
		"chat_id":              n.cfg.ChatID,
		"text":                 text,
		"parse_mode":           "HTML",
		"disable_web_page_preview": "true",
	})
	if err != nil {
		return fmt.Errorf("alert: marshal TG payload: %w", err)
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.apiURL, bytes.NewReader(payload))
	if err != nil {
		return fmt.Errorf("alert: build TG request: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")

	resp, err := n.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("alert: TG http: %w", err)
	}
	defer resp.Body.Close()
	io.Copy(io.Discard, resp.Body) //nolint:errcheck // response body drained for keep-alive

	if resp.StatusCode >= 500 {
		return fmt.Errorf("alert: TG server error HTTP %d", resp.StatusCode)
	}
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("alert: TG unexpected HTTP %d", resp.StatusCode)
	}
	return nil
}

// htmlEscape escapes the five HTML special characters relevant to Telegram HTML mode.
func htmlEscape(s string) string {
	s = strings.ReplaceAll(s, "&", "&amp;")
	s = strings.ReplaceAll(s, "<", "&lt;")
	s = strings.ReplaceAll(s, ">", "&gt;")
	s = strings.ReplaceAll(s, "\"", "&quot;")
	return s
}