3x-ui/web/service/node.go

package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/mhsanaei/3x-ui/v3/database"
	"github.com/mhsanaei/3x-ui/v3/database/model"
	"github.com/mhsanaei/3x-ui/v3/util/common"
	"github.com/mhsanaei/3x-ui/v3/web/runtime"
)

// HeartbeatPatch is the slice of fields a single Probe() result writes
// back to a Node row. We pass it as a struct (not a *model.Node) so the
// heartbeat path can't accidentally clobber configuration columns the
// user just edited.
type HeartbeatPatch struct {
	Status        string
	LastHeartbeat int64
	LatencyMs     int
	XrayVersion   string
	CpuPct        float64
	MemPct        float64
	UptimeSecs    uint64
	LastError     string
}

// NodeService manages remote 3x-ui nodes registered with this panel.
// It owns CRUD for the Node model and the HTTP probe used by both the
// heartbeat job and the on-demand "test connection" UI action.
type NodeService struct{}

// httpClient is shared so repeated probes reuse TCP/TLS connections.
// Timeout is per-request, set on each Do() via context.
var nodeHTTPClient = &http.Client{
	Transport: &http.Transport{
		MaxIdleConns:        64,
		MaxIdleConnsPerHost: 4,
		IdleConnTimeout:     60 * time.Second,
	},
}

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
	return nodes, err
}

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
}

// normalize fills in defaults and trims accidental whitespace before save.
// Pulled out so Create and Update share the same rules.
func (s *NodeService) normalize(n *model.Node) error {
	n.Name = strings.TrimSpace(n.Name)
	n.Address = strings.TrimSpace(n.Address)
	n.ApiToken = strings.TrimSpace(n.ApiToken)
	if n.Name == "" {
		return common.NewError("node name is required")
	}
	if n.Address == "" {
		return common.NewError("node address is required")
	}
	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"
	}
	if n.BasePath == "" {
		n.BasePath = "/"
	}
	if !strings.HasPrefix(n.BasePath, "/") {
		n.BasePath = "/" + n.BasePath
	}
	if !strings.HasSuffix(n.BasePath, "/") {
		n.BasePath = n.BasePath + "/"
	}
	return nil
}

func (s *NodeService) Create(n *model.Node) error {
	if err := s.normalize(n); err != nil {
		return err
	}
	db := database.GetDB()
	return db.Create(n).Error
}

func (s *NodeService) Update(id int, in *model.Node) error {
	if err := s.normalize(in); err != nil {
		return err
	}
	db := database.GetDB()
	existing := &model.Node{}
	if err := db.Where("id = ?", id).First(existing).Error; err != nil {
		return err
	}
	// Only persist user-controlled columns. Heartbeat fields stay where
	// the heartbeat job last wrote them so a no-op edit doesn't blank
	// the dashboard out for ten seconds.
	updates := map[string]any{
		"name":      in.Name,
		"remark":    in.Remark,
		"scheme":    in.Scheme,
		"address":   in.Address,
		"port":      in.Port,
		"base_path": in.BasePath,
		"api_token": in.ApiToken,
		"enable":    in.Enable,
	}
	if err := db.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
		return err
	}
	// Drop any cached Remote so the next inbound op picks up the fresh
	// address/token. Cheap to do unconditionally — the next miss rebuilds.
	if mgr := runtime.GetManager(); mgr != nil {
		mgr.InvalidateNode(id)
	}
	return nil
}

func (s *NodeService) Delete(id int) error {
	db := database.GetDB()
	if err := db.Where("id = ?", id).Delete(model.Node{}).Error; err != nil {
		return err
	}
	if mgr := runtime.GetManager(); mgr != nil {
		mgr.InvalidateNode(id)
	}
	// Drop in-memory series so a freshly created node with the same id
	// doesn't inherit stale points (sqlite reuses ids freely).
	nodeMetrics.drop(nodeMetricKey(id, "cpu"))
	nodeMetrics.drop(nodeMetricKey(id, "mem"))
	return nil
}

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

// UpdateHeartbeat persists the slice of fields written by a probe. We
// don't touch updated_at via gorm autoUpdateTime here — that field is
// reserved for user-driven config edits.
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,
		"cpu_pct":        p.CpuPct,
		"mem_pct":        p.MemPct,
		"uptime_secs":    p.UptimeSecs,
		"last_error":     p.LastError,
	}
	if err := db.Model(model.Node{}).Where("id = ?", id).Updates(updates).Error; err != nil {
		return err
	}
	// Only record online ticks. Offline probes carry zeroed cpu/mem and
	// would draw a misleading dip on the chart; the gap on the x-axis is
	// the truthful representation of "we couldn't reach the node".
	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)
	}
	return nil
}

// nodeMetricKey is the namespacing used inside the singleton ring buffer
// so per-node metrics don't collide with each other or with the system
// metrics in the sibling singleton.
func nodeMetricKey(id int, metric string) string {
	return "node:" + strconv.Itoa(id) + ":" + metric
}

// AggregateNodeMetric returns up to maxPoints averaged buckets for one
// node's metric (currently "cpu" or "mem"). Output shape matches
// AggregateSystemMetric: {"t": unixSec, "v": value}.
func (s *NodeService) AggregateNodeMetric(id int, metric string, bucketSeconds int, maxPoints int) []map[string]any {
	return nodeMetrics.aggregate(nodeMetricKey(id, metric), bucketSeconds, maxPoints)
}

// Probe issues a single GET to the node's /panel/api/server/status and
// returns a HeartbeatPatch. On error the patch is zero-valued except
// for LastError; the caller is responsible for setting Status="offline".
//
// The remote endpoint requires authentication: we send the per-node
// ApiToken as a Bearer token, which the remote APIController.checkAPIAuth
// validates. Calls without a token would just get a 404, which masks
// the existence of the API entirely.
func (s *NodeService) Probe(ctx context.Context, n *model.Node) (HeartbeatPatch, error) {
	patch := HeartbeatPatch{LastHeartbeat: time.Now().Unix()}
	url := fmt.Sprintf("%s://%s:%d%spanel/api/server/status",
		n.Scheme, n.Address, n.Port, n.BasePath)

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		patch.LastError = err.Error()
		return patch, err
	}
	if n.ApiToken != "" {
		req.Header.Set("Authorization", "Bearer "+n.ApiToken)
	}
	req.Header.Set("Accept", "application/json")

	start := time.Now()
	resp, err := nodeHTTPClient.Do(req)
	if err != nil {
		patch.LastError = err.Error()
		return patch, err
	}
	defer resp.Body.Close()
	patch.LatencyMs = int(time.Since(start) / time.Millisecond)

	if resp.StatusCode != http.StatusOK {
		patch.LastError = fmt.Sprintf("HTTP %d from remote panel", resp.StatusCode)
		return patch, errors.New(patch.LastError)
	}

	// The remote wraps Status in entity.Msg. We decode into a typed
	// envelope rather than map[string]any so a schema change on the
	// remote shows up as a Go error instead of a silent zero-fill.
	var envelope struct {
		Success bool   `json:"success"`
		Msg     string `json:"msg"`
		Obj     *struct {
			Cpu uint64 `json:"-"`
			// Status fields we care about. Decode CPU/Mem nested
			// structs minimally — anything else gets discarded.
			CpuPct float64 `json:"cpu"`
			Mem    struct {
				Current uint64 `json:"current"`
				Total   uint64 `json:"total"`
			} `json:"mem"`
			Xray struct {
				Version string `json:"version"`
			} `json:"xray"`
			Uptime uint64 `json:"uptime"`
		} `json:"obj"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
		patch.LastError = "decode response: " + err.Error()
		return patch, err
	}
	if !envelope.Success || envelope.Obj == nil {
		patch.LastError = "remote returned success=false: " + envelope.Msg
		return patch, errors.New(patch.LastError)
	}
	o := envelope.Obj
	patch.CpuPct = o.CpuPct
	if o.Mem.Total > 0 {
		patch.MemPct = float64(o.Mem.Current) * 100.0 / float64(o.Mem.Total)
	}
	patch.XrayVersion = o.Xray.Version
	patch.UptimeSecs = o.Uptime
	return patch, nil
}

// EnvelopeForUI is the shape a frontend test-connection action expects.
// Pulling it out keeps the controller dumb.
type ProbeResultUI struct {
	Status      string  `json:"status"`
	LatencyMs   int     `json:"latencyMs"`
	XrayVersion string  `json:"xrayVersion"`
	CpuPct      float64 `json:"cpuPct"`
	MemPct      float64 `json:"memPct"`
	UptimeSecs  uint64  `json:"uptimeSecs"`
	Error       string  `json:"error"`
}

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