pangolin/server/internal/scheduler/detect/engine.go

package detect

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

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

	"github.com/wangjia/pangolin/server/internal/idgen"
)

// ─────────────────────────────────────────────────────────────────────────────
// Notifier — 15G interface stub
// ─────────────────────────────────────────────────────────────────────────────

// Notifier is the 15G event sink for fault notifications.
// The real implementation (15G) sends a Telegram/alerting message; the stub
// below writes to the structured logger and is used until 15G is ready.
type Notifier interface {
	NotifyFault(ctx context.Context, nodeID, reason string) error
}

// LogNotifier is a Notifier stub that logs via slog.
// It is used when no real Notifier is wired up.
type LogNotifier struct{}

// NotifyFault implements Notifier.
func (LogNotifier) NotifyFault(_ context.Context, nodeID, reason string) error {
	slog.Warn("node fault detected — manual review required",
		"node_id", nodeID,
		"reason", reason,
	)
	return nil
}

// ─────────────────────────────────────────────────────────────────────────────
// Redis key constants
// ─────────────────────────────────────────────────────────────────────────────

const (
	// replaceQueueKey is the Redis list that 15E (Replenisher) drains to
	// provision replacement nodes.  The queue entry shape is replaceRequest JSON.
	replaceQueueKey = "detect:replace:queue"

	// probeFreqKeyPrefix is the per-node Redis key prefix written when a node
	// enters "blocked_suspect".  The value "60" (seconds) tells 15B to probe
	// at 1-minute instead of the default 5-minute interval.
	probeFreqKeyPrefix = "probe:freq:"

	// probeFreqTTL keeps the elevated-frequency marker alive for 45 minutes —
	// long enough to cover a full suspect → confirmed window (6 × 5 min + buffer).
	probeFreqTTL = 45 * time.Minute
)

// ─────────────────────────────────────────────────────────────────────────────
// Engine
// ─────────────────────────────────────────────────────────────────────────────

// Engine executes one detection cycle per Tick call.
//
// 15H (DetectLoop) is responsible for:
//   - scheduling Ticks at 5-minute intervals, and
//   - leader election (only the elected leader should call Tick).
//
// Engine itself holds no goroutines and no in-process state beyond its
// constructor arguments; all persistent state lives in Redis (via StreakStore).
type Engine struct {
	probeStore ProbeSnapshotter
	lc         LifecycleService
	streaks    *StreakStore
	rdb        *redis.Client
	notifier   Notifier
	cfg        *DetectConfig
}

// NewEngine creates an Engine.  If cfg is nil, DefaultConfig() is used.
// If notifier is nil, LogNotifier is used.
func NewEngine(
	probeStore ProbeSnapshotter,
	lc LifecycleService,
	streaks *StreakStore,
	rdb *redis.Client,
	notifier Notifier,
	cfg *DetectConfig,
) *Engine {
	if cfg == nil {
		d := DefaultConfig()
		cfg = &d
	}
	if notifier == nil {
		notifier = LogNotifier{}
	}
	return &Engine{
		probeStore: probeStore,
		lc:         lc,
		streaks:    streaks,
		rdb:        rdb,
		notifier:   notifier,
		cfg:        cfg,
	}
}

// Tick executes one full detection cycle.
//
// It lists nodes in "up" and "blocked_suspect" states, reads their probe
// snapshots, computes signals, and applies the five classification rules.
// Errors from individual nodes are logged but do not halt the cycle for other
// nodes.
func (e *Engine) Tick(ctx context.Context) error {
	nodes, err := e.lc.ListNodes(ctx, NodeFilter{
		Statuses: []NodeStatus{StatusUp, StatusBlockedSuspect},
	})
	if err != nil {
		return fmt.Errorf("detect: list nodes: %w", err)
	}

	for _, node := range nodes {
		if err := e.processNode(ctx, node); err != nil {
			slog.Error("detect: process node",
				"node_id", node.ID,
				"status", node.Status,
				"error", err,
			)
			// Continue — one node failure must not halt the full cycle.
		}
	}
	return nil
}

// processNode applies all five rules to a single node.
func (e *Engine) processNode(ctx context.Context, node NodeInfo) error {
	snapshots, err := e.probeStore.SnapshotsByNode(ctx, node.ID)
	if err != nil {
		return fmt.Errorf("snapshots: %w", err)
	}

	sig := computeSignals(ctx, node.ID, snapshots, e.lc)

	sk, err := e.streaks.Load(ctx, node.ID)
	if err != nil {
		return fmt.Errorf("streak load: %w", err)
	}

	// ── Rule 5: fault ─────────────────────────────────────────────────────────
	// Both domestic AND overseas probes failing → node-level outage, not GFW.
	// Do NOT transition status; send an alert event for human review.
	// Streaks are left unchanged so that when the node recovers the engine
	// resumes from its current position rather than re-triggering immediately.
	if isFault(sig) {
		reason := fmt.Sprintf("domestic_fail_isps=%d overseas_ok=false", sig.DomesticFailISPs)
		if notifyErr := e.notifier.NotifyFault(ctx, node.ID, reason); notifyErr != nil {
			slog.Error("detect: notify fault", "node_id", node.ID, "error", notifyErr)
		}
		return nil // do not persist streak changes
	}

	switch node.Status {
	case StatusUp:
		return e.processUp(ctx, node, sig, sk)
	case StatusBlockedSuspect:
		return e.processSuspect(ctx, node, sig, sk)
	default:
		return nil
	}
}

// processUp applies the suspect and traffic-warning rules to an "up" node.
func (e *Engine) processUp(ctx context.Context, node NodeInfo, sig NodeSignal, sk Streak) error {
	domesticFailing := e.cfg.isSuspectTriggered(sig)
	overseasBarrier := sig.OverseasHasData && sig.OverseasOK

	if domesticFailing && overseasBarrier {
		// Domestic probes failing, overseas OK → potential GFW block.
		threshold := e.cfg.effectiveSuspectStreakMin(sig)

		if sk.FailStreak+1 >= threshold {
			// ── Rule 1: suspect ───────────────────────────────────────────────
			detail := map[string]any{
				"from":               "up",
				"to":                 "blocked_suspect",
				"domestic_fail_isps": sig.DomesticFailISPs,
				"domestic_total":     sig.DomesticTotalISPs,
				"fail_streak":        sk.FailStreak + 1,
			}
			affected, err := e.lc.TransitionStatus(ctx, node.ID, StatusUp, StatusBlockedSuspect, detail)
			if err != nil {
				return fmt.Errorf("transition up→suspect: %w", err)
			}
			if affected == 0 {
				// Optimistic lock conflict: another writer changed the state
				// concurrently.  Leave streak unchanged; the engine is idempotent.
				slog.Info("detect: suspect transition skipped (lock conflict)", "node_id", node.ID)
				return e.streaks.Save(ctx, node.ID, sk)
			}

			// Transition succeeded: apply side-effects.
			if err := e.lc.SetWeight(ctx, node.ID, e.cfg.SuspectWeight); err != nil {
				slog.Error("detect: set suspect weight", "node_id", node.ID, "error", err)
			}
			if err := e.writeProbeFreqMark(ctx, node.ID); err != nil {
				slog.Error("detect: write probe freq mark", "node_id", node.ID, "error", err)
			}
			if err := e.lc.BumpVersion(ctx); err != nil {
				slog.Error("detect: bump version after suspect", "node_id", node.ID, "error", err)
			}

			// Reset fail streak; suspect streak begins on the next call to
			// processSuspect.
			sk.FailStreak = 0
			sk.SuspectStreak = 0
			sk.RecoverStreak = 0
		} else {
			// Below the threshold: increment and wait for next cycle.
			sk.FailStreak++
			sk.RecoverStreak = 0
		}
	} else {
		// Domestic probes OK (or overseas data absent / also failing — fault is
		// handled above before this function is called).
		sk.FailStreak = 0
		sk.RecoverStreak = 0
	}

	return e.streaks.Save(ctx, node.ID, sk)
}

// processSuspect applies the confirmed and recover rules to a "blocked_suspect" node.
func (e *Engine) processSuspect(ctx context.Context, node NodeInfo, sig NodeSignal, sk Streak) error {
	domesticFailing := e.cfg.isSuspectTriggered(sig)
	overseasBarrier := sig.OverseasHasData && sig.OverseasOK

	if domesticFailing && overseasBarrier {
		// Still looks like a GFW block: advance the suspect streak.
		sk.SuspectStreak++
		sk.RecoverStreak = 0
		sk.FailStreak = 0

		if sk.SuspectStreak >= e.cfg.ConfirmedStreakMin {
			// ── Rule 3: confirmed ─────────────────────────────────────────────
			confirmedDetail := map[string]any{
				"from":           "blocked_suspect",
				"to":             "blocked_confirmed",
				"suspect_streak": sk.SuspectStreak,
			}
			n1, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedSuspect, StatusBlockedConfirmed, confirmedDetail)
			if err != nil {
				return fmt.Errorf("transition suspect→confirmed: %w", err)
			}
			if n1 == 0 {
				slog.Info("detect: confirmed transition skipped (lock conflict)", "node_id", node.ID)
				return e.streaks.Save(ctx, node.ID, sk)
			}

			// Immediately mark down — skip draining per lifecycle policy.
			downDetail := map[string]any{
				"from": "blocked_confirmed",
				"to":   "down",
				"note": "skipped draining (confirmed block policy)",
			}
			if _, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedConfirmed, StatusDown, downDetail); err != nil {
				return fmt.Errorf("transition confirmed→down: %w", err)
			}

			// Push to 15E replace queue.
			if err := e.pushReplaceQueue(ctx, node.ID); err != nil {
				slog.Error("detect: push replace queue", "node_id", node.ID, "error", err)
			}

			// Reset all streaks; the node is now down and will not be processed
			// again (ListNodes only returns "up" and "blocked_suspect").
			sk = Streak{}
		}
	} else {
		// Domestic probes passing (or overseas data absent — treat as recovery).
		// ── Rule 4: recover ───────────────────────────────────────────────────
		sk.RecoverStreak++
		sk.SuspectStreak = 0
		sk.FailStreak = 0

		if sk.RecoverStreak >= e.cfg.RecoverStreakMin {
			recoverDetail := map[string]any{
				"from":           "blocked_suspect",
				"to":             "up",
				"recover_streak": sk.RecoverStreak,
			}
			affected, err := e.lc.TransitionStatus(ctx, node.ID, StatusBlockedSuspect, StatusUp, recoverDetail)
			if err != nil {
				return fmt.Errorf("transition suspect→up: %w", err)
			}
			if affected == 0 {
				slog.Info("detect: recover transition skipped (lock conflict)", "node_id", node.ID)
				return e.streaks.Save(ctx, node.ID, sk)
			}
			// Weight recovery is delegated to 15E's warmup/gradual-ramp channel.
			// The engine intentionally does NOT call SetWeight here.
			sk = Streak{}
		}
	}

	return e.streaks.Save(ctx, node.ID, sk)
}

// ─────────────────────────────────────────────────────────────────────────────
// Helpers
// ─────────────────────────────────────────────────────────────────────────────

// writeProbeFreqMark sets a Redis key that tells 15B to probe this node at
// 1-minute instead of the default 5-minute interval while it is suspect.
func (e *Engine) writeProbeFreqMark(ctx context.Context, nodeID string) error {
	key := probeFreqKeyPrefix + nodeID
	return e.rdb.Set(ctx, key, "60", probeFreqTTL).Err()
}

// replaceRequest is the payload pushed to the 15E replace queue.
type replaceRequest struct {
	NodeID          string `json:"nodeId"`
	ReplacementUUID string `json:"replacementUuid"`
}

// pushReplaceQueue enqueues a replacement request for 15E (Replenisher).
// The replacement UUID is pre-allocated here so 15E can reference it without
// generating its own ID.
func (e *Engine) pushReplaceQueue(ctx context.Context, nodeID string) error {
	payload, err := json.Marshal(replaceRequest{
		NodeID:          nodeID,
		ReplacementUUID: idgen.NewString(),
	})
	if err != nil {
		return fmt.Errorf("detect: marshal replace request: %w", err)
	}
	return e.rdb.LPush(ctx, replaceQueueKey, payload).Err()
}