l-ui/internal/web/service/node.go

package service

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/mhsanaei/3x-ui/v3/internal/config"
	"github.com/mhsanaei/3x-ui/v3/internal/database"
	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
	"github.com/mhsanaei/3x-ui/v3/internal/logger"
	"github.com/mhsanaei/3x-ui/v3/internal/selfupdate"
	"github.com/mhsanaei/3x-ui/v3/internal/util/common"
	"github.com/mhsanaei/3x-ui/v3/internal/util/netsafe"

	"gorm.io/gorm"
)

type HeartbeatPatch struct {
	Status        string
	LastHeartbeat int64
	LatencyMs     int
	XrayVersion   string
	PanelVersion  string
	Guid          string
	CpuPct        float64
	MemPct        float64
	UptimeSecs    uint64
	// NetUp/NetDown are the node's current interface throughput (bytes/sec),
	// summed over non-virtual interfaces, read from its status response.
	NetUp     uint64
	NetDown   uint64
	LastError string
	// XrayState and XrayError come from the remote /panel/api/server/status when the
	// panel API is reachable. They allow distinguishing panel connectivity from
	// Xray core health on the node.
	XrayState string
	XrayError string
}

type NodeService struct{}

func (s *NodeService) GetAll() ([]*model.Node, error) {
	db := database.GetDB()
	var nodes []*model.Node
	err := db.Model(model.Node{}).Order("id asc").Find(&nodes).Error
	if err != nil || len(nodes) == 0 {
		return nodes, err
	}

	type inboundRow struct {
		Id     int
		NodeID int `gorm:"column:node_id"`
	}
	var inboundRows []inboundRow
	if err := db.Table("inbounds").
		Select("id, node_id").
		Where("node_id IS NOT NULL").
		Scan(&inboundRows).Error; err != nil {
		return nodes, nil
	}
	if len(inboundRows) == 0 {
		return nodes, nil
	}
	inboundsByNode := make(map[int][]int, len(nodes))
	for _, row := range inboundRows {
		inboundsByNode[row.NodeID] = append(inboundsByNode[row.NodeID], row.Id)
	}

	type clientCountRow struct {
		NodeID int `gorm:"column:node_id"`
		Count  int `gorm:"column:count"`
	}
	var clientCounts []clientCountRow
	if err := db.Raw(`
		SELECT inbounds.node_id AS node_id, COUNT(DISTINCT client_inbounds.client_id) AS count
		FROM inbounds
		JOIN client_inbounds ON client_inbounds.inbound_id = inbounds.id
		WHERE inbounds.node_id IS NOT NULL
		GROUP BY inbounds.node_id
	`).Scan(&clientCounts).Error; err == nil {
		for _, row := range clientCounts {
			for _, n := range nodes {
				if n.Id == row.NodeID {
					n.ClientCount = row.Count
					break
				}
			}
		}
	}

	depletedByNode := make(map[int]int)
	disabledByNode := make(map[int]int)
	activeByNode := make(map[int]int)
	// Email lists ride along with the counts, appended in the same pass and
	// under the same dedupe, so a category's hover list always names exactly
	// the clients its count tallied.
	depletedEmailsByNode := make(map[int][]string)
	disabledEmailsByNode := make(map[int][]string)
	activeEmailsByNode := make(map[int][]string)
	statuses, _ := s.nodeClientStatuses()
	seen := make(map[int]map[int]struct{}, len(nodes))
	for _, st := range statuses {
		clientsSeen := seen[st.NodeID]
		if clientsSeen == nil {
			clientsSeen = make(map[int]struct{})
			seen[st.NodeID] = clientsSeen
		}
		if _, dup := clientsSeen[st.ClientID]; dup {
			// A client attached to several inbounds of one node counts once,
			// matching the distinct ClientCount above.
			continue
		}
		clientsSeen[st.ClientID] = struct{}{}
		switch {
		case st.Depleted:
			depletedByNode[st.NodeID]++
			if st.Email != "" {
				depletedEmailsByNode[st.NodeID] = append(depletedEmailsByNode[st.NodeID], st.Email)
			}
		case st.Disabled:
			disabledByNode[st.NodeID]++
			if st.Email != "" {
				disabledEmailsByNode[st.NodeID] = append(disabledEmailsByNode[st.NodeID], st.Email)
			}
		default:
			activeByNode[st.NodeID]++
			if st.Email != "" {
				activeEmailsByNode[st.NodeID] = append(activeEmailsByNode[st.NodeID], st.Email)
			}
		}
	}
	onlineByGuid := s.onlineEmailsByGuid()
	selfGuid, _ := (&SettingService{}).GetPanelGuid()
	ambiguous := ambiguousNodeGuids(nodes, selfGuid)
	for _, n := range nodes {
		n.InboundCount = len(inboundsByNode[n.Id])
		n.DepletedCount = depletedByNode[n.Id]
		n.DisabledCount = disabledByNode[n.Id]
		n.ActiveCount = activeByNode[n.Id]
		n.DepletedEmails = depletedEmailsByNode[n.Id]
		n.DisabledEmails = disabledEmailsByNode[n.Id]
		n.ActiveEmails = activeEmailsByNode[n.Id]
		// Online is attributed to the node that physically hosts the client
		// (by GUID): a client on a sub-node counts under the sub-node, not
		// the intermediate node it syncs through (#4983).
		onlineSet := onlineByGuid[effectiveNodeGuid(n, ambiguous)]
		n.OnlineCount = len(onlineSet)
		n.OnlineEmails = sortedEmailSet(onlineSet)
	}

	return nodes, nil
}

// nodeClientStatus is one node-hosted client's classification, carrying enough
// identity for callers to bucket it by node id or by attribution GUID.
type nodeClientStatus struct {
	InboundID int
	NodeID    int
	ClientID  int
	Email     string
	Depleted  bool
	Disabled  bool
}

// nodeClientStatuses classifies every client attached to a node-hosted inbound as
// depleted / disabled / active, matching client_traffics by EMAIL rather than by
// inbound_id. client_traffics.inbound_id goes stale after an inbound is
// delete+recreated, so filtering by it silently drops most rows; the
// client_inbounds -> clients join is the reliable client set and the email join
// pulls each client's live counters. Precedence matches the inbound page:
// depleted (expired/exhausted) wins over disabled.
func (s *NodeService) nodeClientStatuses() ([]nodeClientStatus, error) {
	type row struct {
		InboundID  int    `gorm:"column:inbound_id"`
		NodeID     int    `gorm:"column:node_id"`
		ClientID   int    `gorm:"column:client_id"`
		Email      string `gorm:"column:email"`
		Enable     bool   `gorm:"column:enable"`
		Total      int64  `gorm:"column:total"`
		Up         int64  `gorm:"column:up"`
		Down       int64  `gorm:"column:down"`
		ExpiryTime int64  `gorm:"column:expiry_time"`
	}
	var rows []row
	if err := database.GetDB().Table("inbounds").
		Select("inbounds.id AS inbound_id, inbounds.node_id AS node_id, clients.id AS client_id, " +
			"clients.email AS email, clients.enable AS enable, ct.total AS total, ct.up AS up, ct.down AS down, ct.expiry_time AS expiry_time").
		Joins("JOIN client_inbounds ON client_inbounds.inbound_id = inbounds.id").
		Joins("JOIN clients ON clients.id = client_inbounds.client_id").
		Joins("LEFT JOIN client_traffics ct ON ct.email = clients.email").
		Where("inbounds.node_id IS NOT NULL").
		Scan(&rows).Error; err != nil {
		return nil, err
	}
	now := time.Now().UnixMilli()
	out := make([]nodeClientStatus, 0, len(rows))
	for _, r := range rows {
		st := nodeClientStatus{InboundID: r.InboundID, NodeID: r.NodeID, ClientID: r.ClientID, Email: r.Email}
		expired := r.ExpiryTime > 0 && r.ExpiryTime <= now
		exhausted := r.Total > 0 && r.Up+r.Down >= r.Total
		switch {
		case expired || exhausted:
			st.Depleted = true
		case !r.Enable:
			st.Disabled = true
		}
		out = append(out, st)
	}
	return out, nil
}

func (s *NodeService) onlineEmailsByGuid() map[string]map[string]struct{} {
	svc := InboundService{}
	byGuid := svc.GetOnlineClientsByGuid()
	out := make(map[string]map[string]struct{}, len(byGuid))
	for guid, emails := range byGuid {
		set := make(map[string]struct{}, len(emails))
		for _, email := range emails {
			set[email] = struct{}{}
		}
		out[guid] = set
	}
	return out
}

// sortedEmailSet flattens an email set into a deterministic list for the
// node views' hover popovers. Nil for an empty set.
func sortedEmailSet(set map[string]struct{}) []string {
	if len(set) == 0 {
		return nil
	}
	out := make([]string, 0, len(set))
	for email := range set {
		out = append(out, email)
	}
	slices.Sort(out)
	return out
}

// effectiveNodeGuid is a node's stable online/inbound attribution key: its
// reported panelGuid, or a master-local synthetic node-id fallback when the node
// has no GUID yet (old build) or its GUID is ambiguous. ambiguous comes from
// ambiguousNodeGuids.
func effectiveNodeGuid(n *model.Node, ambiguous map[string]struct{}) string {
	if n.Guid == "" {
		return synthNodeGuid(n.Id)
	}
	if n.Id > 0 {
		if _, bad := ambiguous[n.Guid]; bad {
			return synthNodeGuid(n.Id)
		}
	}
	return n.Guid
}

// ambiguousNodeGuids returns the panelGuids a node must not be attributed under
// directly, because doing so would merge two distinct identities: a GUID
// reported by more than one of this master's direct nodes (cloned node servers
// ship the same panelGuid in their copied settings), or a GUID equal to the
// master's own panelGuid (a node cloned from the master). A node holding such a
// GUID falls back to its node-unique synthNodeGuid. Transitive sub-nodes (Id 0)
// carry distinct descendant GUIDs by construction and are excluded.
func ambiguousNodeGuids(nodes []*model.Node, selfGuid string) map[string]struct{} {
	counts := make(map[string]int, len(nodes))
	for _, n := range nodes {
		if n.Id > 0 && n.Guid != "" {
			counts[n.Guid]++
		}
	}
	ambiguous := make(map[string]struct{})
	for guid, c := range counts {
		if c > 1 {
			ambiguous[guid] = struct{}{}
		}
	}
	if selfGuid != "" {
		if _, ok := counts[selfGuid]; ok {
			ambiguous[selfGuid] = struct{}{}
		}
	}
	return ambiguous
}

// effectiveNodeKey returns one node's attribution key without a preloaded node
// list — its panelGuid when that GUID uniquely identifies it among the master's
// nodes and differs from the master's own, otherwise its node-unique
// synthNodeGuid. Same rule as effectiveNodeGuid + ambiguousNodeGuids, for the
// write paths that handle a single node (online tree, IP attribution).
func effectiveNodeKey(node *model.Node) string {
	if node == nil {
		return ""
	}
	if node.Guid == "" {
		return synthNodeGuid(node.Id)
	}
	var sameGuid int64
	database.GetDB().Model(&model.Node{}).Where("guid = ?", node.Guid).Count(&sameGuid)
	masterGuid, _ := (&SettingService{}).GetPanelGuid()
	if sameGuid > 1 || node.Guid == masterGuid {
		return synthNodeGuid(node.Id)
	}
	return node.Guid
}

func (s *NodeService) GetById(id int) (*model.Node, error) {
	db := database.GetDB()
	n := &model.Node{}
	if err := db.Model(model.Node{}).Where("id = ?", id).First(n).Error; err != nil {
		return nil, err
	}
	return n, nil
}

func (s *NodeService) GetViewById(id int) (*NodeView, error) {
	n, err := s.GetById(id)
	if err != nil {
		return nil, err
	}
	view := toNodeView(n)
	views := []*NodeView{view}
	annotateNodeTypes(views)
	annotateAgentTelemetry(views)
	return view, nil
}

// NodeExists reports whether a node with the given id exists on this panel.
// Used to drop stale, cross-panel node references on inbound import. A Count
// query distinguishes "no such node" (count 0, no error) from a real DB error.
func (s *NodeService) NodeExists(id int) (bool, error) {
	if id <= 0 {
		return false, nil
	}
	var count int64
	if err := database.GetDB().Model(model.Node{}).Where("id = ?", id).Count(&count).Error; err != nil {
		return false, err
	}
	return count > 0, nil
}

func normalizeBasePath(p string) string {
	p = strings.TrimSpace(p)
	if p == "" {
		return "/"
	}
	if !strings.HasPrefix(p, "/") {
		p = "/" + p
	}
	if !strings.HasSuffix(p, "/") {
		p = p + "/"
	}
	return p
}

func (s *NodeService) normalize(n *model.Node) error {
	return s.normalizeMode(n, false)
}

// normalizeMode validates a node row before it is stored. agentEndpoint
// marks an agent-node write: agent nodes have no panel endpoint to dial,
// so an unset address/port pair (the form placeholder, sanitized away by
// sanitizeAgentEndpoint) is legitimate and skips the host/port checks.
// Any value that IS present is validated exactly like a traditional
// node's.
func (s *NodeService) normalizeMode(n *model.Node, agentEndpoint bool) error {
	n.Name = strings.TrimSpace(n.Name)
	// The per-node ApiToken column outlives its write path (legacy
	// rows, no schema change in B6); keep its stored form trimmed.
	n.ApiToken = strings.TrimSpace(n.ApiToken)
	if n.Name == "" {
		return common.NewError("node name is required")
	}
	if !agentEndpoint || n.Address != "" {
		addr, err := netsafe.NormalizeHost(n.Address)
		if err != nil {
			return common.NewError(err.Error())
		}
		n.Address = addr
	}
	if !agentEndpoint || n.Port != 0 {
		if n.Port <= 0 || n.Port > 65535 {
			return common.NewError("node port must be 1-65535")
		}
	}
	if n.Scheme != "http" && n.Scheme != "https" {
		n.Scheme = "https"
	}
	n.BasePath = normalizeBasePath(n.BasePath)
	return nil
}

func (s *NodeService) Create(n *model.Node) error {
	return s.create(n, false)
}

func (s *NodeService) create(n *model.Node, agentEndpoint bool) error {
	if err := s.normalizeMode(n, agentEndpoint); err != nil {
		return err
	}
	return database.GetDB().Create(n).Error
}

func (s *NodeService) CreateFromRequest(req *NodeMutationRequest) (*NodeView, error) {
	if err := req.validate(); err != nil {
		return nil, err
	}
	// Every node is agent-managed: a brand-new node has no binding yet,
	// so no resolved address can be known — the placeholder sanitizes
	// to empty and is filled in by later writes once the agent reports.
	sanitizeAgentEndpoint(req, "")
	n := req.toNode()
	if err := s.create(n, true); err != nil {
		return nil, err
	}
	return toNodeView(n), nil
}

// resolvedAgentAddress returns the externally usable address of the agent
// bound to a node (latest binding wins, via ResolveAgentAddresses), or ""
// when the node has no bound agent or the agent never reported one. The
// stored row of an agent node carries this value (or empty) instead of
// the create form's placeholder — see sanitizeAgentEndpoint.
func (s *NodeService) resolvedAgentAddress(nodeID int) string {
	var ag model.AgentNode
	if err := database.GetDB().
		Where("node_id = ?", nodeID).
		Order("last_seen DESC").
		First(&ag).Error; err != nil {
		return ""
	}
	v4, v6 := ResolveAgentAddresses(&ag)
	if v4 != "" {
		return v4
	}
	return v6
}

func (s *NodeService) Update(id int, in *model.Node) error {
	if err := s.normalize(in); err != nil {
		return err
	}
	db := database.GetDB()
	if err := db.Where("id = ?", id).First(&model.Node{}).Error; err != nil {
		return err
	}
	updates := map[string]any{
		"name":                  in.Name,
		"remark":                in.Remark,
		"scheme":                in.Scheme,
		"address":               in.Address,
		"port":                  in.Port,
		"base_path":             in.BasePath,
		"enable":                in.Enable,
		"allow_private_address": in.AllowPrivateAddress,
	}
	return db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
			return err
		}
		return s.MarkNodeDirtyTx(tx, id)
	})
}

func (s *NodeService) UpdateFromRequest(id int, req *NodeMutationRequest) error {
	if err := req.validate(); err != nil {
		return err
	}
	sanitizeAgentEndpoint(req, s.resolvedAgentAddress(id))
	in := req.toNode()
	if err := s.normalizeMode(in, true); err != nil {
		return err
	}
	db := database.GetDB()
	if err := db.Where("id = ?", id).First(&model.Node{}).Error; err != nil {
		return err
	}
	updates := map[string]any{
		"name":                  in.Name,
		"remark":                in.Remark,
		"scheme":                in.Scheme,
		"address":               in.Address,
		"port":                  in.Port,
		"base_path":             in.BasePath,
		"enable":                in.Enable,
		"allow_private_address": in.AllowPrivateAddress,
	}
	if err := db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
			return err
		}
		// Probe-only (V2.1.0) lives on the bound agent rows, not the
		// node row: flip every binding of this node in the same
		// transaction, so the flag and the node edit land together.
		// Nil means the form didn't touch the switch.
		if req.ProbeOnly != nil {
			if err := tx.Model(&model.AgentNode{}).
				Where("node_id = ?", id).
				Update("probe_only", *req.ProbeOnly).Error; err != nil {
				return err
			}
		}
		return s.MarkNodeDirtyTx(tx, id)
	}); err != nil {
		return err
	}
	return nil
}

func (s *NodeService) Delete(id int) error {
	db := database.GetDB()
	// Refuse to delete a node that still owns inbounds: dropping the node row
	// while inbounds keep its node_id leaves orphaned, dangling references that
	// confuse node sync, subscriptions and cleanup. The operator must detach or
	// remove those inbounds first. (DB-002)
	var attached int64
	if err := db.Model(&model.Inbound{}).Where("node_id = ?", id).Count(&attached).Error; err != nil {
		return err
	}
	if attached > 0 {
		return common.NewError(fmt.Sprintf("cannot delete node: %d inbound(s) still attached to it; detach or delete them first", attached))
	}
	// Capture the node's guid before deleting the row so we can drop its per-node
	// IP attribution. NodeClientIp is keyed by the node's attribution key, which
	// is its guid normally but its node-unique key for a cloned/ambiguous-guid
	// node (see effectiveNodeKey) — so we purge both below.
	var guid string
	var n model.Node
	if err := db.Select("guid").Where("id = ?", id).First(&n).Error; err == nil {
		guid = n.Guid
	}
	// Delete the node row and its per-node child rows atomically. Remove the
	// children (traffic baselines, IP attribution) before the parent node row so
	// the ordering already matches a future ON DELETE constraint. Delete stays
	// tolerant of a missing node row so it can still clean up orphaned baselines.
	if err := db.Transaction(func(tx *gorm.DB) error {
		if err := tx.Where("node_id = ?", id).Delete(&model.NodeClientTraffic{}).Error; err != nil {
			return err
		}
		guids := []string{synthNodeGuid(id)}
		if guid != "" {
			guids = append(guids, guid)
		}
		if err := tx.Where("node_guid IN ?", guids).Delete(&model.NodeClientIp{}).Error; err != nil {
			return err
		}
		return tx.Where("id = ?", id).Delete(&model.Node{}).Error
	}); err != nil {
		return err
	}
	nodeMetrics.drop(nodeMetricKey(id, "cpu"))
	nodeMetrics.drop(nodeMetricKey(id, "mem"))
	return nil
}

func (s *NodeService) SetEnable(id int, enable bool) error {
	db := database.GetDB()
	if err := db.Model(model.Node{}).Where("id = ?", id).Update("enable", enable).Error; err != nil {
		return err
	}
	return nil
}

// NodeUpdateResult reports the outcome of triggering a panel self-update on one
// node so the UI can show per-node success/failure for a bulk request.
type NodeUpdateResult struct {
	Id    int    `json:"id"`
	Name  string `json:"name"`
	OK    bool   `json:"ok"`
	Error string `json:"error,omitempty"`
}

// fetchLatestReleaseTag resolves the newest release tag of the fork's own
// repository; a package variable so tests can stub the network.
var fetchLatestReleaseTag = func(ctx context.Context) (string, error) {
	return selfupdate.FetchLatestReleaseTag(ctx, &http.Client{Timeout: 20 * time.Second})
}

// requestAgentUpdate records a one-click update intent for an agent node
// (item 20). The agent picks the intent up from its next config poll,
// downloads + verifies the release binary for the latest tag, swaps and
// restarts itself; the outcome comes back on heartbeats. The dev channel
// has no meaning for agents and is ignored.
func (s *NodeService) requestAgentUpdate(ctx context.Context, n *model.Node) error {
	db := database.GetDB()
	var ag model.AgentNode
	if err := db.Where("node_id = ?", n.Id).Order("last_seen DESC, id DESC").First(&ag).Error; err != nil {
		return errors.New("no agent bound to this node")
	}
	tag, err := fetchLatestReleaseTag(ctx)
	if err != nil {
		return fmt.Errorf("check latest release: %w", err)
	}
	parts, ok := config.ParseDisplayVersion(tag)
	if !ok {
		return fmt.Errorf("latest release tag %q is not a V-series version", tag)
	}
	target := fmt.Sprintf("A%d.%d.%d", parts[0], parts[1], parts[2])
	if cmp, ok := config.CompareDisplayVersion(ag.AgentVersion, target); ok && cmp >= 0 {
		return fmt.Errorf("agent already up to date (%s)", ag.AgentVersion)
	}
	now := time.Now().UnixMilli()
	return db.Model(&model.AgentNode{}).Where("id = ?", ag.Id).Updates(map[string]any{
		"update_requested_at": now,
		"update_target":       target,
		"update_tag":          tag,
		"update_state":        "pending",
		"update_error":        "",
		"update_state_at":     now,
	}).Error
}

// UpdatePanels triggers the official self-updater on each given node. Only
// enabled, online nodes are eligible — an offline node can't be reached, so it
// is reported as skipped rather than silently dropped.
//
// Agent nodes take the item-20 intent path instead (see requestAgentUpdate):
// the master cannot push to them, so "triggered" means the intent was
// recorded for the agent's next poll.
func (s *NodeService) UpdatePanels(ids []int, dev bool) ([]NodeUpdateResult, error) {
	results := make([]NodeUpdateResult, 0, len(ids))
	for _, id := range ids {
		n, err := s.GetById(id)
		if err != nil || n == nil {
			results = append(results, NodeUpdateResult{Id: id, OK: false, Error: "node not found"})
			continue
		}
		res := NodeUpdateResult{Id: id, Name: n.Name}
		switch {
		case !n.Enable:
			res.Error = "node is disabled"
		case n.Status != "online":
			res.Error = "node is offline"
		case s.IsAgentManaged(id):
			ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
			err := s.requestAgentUpdate(ctx, n)
			cancel()
			if err != nil {
				res.Error = err.Error()
			} else {
				res.OK = true
			}
		default:
			res.Error = "node is not agent-managed"
		}
		results = append(results, res)
	}
	return results, nil
}

func (s *NodeService) UpdateHeartbeat(id int, p HeartbeatPatch) error {
	db := database.GetDB()
	updates := map[string]any{
		"status":         p.Status,
		"last_heartbeat": p.LastHeartbeat,
		"latency_ms":     p.LatencyMs,
		"xray_version":   p.XrayVersion,
		"panel_version":  p.PanelVersion,
		"cpu_pct":        p.CpuPct,
		"mem_pct":        p.MemPct,
		"uptime_secs":    p.UptimeSecs,
		"net_up":         p.NetUp,
		"net_down":       p.NetDown,
		"last_error":     p.LastError,
		"xray_state":     p.XrayState,
		"xray_error":     p.XrayError,
	}
	// Only learn the GUID; never clear a known one if an old-build node (or a
	// failed probe) reports none, so the stable identity survives blips.
	if p.Guid != "" {
		updates["guid"] = p.Guid
		s.warnOnDuplicateGuid(id, p.Guid)
	}
	if err := db.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
		return err
	}
	if p.Status == "online" {
		now := time.Unix(p.LastHeartbeat, 0)
		nodeMetrics.append(nodeMetricKey(id, "cpu"), now, p.CpuPct)
		nodeMetrics.append(nodeMetricKey(id, "mem"), now, p.MemPct)
		nodeMetrics.append(nodeMetricKey(id, "netUp"), now, float64(p.NetUp))
		nodeMetrics.append(nodeMetricKey(id, "netDown"), now, float64(p.NetDown))
	}
	return nil
}

// warnedDupGuid remembers the (nodeID -> guid) pairs already warned about so a
// cloned-server collision is logged once, not every heartbeat.
var warnedDupGuid sync.Map

// warnOnDuplicateGuid logs once when a node reports a panelGuid already held by
// another node or by the master itself (the cloned-server footgun). Attribution
// still works — it falls back to node-unique keys — but the operator should
// regenerate the duplicate panelGuid to restore real identity and per-node IP
// attribution. Re-arms if the collision later clears.
func (s *NodeService) warnOnDuplicateGuid(id int, guid string) {
	var clash int64
	database.GetDB().Model(&model.Node{}).Where("guid = ? AND id <> ?", guid, id).Count(&clash)
	masterGuid, _ := (&SettingService{}).GetPanelGuid()
	if clash == 0 && guid != masterGuid {
		warnedDupGuid.Delete(id)
		return
	}
	if prev, ok := warnedDupGuid.Load(id); ok && prev == guid {
		return
	}
	warnedDupGuid.Store(id, guid)
	logger.Warningf("node %d reports panelGuid %s already used by another node or the master (cloned server?) — regenerate it on that node so online and IP attribution stay per-node", id, guid)
}

func (s *NodeService) MarkNodeDirty(id int) error {
	return s.MarkNodeDirtyTx(database.GetDB(), id)
}

func (s *NodeService) MarkNodeDirtyTx(tx *gorm.DB, id int) error {
	if id <= 0 {
		return nil
	}
	if tx == nil {
		return errors.New("nil db transaction")
	}
	return tx.Model(model.Node{}).
		Where("id = ?", id).
		Updates(map[string]any{
			"config_dirty":    true,
			"config_dirty_at": time.Now().UnixMilli(),
		}).Error
}

func (s *NodeService) ClearNodeDirty(id int, dirtyAt int64) error {
	if id <= 0 {
		return nil
	}
	return database.GetDB().Model(model.Node{}).
		Where("id = ? AND config_dirty_at = ?", id, dirtyAt).
		Update("config_dirty", false).Error
}

func (s *NodeService) NodeSyncState(id int) (enabled bool, status string, dirty bool, dirtyAt int64, err error) {
	if id <= 0 {
		return false, "", false, 0, errors.New("invalid node id")
	}
	var row model.Node
	err = database.GetDB().Model(model.Node{}).
		Select("enable", "status", "config_dirty", "config_dirty_at").
		Where("id = ?", id).
		First(&row).Error
	if err != nil {
		return false, "", false, 0, err
	}
	return row.Enable, row.Status, row.ConfigDirty, row.ConfigDirtyAt, nil
}

// IsNodePending reports whether a save targeting this node was deferred because
// the node is unreachable right now — offline or disabled — so the edit only
// reaches it on the next reconcile. It deliberately ignores config_dirty: that
// flag is set on EVERY node-backed edit as the reconcile self-heal marker,
// including edits pushed live to an online node, so keying the user-facing
// "saved, node offline, will sync" toast off it fired the warning on every save
// to a perfectly healthy online node.
func (s *NodeService) IsNodePending(id int) bool {
	enabled, status, _, _, err := s.NodeSyncState(id)
	if err != nil {
		return false
	}
	return !enabled || status != "online"
}

func nodeMetricKey(id int, metric string) string {
	return "node:" + strconv.Itoa(id) + ":" + metric
}

func (s *NodeService) AggregateNodeMetric(id int, metric string, bucketSeconds int, maxPoints int) []map[string]any {
	return nodeMetrics.aggregate(nodeMetricKey(id, metric), bucketSeconds, maxPoints)
}

// agentOnlineWindow is how fresh an agent heartbeat must be for its
// node to read online — the same 180s the heartbeat job expires agent
// nodes by (agentHeartbeatStaleAfter in internal/web/job: max(3
// heartbeat intervals, 120s), so two lost heartbeats don't flap).
const agentOnlineWindow = 180 * time.Second

var errAgentOffline = errors.New("agent node offline: no fresh heartbeat")

// agentHeartbeatVerdict is the pure decision behind the agent-node
// probe: a fresh stamp reads online with the heartbeat AGE as the
// latency (there is no dial to time); a stale or never-seen stamp
// reads offline with LatencyMs -1. LastError stays empty either way —
// the answer comes from the heartbeat, not from a dial failure.
func agentHeartbeatVerdict(lastSeenMs int64, now time.Time) (HeartbeatPatch, bool) {
	patch := HeartbeatPatch{LatencyMs: -1}
	if lastSeenMs <= 0 {
		return patch, false
	}
	patch.LastHeartbeat = lastSeenMs / 1000
	if age := now.Sub(time.UnixMilli(lastSeenMs)); age <= agentOnlineWindow {
		patch.LastHeartbeat = now.Unix()
		patch.LatencyMs = int(age / time.Millisecond)
		return patch, true
	}
	return patch, false
}

// agentNodeProbe answers a probe for an agent-managed node from the
// bound agent's last_seen stamp (Unix ms, the unit the agent
// controller stamps). ok=false means the node has no bound agent — or
// the lookup failed — and the caller takes the dial path. The stored
// telemetry fields carry through unchanged (same as the heartbeat
// job's agent-expiry patch); only the error is dropped, so a ⚡ click
// also heals the bogus dial errors recorded before this change.
func (s *NodeService) agentNodeProbe(n *model.Node) (patch HeartbeatPatch, err error, ok bool) {
	var lastSeen *int64 // MAX() over zero rows is NULL
	qerr := database.GetDB().Model(&model.AgentNode{}).
		Where("node_id = ?", n.Id).
		Select("MAX(last_seen)").
		Scan(&lastSeen).Error
	if qerr != nil || lastSeen == nil {
		return HeartbeatPatch{}, nil, false
	}
	patch, online := agentHeartbeatVerdict(*lastSeen, time.Now())
	patch.XrayVersion = n.XrayVersion
	patch.PanelVersion = n.PanelVersion
	patch.Guid = n.Guid
	patch.CpuPct = n.CpuPct
	patch.MemPct = n.MemPct
	patch.UptimeSecs = n.UptimeSecs
	patch.NetUp = n.NetUp
	patch.NetDown = n.NetDown
	patch.XrayState = n.XrayState
	patch.XrayError = n.XrayError
	if !online {
		return patch, errAgentOffline, true
	}
	return patch, nil, true
}

func (s *NodeService) Probe(ctx context.Context, n *model.Node) (HeartbeatPatch, error) {
	// Every node is agent-managed: it has no panel to dial — its
	// address row is a placeholder ("agent") — so a probe answers
	// from the agent heartbeat instead of dialing a dead URL and
	// persisting a LastError the agent heartbeat never clears
	// (V2.2.1 ②). A node with no bound agent reads offline.
	if patch, err, ok := s.agentNodeProbe(n); ok {
		return patch, err
	}
	return HeartbeatPatch{LatencyMs: -1}, errAgentOffline
}

type ProbeResultUI struct {
	Status       string  `json:"status" example:"online"`
	LatencyMs    int     `json:"latencyMs" example:"42"`
	XrayVersion  string  `json:"xrayVersion" example:"25.10.31"`
	PanelVersion string  `json:"panelVersion" example:"v3.x.x"`
	CpuPct       float64 `json:"cpuPct" example:"12.5"`
	MemPct       float64 `json:"memPct" example:"45.2"`
	UptimeSecs   uint64  `json:"uptimeSecs" example:"86400"`
	Error        string  `json:"error"`
	// XrayState/XrayError are populated on successful probes even when the node's
	// Xray core is not healthy. The UI uses them for a distinct "panel ok, xray failed" indicator.
	XrayState string `json:"xrayState"`
	XrayError string `json:"xrayError"`
}

func (p HeartbeatPatch) ToUI(ok bool) ProbeResultUI {
	r := ProbeResultUI{
		LatencyMs:    p.LatencyMs,
		XrayVersion:  p.XrayVersion,
		PanelVersion: p.PanelVersion,
		CpuPct:       p.CpuPct,
		MemPct:       p.MemPct,
		UptimeSecs:   p.UptimeSecs,
		Error:        FriendlyProbeError(p.LastError),
		XrayState:    p.XrayState,
		XrayError:    p.XrayError,
	}
	if ok {
		r.Status = "online"
	} else {
		r.Status = "offline"
	}
	return r
}

func FriendlyProbeError(msg string) string {
	if strings.Contains(msg, "server gave HTTP response to HTTPS client") {
		return "the server speaks HTTP, not HTTPS; set the node scheme to http"
	}
	return msg
}