// Package cluster implements the token-ring cooperative network.
// Authoritative design: plan.md §M6 (令牌环网拓扑).
package cluster

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

	"webui4frpc/internal/store"
)

// Handler is what the engine calls when the node must act on a task
// (create the frpc worker / forward). Injected to avoid import cycle.
type Handler interface {
	Claim(ctx context.Context, tk *Task) error
	Revoke(ctx context.Context, tk *Task) error
	RuntimeLoad() Load
}

// Engine drives one member of the token ring.
type Engine struct {
	ID      string
	Addr    string
	User    string
	Pass    string
	Version string
	Cache   []string
	Handler Handler

	// nodeKey is this node's cluster admission key. A newcomer must present
	// the sponsor's nodeKey (as JoinInfo.JoinKey) to join via it. Persisted
	// in store.Settings; stable across restarts so -join-key stays valid.
	// Generated lazily: on CreateCluster (creator) or AdoptState (joiner),
	// NOT at startup — a fresh node that hasn't created/joined has no key.
	// Cleared on detachAsStandalone (leaving the cluster invalidates the key).
	nodeKey    string
	keyPersist func(key string) error // persists nodeKey to store (nil in tests)
	// peerPersist saves the cached peer list (JSON of [{addr,key},...]) so a
	// crashed node can auto-rejoin on restart via any cached peer. Called on
	// every token cycle (OnToken) and on AdoptState. Cleared (pass "") on
	// detachAsStandalone — an explicit leave must NOT auto-rejoin.
	peerPersist func(peersJSON string) error
	// topologySync is called right after e.state = tk.State (OnToken) or
	// e.state = s (AdoptState) so the host can re-apply local store overrides
	// (e.g. group labels) onto the freshly adopted topology. Without this,
	// group changes made via HTTP handlers between token cycles are overwritten
	// by the next state adoption and never propagate to other nodes.
	topologySync func()

	state State
	// myAddr maps our Node ID to the address peers dial.
	myAddr string

	// send moves the token to the next node (injected transport).
	send func(ctx context.Context, next string, tk *Token) error

	// inflight tracks token-in-flight state (leader only).
	inflight    inFlight
	Log         *ClusterLog
	lastLogSent int64
	// lastRingStart time of the previous cycle launch (leader throttle).
	lastRingStart time.Time
	// lastTokenAt: leader-stamped token timestamp of the newest valid token
	// this node accepted. Older tokens (multi-token conflict leftovers) are
	// dropped so only one token effectively circulates.
	lastTokenAt int64
	// lastSyncAt unix-seconds when this node last completed a sync round.
	lastSyncAt int64
	// failCount counts consecutive send failures per neighbor; a node is
	// declared offline only after repeated failures (transient jitter must
	// not break the ring).
	failCount map[string]int
	// failMu guards failCount (HTTP handlers run concurrently).
	failMu sync.Mutex
	// pendingJoin: newcomers accepted via JoinNode but NOT YET injected into
	// the token. Per the authoritative design (plan §新节点加入): a sponsor
	// injects the newcomer into the token only when the token reaches it
	// ("令牌发到自己时，先转发给新节点，并把新节点加入令牌中的集群信息").
	// Until then the newcomer is NOT in e.state.Nodes — so the OnToken
	// state-merge (incoming authoritative) cannot wash it back out.
	pendingJoin []JoinInfo
	// selfRemoved is set when a node-remove command targeted THIS node and
	// it has run SelfRemove; the OnToken "append own node info" step is then
	// skipped (otherwise UpsertNode(self) would re-add the removed node).
	selfRemoved bool
	// removedNext: the original successor captured right before SelfRemove,
	// so Forward can hand the token to it even though this node is no longer
	// in state.Nodes (plan §移除节点 step 3: "令牌传递给自身原本的下一家").
	removedNext string
	// published records task IDs this node placed into the token it last
	// forwarded. When the next token returns WITHOUT one of those IDs, the
	// task was consumed downstream (claimed/revoked) — the localPending
	// re-merge must NOT resurrect it, or the task rides forever (a re-claim /
	// re-revoke storm centered on the submitter). Reset each round to the
	// tasks actually leaving on this token.
	published map[string]struct{}
}

// NewEngine builds the engine; state holds this node as initial leader unless
// a peer list says otherwise (creation node starts the ring).
func NewEngine(id, addr, user, pass, version string, cache []string, h Handler, send func(ctx context.Context, next string, tk *Token) error, selfAddr string, isLeader bool, nodeKey string) *Engine {
	e := &Engine{
		ID: id, Addr: addr, User: user, Pass: pass,
		Version: version, Cache: cache, Handler: h,
		nodeKey: nodeKey,
		state: State{
			LeaderID:     "",
			Cycle:        0,
			PendingTasks: map[string]*Task{},
			Topology:     map[string]*TopoEntry{},
			RoundDelay:   2 * time.Second,
		},
		myAddr:    selfAddr,
		send:      send,
		Log:       NewClusterLog(),
		failCount: map[string]int{},
		published: map[string]struct{}{},
	}
	n := Node{ID: id, Addr: selfAddr, Alive: true, IsLeader: isLeader,
		Load: Load{MemPct: 10, NetPct: 10}, Version: version, Cache: cache, NodeKey: nodeKey}
	e.state.UpsertNode(n)
	return e
}

// State returns the engine's current cluster picture.
func (e *Engine) State() *State { return &e.state }

// nextRecipient picks the successor to hand the token to, skipping dead nodes.
func (e *Engine) nextRecipient() (string, bool) {
	return e.state.AliveSuccessor(e.ID)
}

// myNode returns this node's entry from state.
func (e *Engine) myNode() Node {
	i := e.state.Find(e.ID)
	if i < 0 {
		return Node{ID: e.ID, Addr: e.myAddr, Alive: true}
	}
	return e.state.Nodes[i]
}

// loadSnapshot reads our runtime load. The primary signal is the count of
// forwards this node currently owns (Forwards) — this is what makes the
// lowest-load claim actually distribute tasks across nodes instead of the
// leader hogging every claimable task in a single token pass (its stored
// load was never refreshed between claims, so it stayed "lowest"). mem/net
// from the handler only break ties at equal forward count.
func (e *Engine) loadSnapshot() Load {
	l := Load{Forwards: len(e.state.ForwardsOwnedBy(e.ID))}
	if e.Handler != nil {
		b := e.Handler.RuntimeLoad()
		l.MemPct, l.NetPct = b.MemPct, b.NetPct
	} else {
		l.MemPct, l.NetPct = 20, 20
	}
	return l
}

// OnToken is the SINGLE-ROUND token handler. Per the authoritative design
// (plan §令牌环协议): on receiving the token a node simultaneously
//   (a) ADOPTS the carried cluster picture — incoming state is authoritative
//       for membership + per-node fields. This is safe because structural
//       changes (join/remove) are NOT kept in local state waiting to survive
//       a merge: joins ride a separate pendingJoin channel injected INTO the
//       token here, and a self-remove writes the node out of state so the
//       downstream merge naturally drops it.
//   (b) APPLIES the incremental log delta, then re-attaches own fresh entries.
//   (c) APPENDS own node info (load/version) — skipped if we self-removed.
//   (d) INJECTS pending newcomers right after ourselves + forwards to them.
//   (e) Paces via a parallel rhythm timer (max(ops, timer)).
func (e *Engine) OnToken(ctx context.Context, tk *Token) (*Token, error) {
	// Multi-token guard: only the newest leader-stamped token is kept.
	if tk.SentAt > 0 && e.lastTokenAt > 0 && tk.SentAt < e.lastTokenAt {
		log.Printf("ring[%s] drop stale token sentAt=%d (last=%d)", e.ID, tk.SentAt, e.lastTokenAt)
		return nil, nil
	}
	if tk.SentAt > e.lastTokenAt {
		e.lastTokenAt = tk.SentAt
	}
	log.Printf("ring[%s] OnToken cycle=%d", e.ID, tk.Cycle)

	// Parallel rhythm timer: operations run while the pace clock ticks.
	// Delay scales with alive node count (more nodes → lower per-hop delay,
	// keeping the round time ~constant for real-time sync).
	alive := 0
	for i := range tk.State.Nodes {
		if tk.State.Nodes[i].Alive {
			alive++
		}
	}
	rhythm := time.NewTimer(hopDelayFor(alive))
	defer rhythm.Stop()

	// (a) ADOPT the cluster picture. Incoming state is authoritative: joins
	// ride pendingJoin (injected below), removes write the node out of state
	// so a downstream merge drops it — so a plain assignment is correct and
	// does NOT wash out local structural changes.
	// EXCEPTION: pending COMMANDS (SubmitTask / RemoveNode / AddRevoke) are
	// injected into e.state.PendingTasks by HTTP handlers OUTSIDE OnToken, so
	// a blanket overwrite would drop them before the token carries them. Keep
	// local-only pending commands and re-merge them after the adoption — BUT
	// only those never yet published into a departing token. A task we already
	// published that is now absent from the incoming token was consumed
	// downstream (claimed/revoked); resurrecting it would make it ride forever
	// (a re-claim / re-revoke storm centered on the submitter node).
	localPending := map[string]*Task{}
	for id, t := range e.state.PendingTasks {
		if _, sent := e.published[id]; sent {
			continue
		}
		if _, inToken := tk.State.PendingTasks[id]; inToken {
			continue
		}
		localPending[id] = t
	}
	rd := e.state.RoundDelay // preserve if incoming carries none (zero-guard)
	e.state = tk.State
	if e.state.RoundDelay == 0 {
		e.state.RoundDelay = rd
	}
	if e.state.PendingTasks == nil {
		e.state.PendingTasks = map[string]*Task{}
	}
	// Re-apply local store overrides (group labels, disabled flags) onto
	// the freshly adopted topology so they survive state adoption and
	// propagate to all nodes via the next token forward.
	if e.topologySync != nil {
		e.topologySync()
	}
	for id, t := range localPending {
		e.state.PendingTasks[id] = t
	}

	// (b) apply incremental log delta; trim consumed entries off the token.
	if e.Log != nil && len(tk.Log) > 0 {
		wm, err := e.Log.ApplyDelta(tk.Log)
		if err != nil {
			log.Printf("ring[%s] log delta gap: %v (request full sync later)", e.ID, err)
		}
		keep := tk.Log[:0]
		for _, en := range tk.Log {
			if en.Seq > wm {
				keep = append(keep, en)
			}
		}
		tk.Log = keep
	}

	// execute pending commands addressed to us or claimable by lowest load.
	// runCommands may set e.selfRemoved + e.removedNext on a self-remove.
	if err := e.runCommands(ctx, tk); err != nil {
		return tk, err
	}

	// (c) append own node info (refresh load/lastSeen) — skipped when we
	// just self-removed, otherwise UpsertNode(self) would resurrect us and
	// undo the removal the command just performed.
	if !e.selfRemoved {
		e.state.UpsertNode(Node{
			ID: e.ID, Addr: e.myAddr, Alive: true,
			IsLeader: e.state.LeaderID == e.ID,
			Load:     e.loadSnapshot(),
			Version:  e.Version, Cache: e.Cache, NodeKey: e.nodeKey,
		})
	}

	// (d) inject pending newcomers right after ourselves + log node.join.
	// This is the plan's "令牌发到自己时把新节点加入令牌" step: the
	// sponsor writes the newcomer into the token's state and the token then
	// flows to the newcomer (its successor) so it can participate.
	e.injectPendingJoin()

	// re-attach own fresh log entries so peers converge.
	if !e.selfRemoved && e.Log != nil {
		mine := e.Log.EntriesAfter(e.lastLogSent)
		if len(mine) > 0 {
			tk.Log = append(tk.Log, mine...)
			if last := mine[len(mine)-1]; last.Seq > e.lastLogSent {
				e.lastLogSent = last.Seq
			}
		}
	}

	// Publish our consolidated state back into the token.
	tk.State = e.state
	e.lastSyncAt = time.Now().Unix()
	// Persist the cached peer list so a crash/restart can auto-rejoin.
	e.persistPeers()
	// Record the tasks leaving on this token so their absence from the next
	// incoming token is recognized as "consumed downstream" rather than
	// "never sent" — otherwise the localPending re-merge above would
	// resurrect them and they would ride the ring forever.
	e.published = make(map[string]struct{}, len(e.state.PendingTasks))
	for id := range e.state.PendingTasks {
		e.published[id] = struct{}{}
	}
	// Forward only after BOTH the operations and the rhythm timer are done.
	select {
	case <-rhythm.C:
	case <-ctx.Done():
		return tk, ctx.Err()
	}
	return tk, nil
}

// runCommands executes pending commands carried by the token that this node
// must handle. Per the authoritative design (plan §M6), commands are DIRECTED
// to their executor — they are NOT blindly claimed by the lowest-load node:
//   - node-removal: rides the token until it reaches the TARGET node, which
//     self-removes (plan §移除节点: "令牌传递到被移除节点自身时，该节点执行
//     自移除"). A remove command for ANOTHER node is left in pending so it keeps
//     riding; if its target is already gone/offline, the lowest node consumes
//     it as a no-op so it cannot ride forever.
//   - revocation: rides to the OWNING node (plan §任务撤销: "持有该转发的节点
//     收到撤销任务后取消"); a revoke whose forward is already absent from the
//     topology is an idempotent no-op, consumed by the lowest-load node.
//   - forward creation: claimed by the lowest-load node (plan §负载摘取).
func (e *Engine) runCommands(ctx context.Context, tk *Token) error {
	for {
		pending := e.state.PendingList()
		if len(pending) == 0 {
			break
		}
		selfIsLowest := false
		if low := e.state.LowestAlive(); low != nil && low.ID == e.ID {
			selfIsLowest = true
		}
		// Pick the first command this node may act on this pass.
		var target *Task
		for _, t := range pending {
			if t.RemoveNode == e.ID {
				target = t // directed at us — execute regardless of load
				break
			}
			if t.RemoveNode != "" {
				// directed at another node; let it ride unless the target is
				// already gone (not in ring / offline) — then the lowest node
				// consumes the stale command as a no-op so it cannot loop.
				if selfIsLowest {
					if idx := e.state.Find(t.RemoveNode); idx < 0 || !e.state.Nodes[idx].Alive {
						target = t
						break
					}
				}
				continue
			}
			if t.Revoke {
				if owner := e.state.TopologyOwner(t); owner == e.ID {
					target = t // we own the forward — execute the revoke
					break
				} else if owner == "" && selfIsLowest {
					target = t // forward already gone — idempotent no-op
					break
				}
				continue // owned elsewhere — ride to the owner
			}
			if selfIsLowest {
				target = t // generic forward create — lowest-load claim
				break
			}
		}
		if target == nil {
			break
		}
		claimed := e.state.ClaimPending(target.ID)
		if claimed == nil {
			break
		}
		if claimed.Revoke {
			if e.state.RemoveTopology(claimed.Local.Name, claimed.Remote.Name, claimed.Link.RemotePort) {
				if e.Handler != nil {
					if err := e.Handler.Revoke(ctx, claimed); err != nil {
						log.Printf("ring[%s] revoke %s: %v", e.ID, claimed.ID, err)
					}
				}
				if e.Log != nil {
					_, _ = e.Log.Append(e.ID, LogForwardRemove, map[string]any{
						"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
					})
				}
			}
			continue
		}
		if claimed.RemoveNode != "" && claimed.RemoveNode == e.ID {
			// Plan §移除节点: self-remove re-queues own forwards, drops ring
			// position, and the token continues to the original successor.
			// Stop the local workers for every forward we own FIRST (plan:
			// "完全取消一切集群远程转发，停其 worker"), then SelfRemove
			// moves them back to pending for another member to claim.
			if e.Handler != nil {
				for _, te := range e.state.ForwardsOwnedBy(e.ID) {
					t := &Task{ID: te.TaskID, Local: te.Local, Remote: te.Remote, Link: te.Link}
					if err := e.Handler.Revoke(ctx, t); err != nil {
						log.Printf("ring[%s] self-remove revoke %s: %v", e.ID, te.TaskID, err)
					}
				}
			}
			wasLeader := e.state.LeaderID == e.ID
			if succ, ok := e.state.AliveSuccessor(e.ID); ok && succ != e.ID {
				e.removedNext = succ
			}
			e.state.SelfRemove(e.ID)
			e.selfRemoved = true
			// Leader failover: if we were the leader, the ring would run
			// leaderless after our departure — LeaderID would be "" (SelfRemove
			// clears it), no node would call Send (cycle never advances), and
			// WatchLeader can't find AlivePredecessor("") to promote a
			// successor. Designate the captured successor as the new leader
			// so the token carries a valid LeaderID downstream; the successor
			// then calls Send on its turn and the ring keeps cycling.
			if wasLeader && e.removedNext != "" {
				e.state.LeaderID = e.removedNext
				for i := range e.state.Nodes {
					e.state.Nodes[i].IsLeader = e.state.Nodes[i].ID == e.removedNext
				}
				if e.Log != nil {
					_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.removedNext})
				}
			}
			if e.Log != nil {
				_, _ = e.Log.Append(e.ID, LogNodeLeave, map[string]string{"node": e.ID})
			}
			log.Printf("ring[%s] self-removed from cluster (token command %s); next=%s leader=%s",
				e.ID, claimed.ID, e.removedNext, e.state.LeaderID)
			continue
		}
		if claimed.RemoveNode != "" {
			// A remove command whose target is no longer in the ring: the
			// target already self-removed (or was never a member). Drop it
			// here — it MUST NOT fall through to the forward-create path,
			// which would spawn a phantom empty forward (the remove task
			// carries no local/remote/link). This branch also absorbs the
			// command after the injector's localPending merge resurrects a
			// copy once the original has been consumed downstream.
			log.Printf("ring[%s] drop fulfilled remove %s (target %s gone)",
				e.ID, claimed.ID, claimed.RemoveNode)
			continue
		}
		// Defense-in-depth against duplicate claims: if an active topology
		// entry for this forward already exists (owned by us or another
		// node), this task is a stale resurrected copy or a multi-token
		// collision — drop it WITHOUT spawning, so we never end up with an
		// orphaned worker running a forward the topology attributes to a
		// different node. Safe for OfflineReassign: that path deletes the
		// topology entry BEFORE re-queueing, so TopologyOwner returns "" and
		// the legitimate re-claim passes through.
		if owner := e.state.TopologyOwner(claimed); owner != "" {
			log.Printf("ring[%s] drop duplicate task %s: %s→%s:%d already owned by %s",
				e.ID, claimed.ID, claimed.Local.Name, claimed.Remote.Name,
				claimed.Link.RemotePort, owner)
			continue
		}
		if e.Handler != nil {
			if err := e.Handler.Claim(ctx, claimed); err != nil {
				e.state.PendingTasks[claimed.ID] = claimed
				log.Printf("ring[%s] claim %s failed: %v", e.ID, claimed.ID, err)
				break
			}
		}
		if e.Log != nil {
			_, _ = e.Log.Append(e.ID, LogForwardAdd, map[string]any{
				"taskId": claimed.ID, "local": claimed.Local.Name, "remote": claimed.Remote.Name,
			})
		}
		e.state.AddTopology(claimed, e.ID)
		// Refresh our own stored load so the next selfIsLowest check in this
		// same pass sees the incremented Forwards count — otherwise we'd keep
		// claiming (stored load stays stale until we forward the token) and
		// hog every claimable task, defeating lowest-load distribution.
		if i := e.state.Find(e.ID); i >= 0 {
			e.state.Nodes[i].Load = e.loadSnapshot()
		}
	}
	return nil
}

// Forward hands the token to this node's successor over the injected send.
// It is the transport hook used by the HTTP handler after OnToken. If this
// node just self-removed, its ID is no longer in state.Nodes so
// AliveSuccessor would fail — use the original successor captured before
// removal (plan §移除节点 step 3: "令牌传递给自身原本的下一家"). Otherwise
// delegates to forwardToNext which handles send-failure → mark offline →
// reassign → try next hop (plan §故障自幽).
func (e *Engine) Forward(ctx context.Context, tk *Token) error {
	if e.removedNext != "" {
		next := e.removedNext
		e.removedNext = ""
		var err error
		if e.send != nil {
			log.Printf("ring[%s] forward (self-removed) cycle=%d to %s", e.ID, tk.Cycle, next)
			err = e.send(ctx, next, tk)
		}
		// After handing off the token to the old successor, detach to a
		// fresh standalone state so Snapshot() no longer serves the old
		// cluster picture (members, topology, log) after self-leave. The
		// token already carries the published state (with the new leader if
		// we designated one); the local reset does not affect the sent token.
		if e.selfRemoved {
			e.detachAsStandalone()
		}
		return err
	}
	return e.forwardToNext(ctx, tk)
}

// detachAsStandalone resets the engine to a fresh standalone state after a
// self-leave has completed (the token was handed to the old successor). This
// prevents Snapshot() from serving the old cluster picture — other members,
// the full topology, pending tasks, and the cluster log — after the node has
// permanently left the ring. Equivalent to CreateCluster minus the IsMember
// guard (we are already detached) plus a fresh log (old cluster events stale).
func (e *Engine) detachAsStandalone() {
	// Leaving the cluster invalidates this node's admission key — a
	// standalone node has no key until it creates/joins again. Clear both
	// the in-memory key and the persisted copy (so a restart doesn't
	// resurrect a stale key for a node that's no longer in any cluster).
	e.nodeKey = ""
	if e.keyPersist != nil {
		_ = e.keyPersist("")
	}
	// Clear the cached peer list so this node does NOT auto-rejoin on
	// restart — it explicitly left the cluster.
	e.clearPeers()
	e.state = State{
		LeaderID:     e.ID,
		PendingTasks: map[string]*Task{},
		Topology:     map[string]*TopoEntry{},
		RoundDelay:   2 * time.Second,
	}
	e.state.UpsertNode(Node{
		ID: e.ID, Addr: e.myAddr, Alive: true, IsLeader: true,
		Load: e.loadSnapshot(), Version: e.Version, Cache: e.Cache, NodeKey: e.nodeKey,
	})
	e.selfRemoved = false
	e.removedNext = ""
	e.lastRingStart = time.Time{}
	e.lastTokenAt = 0
	e.lastSyncAt = 0
	e.inflight.clear()
	e.failCount = map[string]int{}
	e.published = map[string]struct{}{}
	if e.Log != nil {
		e.Log = NewClusterLog()
		_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.ID})
	}
	log.Printf("ring[%s] detached to standalone after self-leave", e.ID)
}

// Snapshot returns a serializable view of the ring for the frontend.
type RingSnapshot struct {
	SelfID     string       `json:"selfId"`
	LeaderID   string       `json:"leaderId"`
	Cycle      int64        `json:"cycle"`
	LastSync   int64        `json:"lastSync"`
	RoundDelay int64        `json:"roundDelayMs"`
	Nodes      []Node       `json:"nodes"`
	Pending    []*Task      `json:"pending"`
	Topology   []*TopoEntry `json:"topology"`
	Log        []LogEntry   `json:"log,omitempty"`
	// NodeKey: this node's cluster admission key. The cluster page displays it
	// so the operator can copy it for newcomers joining via this node.
	NodeKey string `json:"nodeKey,omitempty"`
}

func (e *Engine) Snapshot() *RingSnapshot {
	snap := &RingSnapshot{
		SelfID:     e.ID,
		LeaderID:   e.state.LeaderID,
		Cycle:      e.state.Cycle,
		LastSync:   e.lastSyncAt,
		RoundDelay: e.state.RoundDelay.Milliseconds(),
		Nodes:      e.state.Nodes,
		Pending:    e.state.PendingList(),
		Topology:   e.state.TopologyList(),
		NodeKey:    e.nodeKey,
	}
	if e.Log != nil {
		snap.Log = e.Log.Snapshot()
	}
	return snap
}

// tkDelaySince measures elapsed since token SentAt (leader round delay).
func tkDelaySince(tk *Token) time.Duration {
	if tk.SentAt == 0 {
		return 2 * time.Second
	}
	return time.Duration(time.Now().UnixMilli()-tk.SentAt) * time.Millisecond
}

func contains(xs []string, v string) bool {
	for _, x := range xs {
		if x == v {
			return true
		}
	}
	return false
}

// StartRing kicks off a fresh token from the leader. The leader stamps a new
// SentAt (so older in-flight tokens are dropped downstream) and marks inflight
// so WatchTokenLoss can detect a lost first hop — previously StartRing sent
// without marking, so a first-round loss was never noticed.
func (e *Engine) StartRing(ctx context.Context) {
	if e.state.LeaderID != e.ID {
		return
	}
	next, ok := e.state.AliveSuccessor(e.ID)
	if !ok || next == e.ID {
		return // single-node ring; resumes once a newcomer joins
	}
	// Throttle: do not start a new cycle until RoundDelay has elapsed.
	if !e.lastRingStart.IsZero() && time.Since(e.lastRingStart) < e.state.RoundDelay {
		return
	}
	e.lastRingStart = time.Now()
	e.state.Cycle++
	tk := &Token{
		Cycle:  e.state.Cycle,
		State:  e.state,
		SentAt: time.Now().UnixMilli(),
	}
	if e.send != nil {
		if err := e.send(ctx, next, tk); err != nil {
			log.Printf("ring[%s] start cycle %d send: %v", e.ID, tk.Cycle, err)
			return
		}
		// Mark inflight ONLY on a successful first send so token-loss
		// detection covers the very first hop too.
		e.inflight.mark(e.state.RoundDelay)
	}
}

// JoinInfo is a node's self-description sent when requesting to join a ring.
type JoinInfo struct {
	ID      string   `json:"id"`
	Addr    string   `json:"addr"`
	Version string   `json:"version,omitempty"`
	Cache   []string `json:"cache,omitempty"`
	// JoinKey is the sponsor's nodeKey — the newcomer must present it to
	// prove it is authorized to join via the sponsor. The sponsor verifies
	// ji.JoinKey == e.nodeKey; mismatch → 403.
	JoinKey string `json:"joinKey,omitempty"`
}

// injectPendingJoin writes every queued newcomer into state right after
// ourselves (so the newcomer becomes our successor) and logs node.join. Used
// both by OnToken (plan: "令牌发到自己时把新节点加入令牌") and by JoinNode
// when the leader is a single node with no token circulating.
func (e *Engine) injectPendingJoin() {
	for _, ji := range e.pendingJoin {
		nn := Node{ID: ji.ID, Addr: ji.Addr, Alive: true,
			Load: Load{MemPct: 50, NetPct: 50}, Version: ji.Version, Cache: ji.Cache}
		e.state.InsertAfter(e.ID, nn)
		if e.Log != nil {
			_, _ = e.Log.Append(e.ID, LogNodeJoin, map[string]string{"node": nn.ID, "addr": nn.Addr})
		}
		log.Printf("ring[%s] injected newcomer %s after self", e.ID, nn.ID)
	}
	e.pendingJoin = nil
}

// JoinNode accepts a newcomer's join request. Per plan §新节点加入 the
// sponsor does NOT mutate its own state immediately (it would be washed out
// by the next incoming token's authoritative merge). Instead it queues the
// newcomer on pendingJoin; OnToken injects it into the token right after
// the sponsor. EXCEPTION: a single-node leader has no token circulating
// (StartRing refuses for lack of a successor), so the newcomer would never
// be injected — in that case we inject immediately and kick off the ring.
func (e *Engine) JoinNode(j JoinInfo) *State {
	e.pendingJoin = append(e.pendingJoin, j)
	if e.state.LeaderID == "" {
		e.state.LeaderID = e.ID
	}
	// Single-node leader: no successor, no circulating token → the pending
	// newcomer would never be injected. Inject now (safe: no foreign token
	// can overwrite a single-node leader) and start the ring.
	if e.state.LeaderID == e.ID && len(e.state.Nodes) == 1 {
		e.injectPendingJoin()
		go e.StartRing(context.Background())
	}
	return &e.state
}

// AdoptState replaces this node's cluster picture with the state provided by
// the join target, then re-inserts us (in case we were absent).
func (e *Engine) AdoptState(s State) {
	ns := make([]string, 0, len(s.Nodes))
	for _, n := range s.Nodes {
		ns = append(ns, n.ID)
	}
	log.Printf("ring[%s] adopt-state nodes=%v", e.ID, ns)
	kept := map[string]*Task{}
	for id, t := range e.state.PendingTasks {
		if _, ok := s.PendingTasks[id]; !ok {
			kept[id] = t
		}
	}
	e.state = s
	if e.state.PendingTasks == nil {
		e.state.PendingTasks = map[string]*Task{}
	}
	for id, t := range kept {
		e.state.PendingTasks[id] = t
	}
	if e.topologySync != nil {
		e.topologySync()
	}
	e.state.UpsertNode(Node{ID: e.ID, Addr: e.myAddr, Alive: true,
		Load: e.loadSnapshot(), Version: e.Version, Cache: e.Cache, NodeKey: e.nodeKey})
	if e.Log != nil {
		e.Log.Append(e.ID, LogNodeJoin, map[string]string{"node": e.ID, "addr": e.myAddr})
	}
	// Newcomer generates its own admission key after joining, so future
	// nodes can join via it. Per the user's design: "每个节点加入集群后
	// 生成自身密钥". A node that already has a persisted key (restart
	// re-join) keeps it.
	e.ensureNodeKey()
	// Persist the cached peer list from the adopted ring state.
	e.persistPeers()
}

// CreateCluster reseeds this node as a fresh standalone leader (single-node
// ring). Used after a self-leave left the ring empty, or to (re)affirm seed
// state on a standalone node. Refuses if this node is still a multi-node
// member — reseeding mid-cluster would split the ring (split-brain). Safe to
// call when standalone/empty: no token circulates to a non-member, so no
// concurrent OnToken can overwrite the reset.
func (e *Engine) CreateCluster() error {
	if e.IsMember() {
		return fmt.Errorf("node is a multi-node cluster member; leave first")
	}
	e.ensureNodeKey()
	e.state = State{
		LeaderID:     e.ID,
		Cycle:        0,
		PendingTasks: map[string]*Task{},
		Topology:     map[string]*TopoEntry{},
		RoundDelay:   2 * time.Second,
	}
	e.state.UpsertNode(Node{
		ID: e.ID, Addr: e.myAddr, Alive: true, IsLeader: true,
		Load: e.loadSnapshot(), Version: e.Version, Cache: e.Cache, NodeKey: e.nodeKey,
	})
	e.selfRemoved = false
	e.removedNext = ""
	e.lastRingStart = time.Time{}
	e.lastTokenAt = 0
	e.inflight.clear()
	if e.Log != nil {
		_, _ = e.Log.Append(e.ID, LogLeaderChange, map[string]string{"leader": e.ID})
	}
	log.Printf("ring[%s] created/seeded fresh standalone cluster as leader", e.ID)
	return nil
}

// SubmitTask adds a new forward request to pending; it rides the next token
// round and is claimed by the lowest-load member.
// RemoveNode publishes a node-removal command via the token; the target
// node self-removes when the command reaches it.
func (e *Engine) RemoveNode(nodeID string) *Task {
	return e.state.AddRemoveNode(nodeID)
}

// RevokeTask publishes a revocation for an established forward through the
// same token channel; the owning node stops the worker and drops topology.
func (e *Engine) RevokeTask(local store.Local, remote store.Remote, link store.Link) *Task {
	return e.state.AddRevoke(local, remote, link)
}

func (e *Engine) SubmitTask(local store.Local, remote store.Remote, link store.Link) *Task {
	if e.HasTask(local.Name, remote.Name, link.RemotePort) {
		return nil
	}
	return e.state.AddPending(local, remote, link)
}

// HasTask reports whether a forward with the same local/remote/remotePort is
// already pending or active in the topology (idempotency guard for resaves).
func (e *Engine) HasTask(local, remote string, port int) bool {
	for _, t := range e.state.PendingList() {
		if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
			return true
		}
	}
	for _, t := range e.state.TopologyList() {
		if t.Local.Name == local && t.Remote.Name == remote && t.Link.RemotePort == port {
			return true
		}
	}
	return false
}

// UpdateTopologyGroup sets the group label on a topology entry. The change
// propagates to all nodes via the next token cycle. Used by the forwards
// assign handler so group labels sync through the ring.
func (e *Engine) UpdateTopologyGroup(local, remote string, port int, group string) bool {
	return e.state.UpdateTopologyGroup(local, remote, port, group)
}

// IsLeader reports whether this node is the current ring leader.
func (e *Engine) IsLeader() bool { return e.state.LeaderID == e.ID }

// NodeKey returns this node's cluster admission key (for the frontend to
// display so the operator can copy it for newcomers).
func (e *Engine) NodeKey() string { return e.nodeKey }

// SetKeyPersist installs the callback used to persist the nodeKey to durable
// storage (store.SetNodeKey). Called once from main.go after NewEngine. Tests
// leave it nil — ensureNodeKey still generates the key in-memory.
func (e *Engine) SetKeyPersist(fn func(key string) error) { e.keyPersist = fn }

// SetPeerPersist installs the callback used to persist the cached peer list
// to durable storage (store.SetClusterPeers). Called once from main.go.
func (e *Engine) SetPeerPersist(fn func(peersJSON string) error) { e.peerPersist = fn }

// SetTopologySync installs the callback used to re-apply local store overrides
// (group labels, disabled flags) onto the ring topology after each state
// adoption. Called once from main.go. Without this, group changes made via
// HTTP handlers are overwritten by the next e.state = tk.State.
func (e *Engine) SetTopologySync(fn func()) { e.topologySync = fn }

// persistPeers extracts all alive peers (addr + nodeKey, excluding self)
// from the current ring state and persists them via the peerPersist callback.
// Called on every token cycle (OnToken) and on AdoptState so a crashed node
// always has the latest peer list to rejoin through. Skipped for standalone
// (single-node) rings — a standalone node has no peers to cache.
func (e *Engine) persistPeers() {
	if e.peerPersist == nil {
		return
	}
	type peerEntry struct {
		Addr string `json:"addr"`
		Key  string `json:"key"`
	}
	var peers []peerEntry
	for _, n := range e.state.Nodes {
		if n.ID == e.ID || !n.Alive {
			continue
		}
		if n.Addr == "" || n.NodeKey == "" {
			continue
		}
		peers = append(peers, peerEntry{Addr: n.Addr, Key: n.NodeKey})
	}
	if len(peers) == 0 {
		return // standalone or all-offline: don't overwrite a good cache
	}
	data, err := json.Marshal(peers)
	if err != nil {
		return
	}
	if err := e.peerPersist(string(data)); err != nil {
		log.Printf("ring[%s] persist cluster peers failed: %v", e.ID, err)
	}
}

// clearPeers wipes the cached peer list (called from detachAsStandalone so
// an explicit leave does NOT auto-rejoin on restart).
func (e *Engine) clearPeers() {
	if e.peerPersist != nil {
		_ = e.peerPersist("")
	}
}

// ensureNodeKey generates a random admission key if this node doesn't have one
// yet, and persists it via the keyPersist callback (so it survives restarts).
// Called from CreateCluster (the creator generates a key so others can join
// via it) and AdoptState (a newcomer generates its own key after joining, so
// future nodes can join via it). Per the user's design: "每个节点加入集群后
// 生成自身密钥" — the key is born with cluster membership, not at startup.
func (e *Engine) ensureNodeKey() {
	if e.nodeKey != "" {
		return
	}
	e.nodeKey = GenerateNodeKey()
	if e.keyPersist != nil {
		if err := e.keyPersist(e.nodeKey); err != nil {
			log.Printf("ring[%s] persist nodeKey failed: %v", e.ID, err)
		}
	}
}

// IsMember reports whether this node is currently an active multi-node member
// (self is in the ring alongside others). Used by the create/join gates to
// refuse actions that would split an active ring. A detached node (self not
// in ring — e.g. after a self-leave) or a standalone node returns false and
// may create/join freely. NOTE: after a self-leave the engine keeps the other
// members in state.Nodes (it only dropped self), so a plain len>1 check would
// wrongly block a detached node — the self-in-ring test is essential.
func (e *Engine) IsMember() bool {
	if len(e.state.Nodes) <= 1 {
		return false
	}
	for _, n := range e.state.Nodes {
		if n.ID == e.ID {
			return true
		}
	}
	return false
}

// LeaderAddr returns the current leader's address.
func (e *Engine) LeaderAddr() string {
	i := e.state.Find(e.state.LeaderID)
	if i < 0 {
		return ""
	}
	return e.state.Nodes[i].Addr
}