pangolin/server/internal/httpapi/nodes.go

package httpapi

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

	"github.com/go-chi/chi/v5"

	"github.com/wangjia/pangolin/server/internal/apierr"
	"github.com/wangjia/pangolin/server/internal/auth"
	"github.com/wangjia/pangolin/server/internal/nodes"
	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
)

const (
	// paidCredentialTTL is the default connect credential lifetime for paid users.
	paidCredentialTTL = 24 * time.Hour
	// freeCredentialTTL is the per-minute TTL for free users (per remaining minutes).
	freeMinuteTTL = time.Minute
	// deviceStaleWindow: connect 设备上限 backstop 只数近此窗口活跃的设备(与 devices 侧一致)。
	deviceStaleWindow = 30 * 24 * time.Hour
)

// nodeLoadReader reads a node's last-reported runtime load (for the data-plane
// health flag). *nodes.LoadCache satisfies it; nil = treat all nodes healthy.
type nodeLoadReader interface {
	Get(ctx context.Context, nodeUUID string) (*nodes.NodeLoad, bool, error)
}

// NodeAPI serves the /v1/nodes endpoints.
type NodeAPI struct {
	store     nodes.NodeStore
	hub       *nodes.Hub
	load      nodeLoadReader
	deriveKey string
	// rulesBaseURL 是控制面对外公网基址(PANGOLIN_PUBLIC_URL),供国内分流的
	// rule_set .srs 下载用;空则分流不生效。
	rulesBaseURL string
}

// NewNodeAPI creates a NodeAPI. load may be nil (then all nodes are treated as
// data-plane healthy — agent gRPC liveness still gates status).
func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, load nodeLoadReader, deriveKey, rulesBaseURL string) *NodeAPI {
	return &NodeAPI{store: store, hub: hub, load: load, deriveKey: deriveKey, rulesBaseURL: rulesBaseURL}
}

// dataPlaneHealthy reports the node's last sing-box health (default true when
// unknown: just registered / load expired / read error — agent liveness gates).
func (a *NodeAPI) dataPlaneHealthy(ctx context.Context, nodeUUID string) bool {
	if a.load == nil {
		return true
	}
	l, ok, err := a.load.Get(ctx, nodeUUID)
	if err != nil || !ok || l == nil {
		return true
	}
	return l.DataPlaneHealthy
}

// ─── GET /v1/nodes ───────────────────────────────────────────────────────────

type nodeResponse struct {
	ID     string `json:"id"`     // UUID (used as path param for connect)
	Region string `json:"region"` // HK / JP / SG / US
	NameZH string `json:"name_zh"`
	NameEN string `json:"name_en"`
	Tier   string `json:"tier"`   // "free" | "pro"
	Status string `json:"status"` // "up" only when the agent is also online; "down" otherwise
	// host/port 暴露节点入口,供客户端实测真实延迟(per-client ping 服务端无法代知)。
	Host string `json:"host"`
	Port int    `json:"port"`
}

// effectiveNodeStatus downgrades a DB-'up' node to "down" when its agent is
// offline OR its data plane (sing-box) is unhealthy. The DB status column reflects
// provisioning/scheduler state, NOT live health — without this, a node whose agent
// has been offline for days, or whose sing-box has crashed while the agent stays
// connected, still shows healthy and lets clients "connect" to a node that can't
// serve them.
func effectiveNodeStatus(dbStatus string, agentOnline, dataPlaneHealthy bool) string {
	if dbStatus == "up" && (!agentOnline || !dataPlaneHealthy) {
		return "down"
	}
	return dbStatus
}

// ListNodes handles GET /v1/nodes.
func (a *NodeAPI) ListNodes(w http.ResponseWriter, r *http.Request) {
	nodeRows, err := a.store.ListUp(r.Context())
	if err != nil {
		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
		return
	}

	resp := make([]nodeResponse, 0, len(nodeRows))
	for _, n := range nodeRows {
		host, portStr := splitHostPort(n.Endpoint)
		if host == "" {
			host = n.Endpoint
		}
		port, _ := strconv.Atoi(portStr)
		resp = append(resp, nodeResponse{
			ID:     n.UUID,
			Region: n.Region,
			NameZH: n.NameZH,
			NameEN: n.NameEN,
			Tier:   n.Tier,
			Status: effectiveNodeStatus(n.Status,
				a.hub == nil || a.hub.IsOnline(n.UUID),
				a.dataPlaneHealthy(r.Context(), n.UUID)),
			Host:   host,
			Port:   port,
		})
	}

	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	_ = json.NewEncoder(w).Encode(map[string]any{"nodes": resp})
}

// ─── POST /v1/nodes/{id}/connect ─────────────────────────────────────────────

type connectRequest struct {
	DeviceID string `json:"device_id"`
}

// ConnectNode handles POST /v1/nodes/{id}/connect.
func (a *NodeAPI) ConnectNode(w http.ResponseWriter, r *http.Request) {
	uid, ok := auth.UserIDFromContext(r.Context())
	if !ok {
		apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
		return
	}

	nodeUUID := chi.URLParam(r, "id")
	if nodeUUID == "" {
		apierr.WriteJSON(w, http.StatusBadRequest, apierr.ErrBadRequest)
		return
	}

	var req connectRequest
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 8*1024)).Decode(&req); err != nil || strings.TrimSpace(req.DeviceID) == "" {
		apierr.WriteJSON(w, http.StatusBadRequest, &apierr.Error{
			Code:      "BAD_REQUEST",
			MessageZH: "缺少 device_id",
			MessageEn: "Missing device_id",
		})
		return
	}

	// 1. Load user entitlement (dp_uuid + plan).
	ent, err := a.store.EntitlementForUser(r.Context(), uid)
	if err != nil {
		slog.Error("connect: entitlement load failed", "user", uid, "err", err)
		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
		return
	}
	if ent == nil {
		apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
		return
	}

	// 1.4 设备数量上限 backstop(#16)：活跃设备超套餐上限则拒连，返回 device_limit 信号让
	// 新客户端弹「移除设备」页。登录挡板只拦"新登录"，这里兜住"已登录的超限设备"——
	// 它们下次一连就会被拦、被提示，无需重新登录。max_devices=0 表示不限。
	if ent.MaxDevices > 0 {
		active, cerr := a.store.CountActiveDevices(r.Context(), uid, time.Now().UTC().Add(-deviceStaleWindow))
		if cerr != nil {
			slog.Error("connect: count active devices failed", "user", uid, "err", cerr)
			apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
			return
		}
		if active > ent.MaxDevices {
			w.Header().Set("Content-Type", "application/json; charset=utf-8")
			w.WriteHeader(http.StatusForbidden)
			_ = json.NewEncoder(w).Encode(map[string]any{
				"code":        "DEVICE_LIMIT_EXCEEDED",
				"message_zh":  "设备数已达上限，请移除一台设备后再连接",
				"message_en":  "Device limit reached. Remove a device to connect.",
				"max_devices": ent.MaxDevices,
			})
			return
		}
	}

	// 1.5 GB 综合配额卡控(todo #5 Phase 2)：按账户当日综合流量卡，超 plan.daily_mb 即拒。
	// 对免费(与分钟门双卡)与付费(高上限防滥用)统一生效；daily_mb NULL = 不限。
	if ent.DailyMB.Valid {
		usedBytes, qerr := a.store.AccountDayBytes(r.Context(), uid, time.Now().UTC())
		if qerr != nil {
			slog.Error("connect: account day bytes failed", "user", uid, "err", qerr)
			apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
			return
		}
		if usedBytes >= ent.DailyMB.Int64*(1<<20) {
			apierr.WriteJSON(w, http.StatusForbidden, apierr.ErrQuotaExhausted)
			return
		}
	}

	// 2. Determine TTL from plan.
	var ttl time.Duration
	if ent.AdGate {
		// Free plan: enforce the account-wide daily minute quota (shared across
		// ALL devices). allowance = plan.daily_minutes + 当日看广告累加的 bonus;
		// remaining = allowance − minutes_used. Credential TTL = remaining so the
		// data-plane hard-cuts when time runs out (backstop even if the client
		// countdown is bypassed). remaining ≤ 0 → 拒 QUOTA_EXHAUSTED, client 弹广告/升级.
		dm := int64(10)
		if ent.DailyMinutes.Valid {
			dm = ent.DailyMinutes.Int64
		}
		used, bonus, qerr := a.store.AccountDayMinutes(r.Context(), uid, time.Now().UTC())
		if qerr != nil {
			slog.Error("connect: account day minutes failed", "user", uid, "err", qerr)
			apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
			return
		}
		remaining := dm + int64(bonus) - int64(used)
		if remaining <= 0 {
			w.Header().Set("Content-Type", "application/json; charset=utf-8")
			w.WriteHeader(http.StatusForbidden)
			_ = json.NewEncoder(w).Encode(map[string]any{
				"code":       "QUOTA_EXHAUSTED",
				"message_zh": "今日免费时长已用完，看广告加时或升级会员",
				"message_en": "Daily free minutes used up. Watch an ad to add time or upgrade.",
			})
			return
		}
		ttl = time.Duration(remaining) * freeMinuteTTL
	} else {
		ttl = paidCredentialTTL
	}
	expiresAt := time.Now().UTC().Add(ttl)

	// 2.5 Per-device dp_uuid：每台已注册设备一个专属数据面凭证。设备未注册/已被移除 →
	// **拒连并提示重新登录(重新注册设备)**,不再回退账户级 dp_uuid —— 否则被移除的设备
	// 靠残留 access token + 回退还能钻进来,绕过设备上限(#16)。
	devDp, _, derr := a.store.EnsureDeviceDpUUID(r.Context(), uid, req.DeviceID)
	if errors.Is(derr, nodes.ErrDeviceNotFound) || (derr == nil && devDp == "") {
		w.Header().Set("Content-Type", "application/json; charset=utf-8")
		w.WriteHeader(http.StatusForbidden)
		_ = json.NewEncoder(w).Encode(map[string]any{
			"code":       "DEVICE_NOT_REGISTERED",
			"message_zh": "设备未注册或已被移除，请退出后重新登录以重新注册设备",
			"message_en": "Device not registered. Please sign in again to re-register this device.",
		})
		return
	}
	if derr != nil {
		slog.Error("connect: ensure device dp_uuid failed", "user", uid, "device", req.DeviceID, "err", derr)
		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
		return
	}
	dpUUID := devDp

	// 3. Resolve node.
	node, err := a.store.NodeByUUID(r.Context(), nodeUUID)
	if err != nil {
		slog.Error("connect: node load failed", "node", nodeUUID, "err", err)
		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
		return
	}
	if node == nil || effectiveNodeStatus(node.Status,
		a.hub == nil || a.hub.IsOnline(nodeUUID),
		a.dataPlaneHealthy(r.Context(), nodeUUID)) != "up" {
		apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
		return
	}

	// 4. Build the agentv1.Credential.
	cred := &agentv1.Credential{
		DpUUID:        dpUUID,
		Protocol:      agentv1.ProtocolBoth,
		Flow:          "xtls-rprx-vision",
		ExpiresAtUnix: expiresAt.Unix(),
	}

	// 5. Refuse early if the node's agent is offline. A credential we can't push
	//    now leaves the client "connected" to a node that rejects its UUID — the
	//    exact failure that hid a 6-day agent outage. Persist it so the agent
	//    applies it on resync, then signal unavailable instead of faking success.
	if !a.hub.IsOnline(node.UUID) {
		if persistErr := a.store.PersistCredential(r.Context(), node.ID, cred, expiresAt); persistErr != nil {
			slog.Error("connect: persist credential (agent offline) failed", "node", nodeUUID, "user", uid, "err", persistErr)
		}
		slog.Warn("connect: node agent offline, refusing connect", "node", nodeUUID, "user", uid)
		apierr.WriteJSON(w, http.StatusServiceUnavailable, apierr.ErrNodeUnavailable)
		return
	}

	// 6. Push to the node agent via Hub. The agent is online, so a push failure is
	//    a real error — surface it, don't pretend the tunnel will work.
	if pushErr := a.hub.Push(r.Context(), node.UUID, &agentv1.Command{
		Type:   agentv1.CommandTypeUpsert,
		Upsert: &agentv1.UpsertPayload{Credential: cred},
	}); pushErr != nil {
		slog.Error("connect: push credential to agent failed", "node", nodeUUID, "user", uid, "err", pushErr)
		apierr.WriteJSON(w, http.StatusServiceUnavailable, apierr.ErrNodeUnavailable)
		return
	}

	// 7. Persist credential for agent resync (best-effort; the push above already
	//    delivered it to the live agent).
	if persistErr := a.store.PersistCredential(r.Context(), node.ID, cred, expiresAt); persistErr != nil {
		slog.Error("connect: persist credential failed", "node", nodeUUID, "user", uid, "err", persistErr)
	}

	// 7. Render and return the full sing-box CLIENT config JSON.
	//    split_cn=1/true → 国内 IP/域名直连(#5);客户端按 smartRoute 偏好传。
	splitCN := r.URL.Query().Get("split_cn") == "1" || r.URL.Query().Get("split_cn") == "true"
	cfgJSON, renderErr := BuildClientConfig(node, dpUUID, a.deriveKey,
		ClientConfigOpts{SplitCN: splitCN, RulesBaseURL: a.rulesBaseURL})
	if renderErr != nil {
		slog.Error("connect: build client config failed", "node", nodeUUID, "err", renderErr)
		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
		return
	}
	// 可观测:国内分流是否真正生效(split_active=两个条件都满足才渲染 rule_set)。
	slog.Info("client config rendered", "node", nodeUUID, "split_cn", splitCN,
		"rules_base_set", a.rulesBaseURL != "",
		"split_active", splitCN && a.rulesBaseURL != "", "bytes", len(cfgJSON))

	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	_, _ = w.Write(cfgJSON)
}

// ─── POST /v1/nodes/{id}/disconnect ──────────────────────────────────────────

// disconnectRequest is the optional JSON body: device_id 指认要吊销凭证的设备。
// 旧客户端不带 body(或不带 device_id)→ 仅吊销遗留账户级凭证(历史行为)。
type disconnectRequest struct {
	DeviceID string `json:"device_id"`
}

// DisconnectNode handles POST /v1/nodes/{id}/disconnect.
//
// F4:connect 下发的是每设备凭证(EnsureDeviceDpUUID),吊销也必须对准它——
// 此前这里只撤账户级 ent.DpUUID,设备凭证一直活到 TTL(付费 24h),「断开」在
// 服务端形同空转。现按 device_id 吊销该设备的 dp_uuid,账户级作为遗留兜底仍撤。
func (a *NodeAPI) DisconnectNode(w http.ResponseWriter, r *http.Request) {
	uid, ok := auth.UserIDFromContext(r.Context())
	if !ok {
		apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
		return
	}

	nodeUUID := chi.URLParam(r, "id")
	if nodeUUID == "" {
		apierr.WriteJSON(w, http.StatusBadRequest, apierr.ErrBadRequest)
		return
	}

	// Optional body(旧客户端无 body → device_id 为空,走兜底路径)。
	var req disconnectRequest
	_ = json.NewDecoder(http.MaxBytesReader(w, r.Body, 8*1024)).Decode(&req)

	// Load dp_uuid.
	ent, err := a.store.EntitlementForUser(r.Context(), uid)
	if err != nil {
		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
		return
	}
	if ent == nil {
		w.WriteHeader(http.StatusNoContent) // nothing to revoke
		return
	}

	node, err := a.store.NodeByUUID(r.Context(), nodeUUID)
	if err != nil {
		apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
		return
	}
	if node == nil {
		w.WriteHeader(http.StatusNoContent)
		return
	}

	// 收集待吊销凭证:每设备(connect 真正下发的)+ 账户级(遗留兜底)。
	dpUUIDs := make([]string, 0, 2)
	if devID := strings.TrimSpace(req.DeviceID); devID != "" {
		devDp, _, derr := a.store.EnsureDeviceDpUUID(r.Context(), uid, devID)
		if derr != nil && !errors.Is(derr, nodes.ErrDeviceNotFound) {
			slog.Warn("disconnect: device dp_uuid lookup failed", "user", uid, "device", devID, "err", derr)
		}
		if devDp != "" {
			dpUUIDs = append(dpUUIDs, devDp)
		}
	}
	if ent.DpUUID != "" {
		dpUUIDs = append(dpUUIDs, ent.DpUUID)
	}

	for _, dp := range dpUUIDs {
		// Push revoke command(best-effort:agent 离线时命令进 Redis 队列,重连即达;
		// 删除持久化凭证后 resync 也不会再下发)。nil hub = 测试环境,跳过推送。
		if a.hub != nil {
			_ = a.hub.Push(r.Context(), node.UUID, &agentv1.Command{
				Type:   agentv1.CommandTypeRevoke,
				Revoke: &agentv1.RevokePayload{DpUUID: dp},
			})
		}
		// Delete persisted credential.
		_ = a.store.DeleteCredential(r.Context(), node.ID, dp)
	}

	w.WriteHeader(http.StatusNoContent)
}