pangolin/server/internal/scheduler/orchestrate/replacer.go

package orchestrate

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

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

	"github.com/wangjia/pangolin/server/internal/alert"
	"github.com/wangjia/pangolin/server/internal/scheduler/probe"
)

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

const (
	// replaceQueueKey is the Redis list populated by 15D when a node is confirmed
	// blocked. We read from the right (RPOP) for FIFO ordering; 15D writes with LPUSH.
	replaceQueueKey = "detect:replace:queue"

	// replaceKeyPrefix is the namespace for per-replacement orchestration records.
	// Full key: sched:replace:{uuid}
	replaceKeyPrefix = "sched:replace:"

	// replaceIndexKey is a Redis set holding all in-flight replacement UUIDs.
	replaceIndexKey = "sched:replace:index"

	// replaceTTL is how long a terminal (done / failed) record is kept for auditing.
	replaceTTL = 7 * 24 * time.Hour
)

// ─────────────────────────────────────────────────────────────────────────────
// Persistent record types
// ─────────────────────────────────────────────────────────────────────────────

// ReplaceRecord is the crash-recoverable orchestration state stored in Redis.
// Key: sched:replace:{uuid}  Value: JSON-encoded ReplaceRecord.
//
// All phase transitions read the record first and write it last; a crashed
// process resumes from the persisted state without repeating already-executed
// actions (CreateNode is guarded by an idempotency key; DestroyNode is
// idempotent by design).
type ReplaceRecord struct {
	Phase             Phase     `json:"phase"`
	OldNode           string    `json:"oldNode"`                    // node ID from the 15D replace queue
	NewNode           string    `json:"newNode,omitempty"`           // new node ID, set after CreateNode
	CurrentProviderID string    `json:"currentProviderId,omitempty"` // provider for the current attempt
	Attempts          int       `json:"attempts"`                    // how many create+probe attempts so far
	ProviderTried     []string  `json:"providerTried"`               // provider IDs from failed attempts
	ProbeStreak       int       `json:"probeStreak"`                 // consecutive passing probe Ticks
	PhaseStartedAt    time.Time `json:"phaseStartedAt"`             // when the current phase began
	UpdatedAt         time.Time `json:"updatedAt"`
}

// queueEntry is the JSON shape pushed by 15D (detect/engine.go pushReplaceQueue).
type queueEntry struct {
	NodeID          string `json:"nodeId"`
	ReplacementUUID string `json:"replacementUuid"`
}

// ─────────────────────────────────────────────────────────────────────────────
// Replacer
// ─────────────────────────────────────────────────────────────────────────────

// Replacer drives the replacement orchestration state machine.
// It is called every 30 s by the OrchestrateLoop (task 15H).
type Replacer struct {
	rdb      *redis.Client
	prov     ProvisionService
	lc       LifecycleService
	snaps    ProbeSnapshotter
	breaker  Breaker
	notifier Notifier
	clock    Clock
}

// Config holds all dependencies for NewReplacer.
type Config struct {
	RDB      *redis.Client
	Prov     ProvisionService
	LC       LifecycleService
	Snaps    ProbeSnapshotter
	Breaker  Breaker
	Notifier Notifier
	Clock    Clock
}

// NewReplacer constructs a Replacer. Nil optional deps are replaced with stubs.
func NewReplacer(cfg Config) *Replacer {
	if cfg.Breaker == nil {
		cfg.Breaker = StubBreaker{}
	}
	if cfg.Notifier == nil {
		cfg.Notifier = alert.LogNotifier{}
	}
	if cfg.Clock == nil {
		cfg.Clock = RealClock{}
	}
	return &Replacer{
		rdb:      cfg.RDB,
		prov:     cfg.Prov,
		lc:       cfg.LC,
		snaps:    cfg.Snaps,
		breaker:  cfg.Breaker,
		notifier: cfg.Notifier,
		clock:    cfg.Clock,
	}
}

// Tick drains the 15D replace queue and advances every in-flight record one
// step. It is designed to be called every 30 s and to be idempotent across
// crashes: all state is persisted in Redis before any external action, so a
// restart resumes from the saved phase.
func (r *Replacer) Tick(ctx context.Context) error {
	if err := r.drainQueue(ctx); err != nil {
		// Non-fatal: log and continue so in-flight records still advance.
		slog.Error("orchestrate: drain replace queue", "error", err)
	}
	return r.advanceAll(ctx)
}

// drainQueue pops entries from detect:replace:queue and creates pending records.
func (r *Replacer) drainQueue(ctx context.Context) error {
	const maxDrain = 100 // guard against burst
	for i := 0; i < maxDrain; i++ {
		raw, err := r.rdb.RPop(ctx, replaceQueueKey).Result()
		if err == redis.Nil {
			break // queue empty
		}
		if err != nil {
			return fmt.Errorf("orchestrate: rpop: %w", err)
		}
		var entry queueEntry
		if err := json.Unmarshal([]byte(raw), &entry); err != nil {
			slog.Error("orchestrate: bad queue entry", "raw", raw, "error", err)
			continue
		}
		if err := r.ensureRecord(ctx, entry.ReplacementUUID, entry.NodeID); err != nil {
			slog.Error("orchestrate: ensure record",
				"uuid", entry.ReplacementUUID,
				"node", entry.NodeID,
				"error", err,
			)
		}
	}
	return nil
}

// ensureRecord creates a pending orchestration record for uuid/oldNode if one
// does not already exist, then adds uuid to the in-flight index. It is safe
// to call multiple times (SetNX + SAdd are both idempotent).
func (r *Replacer) ensureRecord(ctx context.Context, uuid, oldNode string) error {
	key := replaceKeyPrefix + uuid
	rec := ReplaceRecord{
		Phase:          PhasePending,
		OldNode:        oldNode,
		ProviderTried:  []string{},
		PhaseStartedAt: r.clock.Now(),
		UpdatedAt:      r.clock.Now(),
	}
	data, err := json.Marshal(rec)
	if err != nil {
		return err
	}
	// SetNX: only stores if the key does not exist, preserving any in-progress record.
	r.rdb.SetNX(ctx, key, data, 0) //nolint:errcheck // best-effort; SAdd follows
	return r.rdb.SAdd(ctx, replaceIndexKey, uuid).Err()
}

// advanceAll loads every UUID from the in-flight index and advances each record.
func (r *Replacer) advanceAll(ctx context.Context) error {
	uuids, err := r.rdb.SMembers(ctx, replaceIndexKey).Result()
	if err != nil {
		return fmt.Errorf("orchestrate: smembers index: %w", err)
	}
	for _, uuid := range uuids {
		if advErr := r.advanceRecord(ctx, uuid); advErr != nil {
			slog.Error("orchestrate: advance record", "uuid", uuid, "error", advErr)
			// continue; one failure must not halt other replacements
		}
	}
	return nil
}

// advanceRecord loads a single record and advances it by one phase step.
func (r *Replacer) advanceRecord(ctx context.Context, uuid string) error {
	rec, err := r.loadRecord(ctx, uuid)
	if err != nil {
		return err
	}
	if rec == nil {
		// Key expired or was deleted; clean up the index entry.
		_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
		return nil
	}

	switch rec.Phase {
	case PhasePending:
		return r.stepPending(ctx, uuid, rec)
	case PhaseCreating:
		return r.stepCreating(ctx, uuid, rec)
	case PhaseProbing:
		return r.stepProbing(ctx, uuid, rec)
	case PhaseActivating:
		return r.stepActivating(ctx, uuid, rec)
	case PhaseDrainingOld:
		return r.stepDrainingOld(ctx, uuid, rec)
	case PhaseDone, PhaseFailed:
		// Terminal: the record exists only for audit; remove from active index.
		_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
		return nil
	default:
		return fmt.Errorf("orchestrate: unknown phase %q for %s", rec.Phase, uuid)
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Phase step functions
// ─────────────────────────────────────────────────────────────────────────────

// stepPending checks the circuit breaker and watermark/quota, then advances
// to creating. If the breaker blocks, the record stays pending (retried next Tick).
func (r *Replacer) stepPending(ctx context.Context, uuid string, rec *ReplaceRecord) error {
	nodeInfo, err := r.lc.GetNode(ctx, rec.OldNode)
	if err != nil {
		return fmt.Errorf("orchestrate: get node %s: %w", rec.OldNode, err)
	}
	if nodeInfo == nil {
		return fmt.Errorf("orchestrate: old node %s not found", rec.OldNode)
	}

	// Circuit-breaker check (15F).
	if !r.breaker.Allow(nodeInfo.Tier, nodeInfo.Region) {
		slog.Info("orchestrate: breaker blocked replacement",
			"uuid", uuid, "tier", nodeInfo.Tier, "region", nodeInfo.Region)
		// Emit 熔断触发 alert (15G exit channel).
		ev := alert.NewEvent(alert.EventTypeBreakerTripped, rec.OldNode, map[string]string{
			"tier":             nodeInfo.Tier,
			"region":           nodeInfo.Region,
			"replacement_uuid": uuid,
		})
		ev.Pool = nodeInfo.Tier + "/" + nodeInfo.Region
		if notifyErr := r.notifier.Notify(ctx, ev); notifyErr != nil {
			slog.Error("orchestrate: notify breaker tripped", "uuid", uuid, "error", notifyErr)
		}
		return nil // stay pending; retry next Tick
	}

	// Watermark / quota check — stub (always passes).
	// TODO(15F): emit EventTypeWatermarkLow when real capacity guard is wired.

	rec.Phase = PhaseCreating
	rec.PhaseStartedAt = r.clock.Now()
	rec.UpdatedAt = r.clock.Now()
	return r.saveRecord(ctx, uuid, rec)
}

// stepCreating calls CreateNode (idempotent via idempotency key) and advances
// to probing. If NewNode is already set (crash recovery after CreateNode
// succeeded but before the record was saved with phase=probing), CreateNode is
// NOT called again.
func (r *Replacer) stepCreating(ctx context.Context, uuid string, rec *ReplaceRecord) error {
	if rec.NewNode == "" {
		nodeInfo, err := r.lc.GetNode(ctx, rec.OldNode)
		if err != nil {
			return fmt.Errorf("orchestrate: get node %s: %w", rec.OldNode, err)
		}
		if nodeInfo == nil {
			return fmt.Errorf("orchestrate: old node %s not found", rec.OldNode)
		}

		providerID, err := r.pickProvider(ctx, nodeInfo.Tier, nodeInfo.Region, rec.ProviderTried)
		if err != nil {
			return fmt.Errorf("orchestrate: pick provider: %w", err)
		}

		spec := NodeSpec{
			Tier:       nodeInfo.Tier,
			Region:     nodeInfo.Region,
			Role:       nodeInfo.Role,
			ProviderID: providerID,
			RealitySNI: nodeInfo.RealitySNI,
			RealityPBK: nodeInfo.RealityPBK,
			HY2Port:    nodeInfo.HY2Port,
			NameZH:     nodeInfo.NameZH,
			NameEn:     nodeInfo.NameEn,
			Tags:       nodeInfo.Tags,
		}
		idemKey := idempotencyKey(uuid, rec.Attempts)

		newNodeID, err := r.prov.CreateNode(ctx, spec, idemKey)
		if err != nil {
			return fmt.Errorf("orchestrate: create node (attempt %d): %w", rec.Attempts, err)
		}
		rec.NewNode = newNodeID
		rec.CurrentProviderID = providerID
	}

	// Advance to probing; reset streak and set phase start time.
	rec.Phase = PhaseProbing
	rec.ProbeStreak = 0
	rec.PhaseStartedAt = r.clock.Now()
	rec.UpdatedAt = r.clock.Now()
	return r.saveRecord(ctx, uuid, rec)
}

// stepProbing checks probe snapshots for the new node.
// On 2 consecutive passing Ticks it advances to activating.
// On timeout (15 min) or repeated failure it retries with a different provider,
// or marks the record failed after MaxAttempts.
func (r *Replacer) stepProbing(ctx context.Context, uuid string, rec *ReplaceRecord) error {
	now := r.clock.Now()

	// Timeout guard.
	if now.Sub(rec.PhaseStartedAt) > ProbeTimeout {
		return r.failProbeAttempt(ctx, uuid, rec, "probe timeout (15 min)")
	}

	// Read probe snapshots for the new node (from the 15A probe store).
	snapshots, err := r.snaps.SnapshotsByNode(ctx, rec.NewNode)
	if err != nil {
		return fmt.Errorf("orchestrate: snapshots for %s: %w", rec.NewNode, err)
	}

	if probePass(snapshots) {
		rec.ProbeStreak++
	} else {
		rec.ProbeStreak = 0
	}

	if rec.ProbeStreak >= ProbeCyclesRequired {
		// Probe window passed: advance to activating.
		rec.Phase = PhaseActivating
		rec.PhaseStartedAt = now
		rec.UpdatedAt = now
		return r.saveRecord(ctx, uuid, rec)
	}

	// Not yet passed: update streak and wait for next Tick.
	rec.UpdatedAt = now
	return r.saveRecord(ctx, uuid, rec)
}

// failProbeAttempt destroys the failed new node and either schedules a retry
// (with a different provider) or marks the replacement as permanently failed.
func (r *Replacer) failProbeAttempt(ctx context.Context, uuid string, rec *ReplaceRecord, reason string) error {
	slog.Warn("orchestrate: probing failed",
		"uuid", uuid,
		"new_node", rec.NewNode,
		"attempt", rec.Attempts,
		"reason", reason,
	)

	// Destroy the bad new node (bad IP must not enter the pool).
	if rec.NewNode != "" {
		if err := r.prov.DestroyNode(ctx, rec.NewNode); err != nil {
			slog.Error("orchestrate: destroy failed new node",
				"uuid", uuid, "node", rec.NewNode, "error", err)
			// Continue: still mark the attempt as failed.
		}
		// Track which provider was tried so the next attempt avoids it.
		if rec.CurrentProviderID != "" {
			rec.ProviderTried = appendUnique(rec.ProviderTried, rec.CurrentProviderID)
		}
		rec.NewNode = ""
		rec.CurrentProviderID = ""
	}
	rec.Attempts++

	if rec.Attempts >= MaxAttempts {
		// All attempts exhausted: mark failed and alert.
		rec.Phase = PhaseFailed
		rec.UpdatedAt = r.clock.Now()
		if err := r.saveRecord(ctx, uuid, rec); err != nil {
			return err
		}
		_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
		_ = r.rdb.Expire(ctx, replaceKeyPrefix+uuid, replaceTTL).Err()

		// Emit 补新连续失败≥3 alert (15G exit channel).
		ev := alert.NewEvent(alert.EventTypeReplenishFailed, rec.OldNode, map[string]string{
			"attempts":         fmt.Sprintf("%d", rec.Attempts),
			"last_reason":      reason,
			"replacement_uuid": uuid,
		})
		if notifyErr := r.notifier.Notify(ctx, ev); notifyErr != nil {
			slog.Error("orchestrate: notify replenish failed", "uuid", uuid, "error", notifyErr)
		}
		slog.Error("orchestrate: replacement permanently failed — manual review required",
			"uuid", uuid, "old_node", rec.OldNode, "attempts", rec.Attempts)
		return nil
	}

	// Schedule retry: go back to creating (different provider picked next call).
	rec.Phase = PhaseCreating
	rec.ProbeStreak = 0
	rec.PhaseStartedAt = r.clock.Now()
	rec.UpdatedAt = r.clock.Now()
	return r.saveRecord(ctx, uuid, rec)
}

// stepActivating sets initial weight, promotes the new node from probing→up,
// bumps the directory version, and registers grayscale warm-up.
func (r *Replacer) stepActivating(ctx context.Context, uuid string, rec *ReplaceRecord) error {
	// Set initial weight for grayscale ramp.
	if err := r.lc.SetWeight(ctx, rec.NewNode, GrayscaleWeights[0]); err != nil {
		return fmt.Errorf("orchestrate: set initial weight: %w", err)
	}

	// Optimistic-lock transition: probing → up.
	detail := map[string]any{
		"from":             "probing",
		"to":               "up",
		"replacement_uuid": uuid,
		"old_node":         rec.OldNode,
	}
	affected, err := r.lc.TransitionStatus(ctx, rec.NewNode, "probing", "up", detail)
	if err != nil {
		return fmt.Errorf("orchestrate: transition probing→up: %w", err)
	}
	if affected == 0 {
		// Lock conflict: another writer changed the state. Retry next Tick.
		slog.Info("orchestrate: probing→up conflict, will retry",
			"uuid", uuid, "new_node", rec.NewNode)
		return nil
	}

	// Bump directory version so clients re-fetch the updated node list.
	if err := r.lc.BumpVersion(ctx); err != nil {
		slog.Error("orchestrate: bump version", "uuid", uuid, "error", err)
	}

	// Register the new node for grayscale warm-up.
	if err := startGrayscale(ctx, r.rdb, r.clock, rec.NewNode); err != nil {
		slog.Error("orchestrate: start grayscale", "uuid", uuid, "node", rec.NewNode, "error", err)
	}

	rec.Phase = PhaseDrainingOld
	rec.PhaseStartedAt = r.clock.Now()
	rec.UpdatedAt = r.clock.Now()
	return r.saveRecord(ctx, uuid, rec)
}

// stepDrainingOld destroys the old node, records the event in the breaker, and
// finalises the replacement.
func (r *Replacer) stepDrainingOld(ctx context.Context, uuid string, rec *ReplaceRecord) error {
	// Look up old node for breaker recording (best-effort; node may be gone).
	nodeInfo, _ := r.lc.GetNode(ctx, rec.OldNode)

	// Destroy the old (already-down) node and release its IP.
	if err := r.prov.DestroyNode(ctx, rec.OldNode); err != nil {
		return fmt.Errorf("orchestrate: destroy old node %s: %w", rec.OldNode, err)
	}

	// Inform the 15F circuit breaker of this successful replacement.
	if nodeInfo != nil {
		r.breaker.Record(nodeInfo.Tier, nodeInfo.Region)
	}

	rec.Phase = PhaseDone
	rec.UpdatedAt = r.clock.Now()
	if err := r.saveRecord(ctx, uuid, rec); err != nil {
		return err
	}

	slog.Info("orchestrate: replacement complete",
		"uuid", uuid, "old_node", rec.OldNode, "new_node", rec.NewNode)

	// Write audit log.
	_ = r.lc.WriteAuditLog(ctx, "orchestrate", "replacement_done",
		"node:"+rec.OldNode,
		fmt.Sprintf(`{"uuid":%q,"new_node":%q}`, uuid, rec.NewNode),
	)

	// Remove from active index; set TTL for 7-day audit retention.
	_ = r.rdb.SRem(ctx, replaceIndexKey, uuid).Err()
	_ = r.rdb.Expire(ctx, replaceKeyPrefix+uuid, replaceTTL).Err()
	return nil
}

// ─────────────────────────────────────────────────────────────────────────────
// Probe-pass check
// ─────────────────────────────────────────────────────────────────────────────

// probePass returns true when the snapshot set indicates the new node is healthy:
//   - Domestic (CN): ≥ 2/3 ISPs are passing.
//   - Overseas: at least one vantage is present and all overseas vantages pass L1.
func probePass(snapshots map[string]probe.ProbeSnapshot) bool {
	if len(snapshots) == 0 {
		return false
	}

	domesticISPs := make(map[string]bool) // ISP name → failed?
	overseasTotal := 0
	overseasOK := 0

	for _, snap := range snapshots {
		v := snap.Vantage
		rpt := snap.Report
		switch {
		case v.Country == "CN":
			// Normalise "3rd-" prefix (third-party vantages share ISP grouping).
			isp := strings.TrimPrefix(v.ISP, "3rd-")
			failed := isSnapshotFailed(rpt)
			if prev, seen := domesticISPs[isp]; seen {
				domesticISPs[isp] = prev || failed
			} else {
				domesticISPs[isp] = failed
			}
		case v.Country != "":
			overseasTotal++
			if rpt.L1.OK {
				overseasOK++
			}
		}
	}

	if len(domesticISPs) == 0 || overseasTotal == 0 {
		return false // insufficient probe data
	}

	// Domestic: ≥ 2/3 ISPs must be passing (not failed).
	passISPs := 0
	for _, failed := range domesticISPs {
		if !failed {
			passISPs++
		}
	}
	total := len(domesticISPs)
	domesticOK := passISPs*3 >= total*2

	// Overseas: every vantage must pass L1.
	overseasPass := overseasOK == overseasTotal

	return domesticOK && overseasPass
}

// isSnapshotFailed mirrors detect/signals.go isReportFailed: check the highest
// available layer (L3 > L2 > L1).
func isSnapshotFailed(rpt probe.NodeReport) bool {
	if rpt.L3 != nil {
		return !rpt.L3.OK
	}
	if !rpt.L1.OK {
		return true
	}
	if rpt.L2 != nil && !rpt.L2.OK {
		return true
	}
	return false
}

// ─────────────────────────────────────────────────────────────────────────────
// Provider selection
// ─────────────────────────────────────────────────────────────────────────────

// pickProvider selects a provider for a new attempt. It prefers providers NOT
// in previouslyTried so each retry rotates to a different vendor.
func (r *Replacer) pickProvider(ctx context.Context, tier, region string, previouslyTried []string) (string, error) {
	providers, err := r.prov.ListProviders(ctx, tier, region)
	if err != nil {
		return "", fmt.Errorf("orchestrate: list providers: %w", err)
	}
	if len(providers) == 0 {
		return "", fmt.Errorf("orchestrate: no providers for tier=%s region=%s", tier, region)
	}

	tried := make(map[string]bool, len(previouslyTried))
	for _, id := range previouslyTried {
		tried[id] = true
	}

	for _, p := range providers {
		if !tried[p.ID] {
			return p.ID, nil
		}
	}
	// All providers have been tried; fall back to the first one.
	return providers[0].ID, nil
}

// ─────────────────────────────────────────────────────────────────────────────
// Redis helpers
// ─────────────────────────────────────────────────────────────────────────────

// loadRecord reads and unmarshals a ReplaceRecord. Returns (nil, nil) if absent.
func (r *Replacer) loadRecord(ctx context.Context, uuid string) (*ReplaceRecord, error) {
	val, err := r.rdb.Get(ctx, replaceKeyPrefix+uuid).Result()
	if err == redis.Nil {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("orchestrate: load %s: %w", uuid, err)
	}
	var rec ReplaceRecord
	if err := json.Unmarshal([]byte(val), &rec); err != nil {
		return nil, fmt.Errorf("orchestrate: unmarshal %s: %w", uuid, err)
	}
	return &rec, nil
}

// saveRecord marshals and persists rec under sched:replace:{uuid}.
func (r *Replacer) saveRecord(ctx context.Context, uuid string, rec *ReplaceRecord) error {
	data, err := json.Marshal(rec)
	if err != nil {
		return fmt.Errorf("orchestrate: marshal %s: %w", uuid, err)
	}
	if err := r.rdb.Set(ctx, replaceKeyPrefix+uuid, data, 0).Err(); err != nil {
		return fmt.Errorf("orchestrate: save %s: %w", uuid, err)
	}
	return nil
}

// ─────────────────────────────────────────────────────────────────────────────
// Misc helpers
// ─────────────────────────────────────────────────────────────────────────────

// idempotencyKey derives the CreateNode idempotency key for the given attempt.
// Attempt 0 uses the replacement UUID directly; retries append a suffix to
// ensure each attempt is independently idempotent.
func idempotencyKey(replacementUUID string, attempt int) string {
	if attempt == 0 {
		return replacementUUID
	}
	return fmt.Sprintf("%s:retry:%d", replacementUUID, attempt)
}

// appendUnique appends s to slice only if s is not already present.
func appendUnique(slice []string, s string) []string {
	for _, v := range slice {
		if v == s {
			return slice
		}
	}
	return append(slice, s)
}