NexusGate/panel-go/orchestrator.go

package main

import (
	"encoding/json"
	"fmt"
	"math"
	"strconv"
	"strings"
	"sync"
	"time"
)

const HeartbeatReconcileIntervalMs = 120000

// ---------- small finders / helpers ----------

func orchFindServer(d *Data, id string) *Server {
	for _, s := range d.Servers {
		if s.ID == id {
			return s
		}
	}
	return nil
}

func orchFindNode(d *Data, id string) *Node {
	for _, n := range d.Nodes {
		if n.ID == id {
			return n
		}
	}
	return nil
}

func orchFindCustomer(d *Data, id string) *Customer {
	for _, c := range d.Customers {
		if c.ID == id {
			return c
		}
	}
	return nil
}

func orchNotFound(label string) *HTTPError { return httpErr(404, "%s不存在", label) }

func xbTruncRunes(s string, n int) string {
	r := []rune(s)
	if len(r) > n {
		return string(r[:n])
	}
	return s
}

func forwardNode(node *Node) bool {
	topology := node.Topology
	if topology == "" {
		topology = "forward"
	}
	return topology == "forward"
}

func tagForNode(node *Node) string {
	var sb strings.Builder
	for _, r := range node.ID {
		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' {
			sb.WriteRune(r)
		}
	}
	s := sb.String()
	if len(s) > 18 {
		s = s[len(s)-18:]
	}
	return "ng-" + s
}

func activeNodeClients(d *Data, node *Node) []*NodeClient {
	out := []*NodeClient{}
	for _, nc := range d.NodeClients {
		if nc.NodeID == node.ID && nc.Status == "active" {
			out = append(out, nc)
		}
	}
	return out
}

func NodePortInUse(d *Data, serverID string, port int, excludeNodeID string) bool {
	for _, n := range d.Nodes {
		if n.ID == excludeNodeID {
			continue
		}
		if n.ServerID == serverID && n.Port == port {
			return true
		}
		if n.ExitServerID != nil && *n.ExitServerID == serverID && n.ExitPort != nil && *n.ExitPort == port {
			return true
		}
	}
	return false
}

// ---------- ports ----------

func AllocatePort(d *Data, server *Server) (int, error) {
	start := server.PortRangeStart
	if start == 0 {
		start = 20000
	}
	end := server.PortRangeEnd
	if end == 0 {
		end = 50000
	}
	if start < 1024 || end > 65535 || start > end {
		return 0, fmt.Errorf("服务器 %s 的端口范围无效", server.Name)
	}
	span := end - start + 1
	offset := xbRandInt(span)
	for i := 0; i < span; i++ {
		candidate := start + ((offset + i) % span)
		if !NodePortInUse(d, server.ID, candidate, "") {
			return candidate, nil
		}
	}
	return 0, fmt.Errorf("服务器 %s 没有可用端口", server.Name)
}

// anyToJSNumber converts a decoded-JSON value the way JS Number() would;
// ok=false mirrors NaN.
func anyToJSNumber(v any) (float64, bool) {
	switch t := v.(type) {
	case json.Number:
		f, err := t.Float64()
		if err != nil {
			return 0, false
		}
		return f, true
	case float64:
		return t, true
	case float32:
		return float64(t), true
	case int:
		return float64(t), true
	case int64:
		return float64(t), true
	case string:
		f := orchJSNumberStr(t)
		if math.IsNaN(f) {
			return 0, false
		}
		return f, true
	case bool:
		if t {
			return 1, true
		}
		return 0, true
	}
	return 0, false
}

func jsFalsy(v any) bool {
	if v == nil {
		return true
	}
	if f, ok := anyToJSNumber(v); ok && f == 0 {
		return true
	}
	if s, isStr := v.(string); isStr && s == "" {
		return true
	}
	if b, isBool := v.(bool); isBool && !b {
		return true
	}
	return false
}

func SelectPort(d *Data, server *Server, mode string, requested any, excludeNodeID string) (int, error) {
	if mode != "fixed" || jsFalsy(requested) {
		return AllocatePort(d, server)
	}
	f, ok := anyToJSNumber(requested)
	if !ok || f != math.Trunc(f) || f < 1024 || f > 65535 {
		return 0, httpErr(400, "服务器 %s 的固定端口无效", server.Name)
	}
	port := int(f)
	if NodePortInUse(d, server.ID, port, excludeNodeID) {
		return 0, httpErr(409, "服务器 %s 的端口 %d 已被占用", server.Name, port)
	}
	return port, nil
}

// ---------- jobs / audit ----------

func QueueJob(d *Data, serverID, nodeID, action string, payload *JobPayload) *Job {
	now := NowISO()
	job := &Job{
		ID:        NewID("job"),
		ServerID:  serverID,
		NodeID:    nodeID,
		Action:    action,
		Payload:   payload,
		Status:    "queued",
		Attempts:  0,
		CreatedAt: now,
		UpdatedAt: now,
		Error:     nil,
	}
	d.Jobs = append(d.Jobs, job)
	return job
}

func Audit(d *Data, actor, action, target string, detail map[string]any) {
	if detail == nil {
		detail = map[string]any{}
	}
	entry := &Activity{
		ID:     NewID("evt"),
		At:     NowISO(),
		Actor:  actor,
		Action: action,
		Target: target,
		Detail: detail,
	}
	d.Activity = append([]*Activity{entry}, d.Activity...)
	if len(d.Activity) > 2000 {
		d.Activity = d.Activity[:2000]
	}
}

// ---------- resource assembly ----------

func payloadWithResource(resource map[string]any, kind string, keyMaterial map[string]KeyPair) *JobPayload {
	raw, err := marshalJSON(resource)
	if err != nil {
		panic(err)
	}
	payload := &JobPayload{Resource: raw, Kind: kind}
	if keyMaterial != nil {
		payload.KeyMaterial = keyMaterial
	}
	return payload
}

func resourceForNode(d *Data, node *Node, clients []*NodeClient) (map[string]any, error) {
	server := orchFindServer(d, node.ServerID)
	if server == nil {
		return nil, orchNotFound("服务器")
	}
	networkMode := node.NetworkMode
	if networkMode == "" {
		networkMode = "ipv4"
	}
	bindAddress, err := ResolveBindAddress(networkMode, server)
	if err != nil {
		return nil, err
	}
	var exitServer *Server
	if forwardNode(node) {
		exitServer = orchFindServer(d, xbDeref(node.ExitServerID))
		if exitServer == nil {
			return nil, orchNotFound("出口服务器")
		}
	}
	if clients == nil {
		clients = activeNodeClients(d, node)
	}
	return BuildNodeResource(BuildNodeResourceParams{
		ResourceID:  node.ResourceID,
		TagPrefix:   tagForNode(node),
		Node:        node,
		NodeClients: clients,
		Server:      server,
		BindAddress: bindAddress,
		ExitServer:  exitServer,
	})
}

func resourceForNodeExit(d *Data, node *Node) (map[string]any, error) {
	server := orchFindServer(d, xbDeref(node.ExitServerID))
	if server == nil {
		return nil, orchNotFound("出口服务器")
	}
	networkMode := node.NetworkMode
	if networkMode == "" {
		networkMode = "ipv4"
	}
	bindAddress, err := ResolveBindAddress(networkMode, server)
	if err != nil {
		return nil, err
	}
	return BuildNodeExitResource(BuildNodeExitResourceParams{
		ResourceID:  node.ExitResourceID,
		TagPrefix:   tagForNode(node),
		Node:        node,
		Server:      server,
		BindAddress: bindAddress,
	})
}

func cancelQueuedDeletes(d *Data, nodeID, reason string) bool {
	cancelled := false
	for _, job := range d.Jobs {
		if job.NodeID == nodeID && job.Action == "delete_resource" && job.Status == "queued" {
			job.Status = "failed"
			job.Error = StrPtr(reason)
			job.UpdatedAt = NowISO()
			cancelled = true
		}
	}
	return cancelled
}

func RebuildNode(d *Data, node *Node, includeExit bool) error {
	clients := activeNodeClients(d, node)
	if len(clients) == 0 {
		if node.Status != "draft" && node.Status != "removing" {
			QueueJob(d, node.ServerID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ResourceID, Kind: "node"})
			if forwardNode(node) && node.ExitResourceID != "" {
				QueueJob(d, xbDeref(node.ExitServerID), node.ID, "delete_resource", &JobPayload{ResourceID: node.ExitResourceID, Kind: "node-exit"})
			}
			node.Status = "removing"
			node.AutoRemoving = BoolPtr(true)
			node.PendingDelete = false
			node.Error = nil
		}
		node.UpdatedAt = NowISO()
		return nil
	}
	node.AutoRemoving = BoolPtr(false)
	resource, err := resourceForNode(d, node, clients)
	if err != nil {
		return err
	}
	payload := payloadWithResource(resource, "node", KeyMaterialForNode(d, node))
	QueueJob(d, node.ServerID, node.ID, "apply_resource", payload)
	if forwardNode(node) && includeExit {
		exitResource, err := resourceForNodeExit(d, node)
		if err != nil {
			return err
		}
		QueueJob(d, xbDeref(node.ExitServerID), node.ID, "apply_resource", payloadWithResource(exitResource, "node-exit", nil))
	}
	node.Status = "deploying"
	node.Error = nil
	node.UpdatedAt = NowISO()
	return nil
}

func DeriveNodeStatus(node *Node) {
	if !node.EntryReady {
		if node.Status != "failed" {
			node.Status = "deploying"
		}
		return
	}
	if forwardNode(node) && !node.ExitReady {
		if node.Status != "degraded" {
			node.Status = "deploying"
		}
		return
	}
	node.Status = "active"
	node.Error = nil
}

// ---------- customer suspend / resume ----------

func SuspendCustomerResources(d *Data, customer *Customer, actor string) (int, error) {
	if actor == "" {
		actor = "system"
	}
	affected := []string{}
	seen := map[string]bool{}
	for _, nc := range d.NodeClients {
		if nc.CustomerID == customer.ID && nc.Status == "active" {
			nc.Status = "suspended"
			nc.UpdatedAt = NowISO()
			if !seen[nc.NodeID] {
				seen[nc.NodeID] = true
				affected = append(affected, nc.NodeID)
			}
		}
	}
	for _, nodeID := range affected {
		node := orchFindNode(d, nodeID)
		if node != nil && node.Status != "draft" && node.Status != "removing" {
			if err := RebuildNode(d, node, false); err != nil {
				return len(affected), err
			}
		}
	}
	Audit(d, actor, "suspend_customer_resources", customer.ID, map[string]any{"nodes": len(affected)})
	return len(affected), nil
}

func ResumeCustomerResources(d *Data, customer *Customer, actor string) (int, error) {
	if actor == "" {
		actor = "admin"
	}
	affected := []string{}
	seen := map[string]bool{}
	for _, nc := range d.NodeClients {
		if nc.CustomerID == customer.ID && nc.Status == "suspended" {
			nc.Status = "active"
			nc.UpdatedAt = NowISO()
			if !seen[nc.NodeID] {
				seen[nc.NodeID] = true
				affected = append(affected, nc.NodeID)
			}
		}
	}
	for _, nodeID := range affected {
		node := orchFindNode(d, nodeID)
		if node == nil {
			continue
		}
		if node.Status == "removing" && !boolVal(node.AutoRemoving) {
			continue
		}
		if node.Status == "draft" && !boolVal(node.AutoRemoving) {
			continue
		}
		if node.Status == "removing" && boolVal(node.AutoRemoving) {
			deleteRunning := false
			for _, job := range d.Jobs {
				if job.NodeID == nodeID && job.Action == "delete_resource" && job.Status == "running" {
					deleteRunning = true
					break
				}
			}
			cancelQueuedDeletes(d, nodeID, "客户恢复，清理任务已取消")
			node.AutoRemoving = BoolPtr(false)
			if !deleteRunning {
				DeriveNodeStatus(node)
				node.UpdatedAt = NowISO()
				continue
			}
		}
		includeExit := forwardNode(node) && !node.ExitReady
		if err := RebuildNode(d, node, includeExit); err != nil {
			return len(affected), err
		}
	}
	Audit(d, actor, "resume_customer_resources", customer.ID, map[string]any{"nodes": len(affected)})
	return len(affected), nil
}

// ---------- usage delta accounting ----------

type DeltaResult struct {
	UpDelta   int64
	DownDelta int64
}

// UsageHolder is the small interface ApplyDelta operates on; *UsageSlot and
// *Node are adapted onto it by SlotHolder / NodeHolder.
type UsageHolder interface {
	GetCounters() *UsageCounters
	SetCounters(c *UsageCounters)
	GetRetired() []string
	SetRetired(v []string)
}

type slotUsageHolder struct{ slot *UsageSlot }

func (h slotUsageHolder) GetCounters() *UsageCounters  { return h.slot.UsageCounters }
func (h slotUsageHolder) SetCounters(c *UsageCounters) { h.slot.UsageCounters = c }
func (h slotUsageHolder) GetRetired() []string         { return h.slot.RetiredUsageEpochs }
func (h slotUsageHolder) SetRetired(v []string)        { h.slot.RetiredUsageEpochs = v }

type nodeUsageHolder struct{ node *Node }

func (h nodeUsageHolder) GetCounters() *UsageCounters  { return h.node.UsageCounters }
func (h nodeUsageHolder) SetCounters(c *UsageCounters) { h.node.UsageCounters = c }
func (h nodeUsageHolder) GetRetired() []string         { return h.node.RetiredUsageEpochs }
func (h nodeUsageHolder) SetRetired(v []string)        { h.node.RetiredUsageEpochs = v }

func SlotHolder(s *UsageSlot) UsageHolder { return slotUsageHolder{slot: s} }
func NodeHolder(n *Node) UsageHolder      { return nodeUsageHolder{node: n} }

// ApplyDelta mirrors applyDelta; a nil result means the sample is skipped.
func ApplyDelta(holder UsageHolder, uplink, downlink int64, sampleEpoch *string) *DeltaResult {
	if sampleEpoch != nil && *sampleEpoch == "" {
		sampleEpoch = nil
	}
	previous := holder.GetCounters()
	if previous == nil {
		previous = &UsageCounters{}
	}
	retired := holder.GetRetired()
	if sampleEpoch != nil {
		for _, e := range retired {
			if e == *sampleEpoch {
				return nil
			}
		}
	}
	restarted := sampleEpoch != nil && previous.Epoch != nil && *sampleEpoch != *previous.Epoch
	if !restarted && sampleEpoch != nil && previous.Epoch != nil &&
		!strings.HasSuffix(*sampleEpoch, ":unknown") &&
		(uplink < previous.Uplink || downlink < previous.Downlink) {
		return nil
	}
	var upDelta, downDelta int64
	if !restarted && uplink >= previous.Uplink {
		upDelta = uplink - previous.Uplink
	} else {
		upDelta = uplink
	}
	if !restarted && downlink >= previous.Downlink {
		downDelta = downlink - previous.Downlink
	} else {
		downDelta = downlink
	}
	if restarted && previous.Epoch != nil {
		next := append(append([]string{}, retired...), *previous.Epoch)
		if len(next) > 8 {
			next = next[len(next)-8:]
		}
		holder.SetRetired(next)
	}
	holder.SetCounters(&UsageCounters{Uplink: uplink, Downlink: downlink, Epoch: sampleEpoch})
	return &DeltaResult{UpDelta: upDelta, DownDelta: downDelta}
}

// ---------- Reality key backfill ----------

func MergeKeyMaterialBackfill(d *Data, node *Node, artifacts map[string]json.RawMessage) {
	if artifacts == nil {
		return
	}
	reportedRaw, hasReported := artifacts["keyMaterial"]
	delete(artifacts, "keyMaterial")
	if hasReported && len(reportedRaw) > 0 {
		var reported map[string]any
		if err := unmarshalJSONUseNumber(reportedRaw, &reported); err == nil && reported != nil {
			if d.KeyStore == nil {
				d.KeyStore = map[string]*KeyEntry{}
			}
			for keyID, pairAny := range reported {
				pair, isObj := pairAny.(map[string]any)
				if !isObj {
					continue
				}
				priv, privOK := pair["privateKey"].(string)
				pub, pubOK := pair["publicKey"].(string)
				if !privOK || !pubOK {
					continue
				}
				existing := d.KeyStore[keyID]
				if existing == nil {
					d.KeyStore[keyID] = &KeyEntry{PrivateKey: priv, PublicKey: pub, UpdatedAt: NowISO()}
				} else if existing.PrivateKey != priv || existing.PublicKey != pub {
					Audit(d, "agent", "reality_key_conflict", node.ID, map[string]any{"keyId": keyID})
				}
			}
		}
	}
	keyID := ""
	if node.Reality != nil {
		keyID = node.Reality.KeyID
	}
	if keyID != "" && d.KeyStore != nil {
		if entry, ok := d.KeyStore[keyID]; ok && entry != nil {
			raw, err := marshalJSON(entry.PublicKey)
			if err == nil {
				artifacts["realityPublicKey"] = raw
			}
		}
	}
}

// ---------- deploy validation ----------

// orchJSNumberStr mirrors JS Number(string): "" is 0, unparsable is NaN.
func orchJSNumberStr(s string) float64 {
	t := strings.TrimSpace(s)
	if t == "" {
		return 0
	}
	f, err := strconv.ParseFloat(t, 64)
	if err != nil {
		return math.NaN()
	}
	return f
}

func versionPart(parts []string, i int) float64 {
	if i >= len(parts) {
		return math.NaN()
	}
	return orchJSNumberStr(parts[i])
}

func ValidateNodeDeploy(d *Data, node *Node) error {
	direct := !forwardNode(node)
	if direct {
		if herr := ValidateEntryProtocol(node.Protocol); herr != nil {
			return herr
		}
	} else {
		if herr := ValidateProtocolPair(node.Protocol, xbDeref(node.ExitProtocol)); herr != nil {
			return herr
		}
	}
	server := orchFindServer(d, node.ServerID)
	if server == nil {
		return orchNotFound("服务器")
	}
	var exitServer *Server
	if !direct {
		exitServer = orchFindServer(d, xbDeref(node.ExitServerID))
		if exitServer == nil {
			return orchNotFound("出口服务器")
		}
	}
	if len(activeNodeClients(d, node)) == 0 {
		return httpErr(400, "请先给节点分配至少一个客户再部署")
	}
	switch node.Protocol {
	case "vless-ws-tls", "hysteria2", "anytls":
		if server.TLSDomain == "" {
			return httpErr(400, "%s 需要先填写设备 TLS 域名并在节点上安装证书", node.Protocol)
		}
		certOK := false
		if server.Engine != nil {
			for _, c := range server.Engine.Certificates {
				if c == server.TLSDomain {
					certOK = true
					break
				}
			}
		}
		if !certOK {
			return httpErr(409, "入口证书未由 Agent 确认。在入口机运行 ng-agent update && ng-agent cert，证书申请成功后刷新设备状态")
		}
	}
	if node.Protocol == "anytls" {
		var agent *Agent
		for _, item := range d.Agents {
			if item.ServerID == server.ID && item.Status == "active" {
				agent = item
				break
			}
		}
		version := ""
		if agent != nil {
			version = agent.Version
		}
		parts := strings.Split(version, ".")
		p0, p1, p2 := versionPart(parts, 0), versionPart(parts, 1), versionPart(parts, 2)
		versionOK := p0 > 0 || (p0 == 0 && (p1 > 6 || (p1 == 6 && p2 >= 1)))
		if agent == nil || !versionOK {
			return httpErr(409, "AnyTLS 入口需要新版 Agent：请在 %s 运行 ng-agent update", server.Name)
		}
		singBoxInstalled := server.Engine != nil && server.Engine.SingBoxInstalled
		if !singBoxInstalled {
			return httpErr(409, "AnyTLS 入口缺少 sing-box 及统计组件：请在 %s 运行 ng-agent engine install，完成后等待 Agent 心跳并刷新页面", server.Name)
		}
	}
	servers := []*Server{server}
	if !direct {
		servers = []*Server{server, exitServer}
	}
	nowMs := time.Now().UnixMilli()
	unavailable := []*Server{}
	for _, s := range servers {
		stale := false
		if s.LastSeenAt != nil {
			if ms, ok := ParseISOMs(*s.LastSeenAt); ok && nowMs-ms > 120000 {
				stale = true
			}
		}
		if s.Status != "online" || stale {
			unavailable = append(unavailable, s)
		}
	}
	if len(unavailable) > 0 {
		parts := make([]string, len(unavailable))
		for i, s := range unavailable {
			parts[i] = fmt.Sprintf("%s（Agent 离线；请在该机器运行 ng-agent doctor）", s.Name)
		}
		return httpErr(409, "目标设备尚未就绪：%s", strings.Join(parts, "、"))
	}
	networkMode := node.NetworkMode
	if networkMode == "" {
		networkMode = "ipv4"
	}
	if networkMode == "ipv6" {
		missing := []*Server{}
		for _, s := range servers {
			if s.PublicAddressV6 == "" {
				missing = append(missing, s)
			}
		}
		if len(missing) > 0 {
			names := make([]string, len(missing))
			for i, s := range missing {
				names[i] = s.Name
			}
			return httpErr(400, "IPv6 模式需要先为这些服务器填写 IPv6 地址：%s", strings.Join(names, "、"))
		}
	}
	if NodePortInUse(d, node.ServerID, node.Port, node.ID) {
		return httpErr(409, "端口 %d 已被该服务器上的其他节点占用", node.Port)
	}
	if !direct && node.ExitPort != nil && NodePortInUse(d, xbDeref(node.ExitServerID), *node.ExitPort, node.ID) {
		return httpErr(409, "出口端口 %d 已被该服务器上的其他节点占用", *node.ExitPort)
	}
	return nil
}

// ---------- reconciliation ----------

type ReconcileOptions struct {
	OnlyNodeIDs  map[string]bool
	FailInFlight *bool // nil mirrors the JS default: true
}

func reconcileServerData(d *Data, serverID, actor string, opts ReconcileOptions) (int, error) {
	server := orchFindServer(d, serverID)
	if server == nil {
		return 0, orchNotFound("服务器")
	}
	failInFlight := true
	if opts.FailInFlight != nil {
		failInFlight = *opts.FailInFlight
	}
	if failInFlight {
		for _, job := range d.Jobs {
			if job.ServerID == serverID && (job.Status == "queued" || job.Status == "running") {
				job.Status = "failed"
				job.Error = StrPtr("Agent 重新注册，旧任务已由对账任务替代")
				job.UpdatedAt = NowISO()
			}
		}
	}
	count := 0
	for _, node := range d.Nodes {
		if node.ServerID != serverID && xbDeref(node.ExitServerID) != serverID {
			continue
		}
		if opts.OnlyNodeIDs != nil && !opts.OnlyNodeIDs[node.ID] {
			continue
		}
		isEntry := node.ServerID == serverID
		isExit := xbDeref(node.ExitServerID) == serverID && forwardNode(node)
		if !isEntry && !isExit {
			continue
		}
		if node.Status == "draft" {
			continue
		}
		if node.Status == "removing" {
			if isEntry {
				QueueJob(d, serverID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ResourceID, Kind: "node"})
			}
			if isExit && node.ExitResourceID != "" {
				QueueJob(d, serverID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ExitResourceID, Kind: "node-exit"})
			}
		} else {
			clients := activeNodeClients(d, node)
			if len(clients) == 0 {
				continue
			}
			if isEntry {
				resource, err := resourceForNode(d, node, clients)
				if err != nil {
					return count, err
				}
				QueueJob(d, serverID, node.ID, "apply_resource", payloadWithResource(resource, "node", KeyMaterialForNode(d, node)))
			}
			if isExit {
				resource, err := resourceForNodeExit(d, node)
				if err != nil {
					return count, err
				}
				QueueJob(d, serverID, node.ID, "apply_resource", payloadWithResource(resource, "node-exit", nil))
			}
			node.Status = "deploying"
		}
		node.UpdatedAt = NowISO()
		count++
	}
	Audit(d, actor, "reconcile_server", server.ID, map[string]any{"nodes": count})
	return count, nil
}

// ---------- Orchestrator ----------

type Orchestrator struct {
	store *Store

	mu                   sync.Mutex
	heartbeatReconcileAt map[string]int64
}

// tx adapts the store's (any, error) transaction signature to the
// error-only closures used by the orchestrator methods.
func (o *Orchestrator) tx(fn func(d *Data) error) error {
	_, err := o.store.Transaction(func(d *Data) (any, error) { return nil, fn(d) })
	return err
}

func (o *Orchestrator) transient(fn func(d *Data) error) error {
	o.store.Transient(func(d *Data) { _ = fn(d) })
	return nil
}

func NewOrchestrator(store *Store) *Orchestrator {
	return &Orchestrator{store: store, heartbeatReconcileAt: map[string]int64{}}
}

func cloneNodeRecord(n *Node) *Node {
	raw, err := marshalJSON(n)
	if err != nil {
		panic(err)
	}
	var out Node
	if err := unmarshalJSONUseNumber(raw, &out); err != nil {
		panic(err)
	}
	return &out
}

func cloneNodeClientRecord(nc *NodeClient) *NodeClient {
	raw, err := marshalJSON(nc)
	if err != nil {
		panic(err)
	}
	var out NodeClient
	if err := unmarshalJSONUseNumber(raw, &out); err != nil {
		panic(err)
	}
	return &out
}

func cloneJobRecord(j *Job) *Job {
	raw, err := marshalJSON(j)
	if err != nil {
		panic(err)
	}
	var out Job
	if err := unmarshalJSONUseNumber(raw, &out); err != nil {
		panic(err)
	}
	return &out
}

func (o *Orchestrator) DeployNode(nodeID, actor string) (*Node, error) {
	if actor == "" {
		actor = "admin"
	}
	var result *Node
	err := o.tx(func(d *Data) error {
		node := orchFindNode(d, nodeID)
		if node == nil {
			return orchNotFound("节点")
		}
		if err := ValidateNodeDeploy(d, node); err != nil {
			return err
		}
		if err := RebuildNode(d, node, true); err != nil {
			return err
		}
		Audit(d, actor, "deploy_node", node.ID, map[string]any{"name": node.Name})
		result = cloneNodeRecord(node)
		return nil
	})
	return result, err
}

func (o *Orchestrator) RemoveNode(nodeID, actor string, deleteRecord bool) (bool, error) {
	if actor == "" {
		actor = "admin"
	}
	err := o.tx(func(d *Data) error {
		node := orchFindNode(d, nodeID)
		if node == nil {
			return orchNotFound("节点")
		}
		node.PendingDelete = deleteRecord
		node.AutoRemoving = BoolPtr(false)
		if node.Status != "draft" {
			QueueJob(d, node.ServerID, node.ID, "delete_resource", &JobPayload{ResourceID: node.ResourceID, Kind: "node"})
			if forwardNode(node) && node.ExitResourceID != "" {
				QueueJob(d, xbDeref(node.ExitServerID), node.ID, "delete_resource", &JobPayload{ResourceID: node.ExitResourceID, Kind: "node-exit"})
			}
			node.Status = "removing"
			node.Error = nil
		} else if deleteRecord {
			DropRealityKey(d, node)
			kept := []*Node{}
			for _, n := range d.Nodes {
				if n.ID != node.ID {
					kept = append(kept, n)
				}
			}
			d.Nodes = kept
			keptClients := []*NodeClient{}
			for _, nc := range d.NodeClients {
				if nc.NodeID != node.ID {
					keptClients = append(keptClients, nc)
				}
			}
			d.NodeClients = keptClients
		}
		node.UpdatedAt = NowISO()
		action := "remove_node"
		if deleteRecord {
			action = "delete_node"
		}
		Audit(d, actor, action, node.ID, map[string]any{"name": node.Name})
		return nil
	})
	return err == nil, err
}

func (o *Orchestrator) RepairNode(nodeID, actor string) (bool, error) {
	if actor == "" {
		actor = "admin"
	}
	err := o.tx(func(d *Data) error {
		node := orchFindNode(d, nodeID)
		if node == nil {
			return orchNotFound("节点")
		}
		if node.Status != "failed" && node.Status != "degraded" {
			return httpErr(409, "节点状态正常，无需修复")
		}
		if err := RebuildNode(d, node, true); err != nil {
			return err
		}
		Audit(d, actor, "repair_node", node.ID, map[string]any{"name": node.Name})
		return nil
	})
	return err == nil, err
}

func (o *Orchestrator) AssignCustomer(nodeID, customerID, actor string) (*NodeClient, error) {
	if actor == "" {
		actor = "admin"
	}
	var result *NodeClient
	err := o.tx(func(d *Data) error {
		node := orchFindNode(d, nodeID)
		if node == nil {
			return orchNotFound("节点")
		}
		customer := orchFindCustomer(d, customerID)
		if customer == nil {
			return orchNotFound("客户")
		}
		if customer.Status != "active" {
			return httpErr(409, "只能给状态正常的客户分配节点")
		}
		var nc *NodeClient
		for _, entry := range d.NodeClients {
			if entry.NodeID == nodeID && entry.CustomerID == customerID {
				nc = entry
				break
			}
		}
		if nc != nil && nc.Status == "active" {
			return httpErr(409, "该客户已分配到此节点")
		}
		if nc != nil {
			nc.Status = "active"
			nc.UpdatedAt = NowISO()
		} else {
			now := NowISO()
			nc = &NodeClient{
				ID:          NewID("ncl"),
				NodeID:      nodeID,
				CustomerID:  customerID,
				Credentials: NewNodeClientCredentials(node.Protocol),
				Status:      "active",
				CreatedAt:   now,
				UpdatedAt:   now,
			}
			d.NodeClients = append(d.NodeClients, nc)
		}
		if (node.Status == "removing" || node.Status == "draft") && !boolVal(node.AutoRemoving) {
			// Record the assignment only.
		} else {
			if node.Status == "removing" && boolVal(node.AutoRemoving) {
				cancelQueuedDeletes(d, node.ID, "已取消：节点恢复服务")
				node.AutoRemoving = BoolPtr(false)
			}
			if err := RebuildNode(d, node, false); err != nil {
				return err
			}
		}
		Audit(d, actor, "assign_customer", node.ID, map[string]any{"customerId": customerID, "customer": customer.Name})
		result = cloneNodeClientRecord(nc)
		return nil
	})
	return result, err
}

func (o *Orchestrator) UnassignCustomer(nodeID, customerID, actor string) (bool, error) {
	if actor == "" {
		actor = "admin"
	}
	err := o.tx(func(d *Data) error {
		node := orchFindNode(d, nodeID)
		if node == nil {
			return orchNotFound("节点")
		}
		index := -1
		for i, entry := range d.NodeClients {
			if entry.NodeID == nodeID && entry.CustomerID == customerID {
				index = i
				break
			}
		}
		if index < 0 {
			return httpErr(404, "该客户未分配到此节点")
		}
		d.NodeClients = append(d.NodeClients[:index], d.NodeClients[index+1:]...)
		if node.Status != "draft" && node.Status != "removing" {
			if err := RebuildNode(d, node, false); err != nil {
				return err
			}
		}
		Audit(d, actor, "unassign_customer", node.ID, map[string]any{"customerId": customerID})
		return nil
	})
	return err == nil, err
}

func (o *Orchestrator) ReconcileServer(serverID, actor string) (int, error) {
	if actor == "" {
		actor = "system"
	}
	var count int
	err := o.tx(func(d *Data) error {
		var err error
		count, err = reconcileServerData(d, serverID, actor, ReconcileOptions{})
		return err
	})
	return count, err
}

func (o *Orchestrator) ReconcileFromHeartbeat(serverID string, reportedResourceIds []string) (int, error) {
	if reportedResourceIds == nil {
		return 0, nil
	}
	o.mu.Lock()
	last := o.heartbeatReconcileAt[serverID]
	o.mu.Unlock()
	if time.Now().UnixMilli()-last < HeartbeatReconcileIntervalMs {
		return 0, nil
	}
	reported := map[string]bool{}
	limit := len(reportedResourceIds)
	if limit > 1000 {
		limit = 1000
	}
	for _, v := range reportedResourceIds[:limit] {
		if v != "" {
			reported[v] = true
		}
	}
	triggered := false
	var count int
	err := o.tx(func(d *Data) error {
		server := orchFindServer(d, serverID)
		if server == nil {
			return nil
		}
		missing := map[string]bool{}
		for _, node := range d.Nodes {
			if node.Status == "draft" || node.Status == "removing" {
				continue
			}
			if len(activeNodeClients(d, node)) == 0 {
				continue
			}
			if node.ServerID == serverID && node.ResourceID != "" && !reported[node.ResourceID] {
				missing[node.ID] = true
			}
			if xbDeref(node.ExitServerID) == serverID && forwardNode(node) && node.ExitResourceID != "" && !reported[node.ExitResourceID] {
				missing[node.ID] = true
			}
		}
		if len(missing) == 0 {
			return nil
		}
		for _, job := range d.Jobs {
			if job.ServerID == serverID && (job.Status == "queued" || job.Status == "running") {
				return nil
			}
		}
		triggered = true
		fail := false
		var err error
		count, err = reconcileServerData(d, serverID, "system:heartbeat", ReconcileOptions{OnlyNodeIDs: missing, FailInFlight: &fail})
		return err
	})
	if err != nil {
		return 0, err
	}
	if triggered {
		o.mu.Lock()
		o.heartbeatReconcileAt[serverID] = time.Now().UnixMilli()
		o.mu.Unlock()
	}
	return count, nil
}

type CompleteJobResult struct {
	Artifacts map[string]json.RawMessage
}

func (o *Orchestrator) CompleteJob(jobID string, success bool, result CompleteJobResult, errorMessage *string) (*Job, error) {
	var out *Job
	err := o.tx(func(d *Data) error {
		var job *Job
		for _, item := range d.Jobs {
			if item.ID == jobID {
				job = item
				break
			}
		}
		if job == nil {
			return orchNotFound("任务")
		}
		if job.Status != "running" {
			if job.Status == "completed" && success {
				out = cloneJobRecord(job)
				return nil
			}
			return httpErr(409, "任务已被重试或取消，请等待当前任务完成")
		}
		now := NowISO()
		if success {
			job.Status = "completed"
		} else {
			job.Status = "failed"
		}
		job.UpdatedAt = now
		job.CompletedAt = StrPtr(now)
		job.LeaseUntil = nil
		if success {
			job.Error = nil
		} else {
			msg := "Agent operation failed"
			if errorMessage != nil {
				msg = *errorMessage
			}
			job.Error = StrPtr(xbTruncRunes(RedactSecrets(msg), 1800))
		}
		node := orchFindNode(d, job.NodeID)
		if node == nil {
			out = cloneJobRecord(job)
			return nil
		}
		kind := job.Payload.KindOf()
		switch {
		case success && job.Action == "apply_resource":
			if kind == "node-exit" {
				node.ExitReady = true
			} else {
				node.EntryReady = true
				artifacts := result.Artifacts
				if artifacts == nil {
					artifacts = map[string]json.RawMessage{}
				}
				node.Artifacts = artifacts
				MergeKeyMaterialBackfill(d, node, node.Artifacts)
			}
			DeriveNodeStatus(node)
		case success && job.Action == "delete_resource":
			if kind == "node-exit" {
				node.ExitReady = false
			} else {
				node.EntryReady = false
			}
			pending := false
			for _, item := range d.Jobs {
				if item.NodeID == node.ID && item.Action == "delete_resource" && (item.Status == "queued" || item.Status == "running") {
					pending = true
					break
				}
			}
			if !pending && node.Status == "removing" {
				if node.PendingDelete {
					DropRealityKey(d, node)
					kept := []*Node{}
					for _, n := range d.Nodes {
						if n.ID != node.ID {
							kept = append(kept, n)
						}
					}
					d.Nodes = kept
					keptClients := []*NodeClient{}
					for _, nc := range d.NodeClients {
						if nc.NodeID != node.ID {
							keptClients = append(keptClients, nc)
						}
					}
					d.NodeClients = keptClients
				} else {
					node.Status = "draft"
					node.Error = nil
				}
			}
		case !success:
			if kind == "node-exit" {
				node.Status = "degraded"
			} else {
				node.Status = "failed"
			}
			node.Error = job.Error
		}
		node.UpdatedAt = NowISO()
		out = cloneJobRecord(job)
		return nil
	})
	return out, err
}

// ---------- usage reporting ----------

type UsageSample struct {
	Kind     string      `json:"kind"`
	ID       string      `json:"id"`
	Uplink   json.Number `json:"uplink"`
	Downlink json.Number `json:"downlink"`
	Epoch    *string     `json:"epoch"`
}

// safeJSInt mirrors Number.isSafeInteger(v) && v >= 0 over a JSON number.
func safeJSInt(n json.Number) (int64, bool) {
	if n == "" {
		return 0, false
	}
	f, err := n.Float64()
	if err != nil {
		return 0, false
	}
	if f != math.Trunc(f) || f < 0 || math.Abs(f) > 9007199254740991 {
		return 0, false
	}
	return int64(f), true
}

func usageErrorPtr(errText string) *string {
	if errText == "" {
		return nil
	}
	return StrPtr(xbTruncRunes(errText, 240))
}

func (o *Orchestrator) RecordUsage(serverID string, samples []UsageSample, errText string, epoch string) error {
	if len(samples) == 0 {
		return o.transient(func(d *Data) error {
			server := orchFindServer(d, serverID)
			if server != nil {
				server.Usage = &ServerUsage{LastReportAt: NowISO(), SampleCount: 0, Error: usageErrorPtr(errText)}
			}
			return nil
		})
	}
	return o.tx(func(d *Data) error {
		reportedAt := NowISO()
		server := orchFindServer(d, serverID)
		if server != nil {
			server.Usage = &ServerUsage{LastReportAt: reportedAt, SampleCount: len(samples), Error: usageErrorPtr(errText)}
		}
		limit := len(samples)
		if limit > 1000 {
			limit = 1000
		}
		for _, sample := range samples[:limit] {
			uplink, okUp := safeJSInt(sample.Uplink)
			downlink, okDown := safeJSInt(sample.Downlink)
			if !okUp || !okDown {
				continue
			}
			var sampleEpoch *string
			if sample.Epoch != nil {
				sampleEpoch = StrPtr(xbTruncRunes(*sample.Epoch, 80))
			} else if epoch != "" {
				sampleEpoch = StrPtr(epoch)
			}
			switch sample.Kind {
			case "user":
				customer := orchFindCustomer(d, sample.ID)
				if customer == nil {
					continue
				}
				key := "srv:" + serverID
				if customer.ServerUsage == nil {
					customer.ServerUsage = map[string]*UsageSlot{}
				}
				slot := customer.ServerUsage[key]
				if slot == nil {
					slot = &UsageSlot{}
					customer.ServerUsage[key] = slot
				}
				delta := ApplyDelta(SlotHolder(slot), uplink, downlink, sampleEpoch)
				if delta == nil {
					continue
				}
				customer.UsedUplinkBytes += delta.UpDelta
				customer.UsedDownlinkBytes += delta.DownDelta
				customer.UsedBytes += delta.UpDelta + delta.DownDelta
				customer.LastUsageAt = StrPtr(reportedAt)
				customer.UpdatedAt = reportedAt
			case "inbound":
				node := orchFindNode(d, sample.ID)
				if node == nil || node.ServerID != serverID {
					continue
				}
				delta := ApplyDelta(NodeHolder(node), uplink, downlink, sampleEpoch)
				if delta == nil {
					continue
				}
				metered := node.MeteredTraffic
				var mu, md int64
				if metered != nil {
					mu, md = metered.Uplink, metered.Downlink
				}
				node.MeteredTraffic = &MeteredTraffic{Uplink: mu + delta.UpDelta, Downlink: md + delta.DownDelta}
				node.LastUsageAt = StrPtr(reportedAt)
			}
		}
		return nil
	})
}