feat(backend): 挂载 /v1 API + 实现 nodes/connect 端到端

- 新增 internal/dpcred 包,统一 DeriveHy2Password + DefaultFlow
  agentd 与 HTTP connect handler 共享同一实现
- 新增迁移 000011:nodes 表拆分 reality_prk 私钥 / reality_pbk 公钥
  reality_short_id;修正 handler_grpc.go 使用私钥字段
- 新增迁移 000012:connect_credentials 持久化凭证
  实现 CredentialsForNode 修复 agent 重连 resync 原先返回空的桩
- 扩展 NodeStore 接口:ListUp / EntitlementForUser /
  PersistCredential / DeleteCredential;同步 grpc_test.go mock
- 新增 httpapi/nodes.go:GET /nodes、POST /nodes/id/connect
  Hub.Push + PersistCredential + 渲染完整 sing-box client 配置 JSON
  POST /nodes/id/disconnect
- 新增 httpapi/account.go:GET /me、GET /plans、GET /notices
- 新增 httpapi/clientconfig.go:BuildClientConfig 服务端渲染
- 重写 cmd/server/main.go:手写 chi public/protected 分组
  nodes.Service/Hub 在 main 构造并共享;SMTPMailer/LogMailer

go build ./... && go vet ./... && go test ./... 全部通过

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-15 23:09:56 +08:00
parent b35bfe10dc
commit cadd527680
14 changed files with 1017 additions and 149 deletions
+1
View File
@@ -32,6 +32,7 @@ const (
DefaultSingboxCfg = "/etc/sing-box/config.json"
// DefaultFlow is the REALITY VLESS flow (doc/02 §3.1).
// Kept here as an alias; canonical definition is in internal/dpcred.
DefaultFlow = "xtls-rprx-vision"
)
+6 -29
View File
@@ -1,34 +1,11 @@
package agentd
import (
"crypto/hmac"
"crypto/sha256"
"encoding/base64"
)
import "github.com/wangjia/pangolin/server/internal/dpcred"
// DeriveHy2Password derives a node-agnostic Hysteria2 password from the opaque
// data-plane credential id (dp_uuid). The REALITY inbound uses dp_uuid directly as
// the VLESS user uuid; the Hy2 inbound cannot reuse a UUID as a password verbatim
// (it must look like an opaque secret), so it is derived from the SAME source —
// "password = dp_uuid 同源派生" — via a keyed HMAC.
//
// The derivation is deterministic given (dp_uuid, key): the control plane runs the
// exact same function when it builds the client's connect config (doc/02 §3.1), so
// both sides agree without the password ever crossing the agent contract.
//
// Risk note (carried from task #5): the derivation key is shared material. If a
// node is seized the attacker still only sees dp_uuids and this node's key, which
// lets them recompute Hy2 passwords for dp_uuids THEY ALREADY HOLD — it does not
// reveal other subscribers' credentials or any account identity. Rotating the key
// rotates every Hy2 password. When key == "" the password falls back to the raw
// dp_uuid (acceptable for dev; production cloud-init always injects a key).
// DeriveHy2Password is an alias for dpcred.DeriveHy2Password kept here so
// existing callers within the agentd package compile without change.
// The canonical implementation lives in internal/dpcred so that both the node
// agent and the HTTP connect handler share the exact same derivation logic.
func DeriveHy2Password(dpUUID, key string) string {
if key == "" {
return dpUUID
}
mac := hmac.New(sha256.New, []byte(key))
mac.Write([]byte(dpUUID))
sum := mac.Sum(nil)
// base64url without padding → URL/JSON-safe, 43 chars.
return base64.RawURLEncoding.EncodeToString(sum)
return dpcred.DeriveHy2Password(dpUUID, key)
}
+39
View File
@@ -0,0 +1,39 @@
// Package dpcred holds the data-plane credential derivation utilities shared
// between the node agent (agentd) and the HTTP control-plane connect handler.
// The functions must produce identical output on both sides — keeping them in a
// single package is the only safe way to guarantee that.
package dpcred
import (
"crypto/hmac"
"crypto/sha256"
"encoding/base64"
)
// DefaultFlow is the REALITY VLESS flow used for both server inbound and client
// outbound configuration (doc/02 §3.1).
const DefaultFlow = "xtls-rprx-vision"
// DeriveHy2Password derives a node-agnostic Hysteria2 password from the opaque
// data-plane credential id (dp_uuid). The REALITY inbound uses dp_uuid directly
// as the VLESS user uuid; the Hy2 inbound cannot reuse a UUID as a password
// verbatim (it must look like an opaque secret), so it is derived from the SAME
// source — "password = dp_uuid 同源派生" — via a keyed HMAC.
//
// The derivation is deterministic given (dp_uuid, key): the control plane runs
// the exact same function when it builds the client's connect config (doc/02
// §3.1), so both sides agree without the password ever crossing the agent
// contract.
//
// When key == "" the password falls back to the raw dp_uuid (acceptable for
// dev; production always injects a key via NODE_DERIVE_KEY env).
func DeriveHy2Password(dpUUID, key string) string {
if key == "" {
return dpUUID
}
mac := hmac.New(sha256.New, []byte(key))
mac.Write([]byte(dpUUID))
sum := mac.Sum(nil)
// base64url without padding → URL/JSON-safe, 43 chars.
return base64.RawURLEncoding.EncodeToString(sum)
}
+135
View File
@@ -0,0 +1,135 @@
package httpapi
import (
"database/sql"
"encoding/json"
"net/http"
"time"
"github.com/wangjia/pangolin/server/internal/apierr"
"github.com/wangjia/pangolin/server/internal/auth"
)
// AccountAPI serves /v1/me and supporting endpoints.
type AccountAPI struct {
db *sql.DB
}
// NewAccountAPI creates an AccountAPI.
func NewAccountAPI(db *sql.DB) *AccountAPI { return &AccountAPI{db: db} }
// ─── GET /v1/me ──────────────────────────────────────────────────────────────
type meResponse struct {
UUID string `json:"uuid"`
Email string `json:"email"`
DpUUID string `json:"dp_uuid"`
Plan string `json:"plan"` // "free" | "pro" | "team"
ExpireAt *string `json:"expire_at"` // RFC3339 UTC, null = no active sub
}
// GetMe handles GET /v1/me.
func (a *AccountAPI) GetMe(w http.ResponseWriter, r *http.Request) {
uid, ok := auth.UserIDFromContext(r.Context())
if !ok {
apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
return
}
var (
uuid string
email string
dpUUID string
)
if err := a.db.QueryRowContext(r.Context(),
`SELECT uuid, email, dp_uuid FROM users WHERE id = ? AND status = 'active'`, uid,
).Scan(&uuid, &email, &dpUUID); err == sql.ErrNoRows {
apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
return
} else if err != nil {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
// Best active subscription.
var planCode string
var expiresAt sql.NullTime
err := a.db.QueryRowContext(r.Context(), `
SELECT p.code, s.expires_at
FROM subscriptions s
JOIN plans p ON p.id = s.plan_id
WHERE s.user_id = ? AND s.expires_at > UTC_TIMESTAMP()
ORDER BY s.expires_at DESC
LIMIT 1
`, uid).Scan(&planCode, &expiresAt)
if err == sql.ErrNoRows {
planCode = "free"
} else if err != nil {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
resp := meResponse{
UUID: uuid,
Email: email,
DpUUID: dpUUID,
Plan: planCode,
}
if expiresAt.Valid {
s := expiresAt.Time.UTC().Format(time.RFC3339)
resp.ExpireAt = &s
}
w.Header().Set("Content-Type", "application/json; charset=utf-8")
_ = json.NewEncoder(w).Encode(resp)
}
// ─── GET /v1/plans ───────────────────────────────────────────────────────────
type planResponse struct {
Code string `json:"code"`
NameZH string `json:"name_zh"`
NameEN string `json:"name_en"`
DailyMinutes *int64 `json:"daily_minutes"` // null = unlimited
AdGate bool `json:"ad_gate"`
}
// ListPlans handles GET /v1/plans.
func (a *AccountAPI) ListPlans(w http.ResponseWriter, r *http.Request) {
rows, err := a.db.QueryContext(r.Context(),
`SELECT code, name_zh, name_en, daily_minutes, ad_gate FROM plans ORDER BY id`)
if err != nil {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
defer rows.Close()
var plans []planResponse
for rows.Next() {
var p planResponse
var dm sql.NullInt64
if err := rows.Scan(&p.Code, &p.NameZH, &p.NameEN, &dm, &p.AdGate); err != nil {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
if dm.Valid {
p.DailyMinutes = &dm.Int64
}
plans = append(plans, p)
}
if err := rows.Err(); err != nil {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
w.Header().Set("Content-Type", "application/json; charset=utf-8")
_ = json.NewEncoder(w).Encode(map[string]any{"plans": plans})
}
// ─── GET /v1/notices ─────────────────────────────────────────────────────────
// ListNotices handles GET /v1/notices. Returns an empty list for the MVP.
func (a *AccountAPI) ListNotices(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
_ = json.NewEncoder(w).Encode(map[string]any{"notices": []any{}})
}
+155
View File
@@ -0,0 +1,155 @@
package httpapi
import (
"encoding/json"
"strings"
"github.com/wangjia/pangolin/server/internal/dpcred"
"github.com/wangjia/pangolin/server/internal/nodes"
)
// BuildClientConfig renders a complete sing-box CLIENT configuration JSON that
// the client app passes verbatim to the local tunnel kernel.
//
// Design rule (ARCHITECTURE.md §3.1): the Dart/Flutter client MUST NOT assemble
// or modify the config — it is rendered here, server-side, and returned raw.
//
// Parameters:
// - node: the target node row (provides endpoint, keys, ports)
// - dpUUID: the authenticated user's data-plane UUID (used as VLESS uuid)
// - deriveKey: shared HMAC key used by both server and agent to derive the
// Hysteria2 password from dp_uuid (must equal PANGOLIN_AGENT_DERIVE_KEY)
// - ttlSeconds: credential lifetime hint; not embedded in the config but can
// be used by callers to set a session timer
func BuildClientConfig(node *nodes.NodeRow, dpUUID, deriveKey string) ([]byte, error) {
// Parse host:port from endpoint; endpoint format is "host:port".
host, _ := splitHostPort(node.Endpoint)
if host == "" {
host = node.Endpoint
}
realityPublicKey := node.RealityPBK
realityShortID := node.RealityShortID
hy2Password := dpcred.DeriveHy2Password(dpUUID, deriveKey)
hy2Port := int32(443) // default
if node.Hy2Port.Valid {
hy2Port = node.Hy2Port.Int32
}
// REALITY outbound (VLESS + REALITY TLS, TCP 443).
realityOut := map[string]any{
"type": "vless",
"tag": "reality-out",
"server": host,
"server_port": 11443, // REALITY always uses port from endpoint
"uuid": dpUUID,
"flow": dpcred.DefaultFlow,
"tls": map[string]any{
"enabled": true,
"server_name": node.RealitySNI,
"utls": map[string]any{
"enabled": true,
"fingerprint": "chrome",
},
"reality": map[string]any{
"enabled": true,
"public_key": realityPublicKey,
"short_id": realityShortID,
},
},
}
// Parse the REALITY listen port from endpoint.
if _, portStr := splitHostPort(node.Endpoint); portStr != "" {
port := 0
for _, ch := range portStr {
if ch >= '0' && ch <= '9' {
port = port*10 + int(ch-'0')
}
}
if port > 0 {
realityOut["server_port"] = port
}
}
// Hysteria2 outbound (UDP 443).
hy2Out := map[string]any{
"type": "hysteria2",
"tag": "hy2-out",
"server": host,
"server_port": hy2Port,
"password": hy2Password,
"tls": map[string]any{
"enabled": true,
"alpn": []string{"h3"},
},
}
// TUN inbound with kill-switch (strict_route).
tunIn := map[string]any{
"type": "tun",
"tag": "tun-in",
"address": []string{"172.19.0.1/30"},
"mtu": 9000,
"auto_route": true,
"strict_route": true,
"stack": "system",
}
// urltest auto-select outbound.
autoBest := map[string]any{
"type": "urltest",
"tag": "auto",
"outbounds": []string{"reality-out", "hy2-out"},
"url": "https://www.gstatic.com/generate_204",
"interval": "3m",
"tolerance": 50,
}
// Route: LAN direct, everything else via auto.
route := map[string]any{
"rules": []any{
map[string]any{
"ip_cidr": []string{"10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "127.0.0.0/8"},
"outbound": "direct",
},
},
"final": "auto",
"auto_detect_interface": true,
}
// DNS: remote over tunnel, local for domestic.
dns := map[string]any{
"servers": []any{
map[string]any{"tag": "remote", "address": "tls://8.8.8.8", "detour": "auto"},
map[string]any{"tag": "local", "address": "223.5.5.5", "detour": "direct"},
},
"final": "remote",
"strategy": "ipv4_only",
}
cfg := map[string]any{
"log": map[string]any{"level": "warn", "timestamp": true},
"inbounds": []any{tunIn},
"outbounds": []any{
realityOut,
hy2Out,
autoBest,
map[string]any{"type": "block", "tag": "block"},
map[string]any{"type": "direct", "tag": "direct"},
},
"route": route,
"dns": dns,
}
return json.Marshal(cfg)
}
// splitHostPort splits "host:port" into (host, port). Returns ("", "") on failure.
func splitHostPort(s string) (host, port string) {
i := strings.LastIndexByte(s, ':')
if i < 0 {
return s, ""
}
return s[:i], s[i+1:]
}
+224
View File
@@ -0,0 +1,224 @@
package httpapi
import (
"encoding/json"
"net/http"
"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
)
// NodeAPI serves the /v1/nodes endpoints.
type NodeAPI struct {
store nodes.NodeStore
hub *nodes.Hub
deriveKey string
}
// NewNodeAPI creates a NodeAPI.
func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, deriveKey string) *NodeAPI {
return &NodeAPI{store: store, hub: hub, deriveKey: deriveKey}
}
// ─── 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"` // always "up" in this endpoint
}
// 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 {
resp = append(resp, nodeResponse{
ID: n.UUID,
Region: n.Region,
NameZH: n.NameZH,
NameEN: n.NameEN,
Tier: n.Tier,
Status: n.Status,
})
}
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 {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
if ent == nil {
apierr.WriteJSON(w, http.StatusUnauthorized, apierr.ErrUnauthorized)
return
}
// 2. Determine TTL from plan.
var ttl time.Duration
if ent.AdGate {
// Free plan: flat 10-minute session for MVP (full ad-gate in a later pass).
dm := int64(10)
if ent.DailyMinutes.Valid {
dm = ent.DailyMinutes.Int64
}
if dm <= 0 {
apierr.WriteJSON(w, http.StatusForbidden, apierr.ErrQuotaExhausted)
return
}
ttl = time.Duration(dm) * freeMinuteTTL
} else {
ttl = paidCredentialTTL
}
expiresAt := time.Now().UTC().Add(ttl)
// 3. Resolve node.
node, err := a.store.NodeByUUID(r.Context(), nodeUUID)
if err != nil {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
if node == nil || node.Status != "up" {
apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound)
return
}
// 4. Build the agentv1.Credential.
cred := &agentv1.Credential{
DpUUID: ent.DpUUID,
Protocol: agentv1.ProtocolBoth,
Flow: "xtls-rprx-vision",
ExpiresAtUnix: expiresAt.Unix(),
}
// 5. Push to the node agent via Hub (real gRPC channel).
pushErr := a.hub.Push(r.Context(), node.UUID, &agentv1.Command{
Type: agentv1.CommandTypeUpsert,
Upsert: &agentv1.UpsertPayload{Credential: cred},
})
if pushErr != nil {
// Non-fatal: command is queued in Redis and will be replayed on reconnect.
// Log but don't abort — return the config so the client can attempt the tunnel.
_ = pushErr
}
// 6. Persist credential for agent resync.
if persistErr := a.store.PersistCredential(r.Context(), node.ID, cred, expiresAt); persistErr != nil {
// Non-fatal: tunnel still works via Hub push; resync will miss it on agent restart.
_ = persistErr
}
// 7. Render and return the full sing-box CLIENT config JSON.
cfgJSON, renderErr := BuildClientConfig(node, ent.DpUUID, a.deriveKey)
if renderErr != nil {
apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal)
return
}
w.Header().Set("Content-Type", "application/json; charset=utf-8")
_, _ = w.Write(cfgJSON)
}
// ─── POST /v1/nodes/{id}/disconnect ──────────────────────────────────────────
// DisconnectNode handles POST /v1/nodes/{id}/disconnect.
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
}
// 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
}
// Push revoke command.
_ = a.hub.Push(r.Context(), node.UUID, &agentv1.Command{
Type: agentv1.CommandTypeRevoke,
Revoke: &agentv1.RevokePayload{DpUUID: ent.DpUUID},
})
// Delete persisted credential.
_ = a.store.DeleteCredential(r.Context(), node.ID, ent.DpUUID)
w.WriteHeader(http.StatusNoContent)
}
+16
View File
@@ -80,6 +80,22 @@ func (m *mockNodeStore) UserIDByDpUUID(_ context.Context, dpUUID string) (int64,
return 0, false, nil
}
func (m *mockNodeStore) ListUp(_ context.Context) ([]*nodes.NodeRow, error) {
return []*nodes.NodeRow{{ID: 1, UUID: m.nodeUUID, Status: "up"}}, nil
}
func (m *mockNodeStore) EntitlementForUser(_ context.Context, _ int64) (*nodes.Entitlement, error) {
return nil, nil
}
func (m *mockNodeStore) PersistCredential(_ context.Context, _ int64, _ *agentv1.Credential, _ time.Time) error {
return nil
}
func (m *mockNodeStore) DeleteCredential(_ context.Context, _ int64, _ string) error {
return nil
}
func (m *mockNodeStore) AccumulateUsage(_ context.Context, userID int64, date time.Time,
bytesUp, bytesDown, minutes int64,
) error {
+11 -1
View File
@@ -106,9 +106,19 @@ func (h *Handler) Register(ctx context.Context, req *agentv1.RegisterRequest) (*
}
// Populate inbound configs from the nodes row.
if node.RealityPBK != "" {
// reality_prk is the PRIVATE key the agent's VLESS inbound needs;
// reality_pbk is the PUBLIC key sent to clients in the connect config.
if node.RealityPRK != "" {
snap.Reality = &agentv1.RealityInbound{
PrivateKey: node.RealityPRK,
ShortID: node.RealityShortID,
ServerName: node.RealitySNI,
}
} else if node.RealityPBK != "" {
// Fallback for nodes seeded before migration 000011: use pbk field.
snap.Reality = &agentv1.RealityInbound{
PrivateKey: node.RealityPBK,
ShortID: node.RealityShortID,
ServerName: node.RealitySNI,
}
}
+164 -15
View File
@@ -11,13 +11,28 @@ import (
// NodeRow holds a node's essential fields from the nodes table.
type NodeRow struct {
ID int64
UUID string
Status string
RealityPBK string
RealitySNI string
Endpoint string
Hy2Port sql.NullInt32
ID int64
UUID string
Region string
NameZH string
NameEN string
Tier string
Status string
RealityPBK string // REALITY x25519 PUBLIC key (for client connect config)
RealityPRK string // REALITY x25519 PRIVATE key (for agent inbound TLS)
RealitySNI string
RealityShortID string
Endpoint string
Hy2Port sql.NullInt32
}
// Entitlement summarises a user's active plan for the connect gate.
type Entitlement struct {
DpUUID string
PlanCode string
AdGate bool // true = free plan, require ad unlock + minute quota
DailyMinutes sql.NullInt64
ExpiresAt sql.NullTime // latest subscription expiry (nil = trial/active)
}
// NodeStore is the persistence interface used by the nodes domain handlers.
@@ -27,6 +42,13 @@ type NodeStore interface {
// Returns (nil, nil) when no matching row exists.
NodeByUUID(ctx context.Context, uuid string) (*NodeRow, error)
// ListUp returns all nodes with status='up', ordered by weight DESC.
ListUp(ctx context.Context) ([]*NodeRow, error)
// EntitlementForUser returns the user's dp_uuid and active plan entitlement.
// Returns (nil, nil) when the user has no active subscription (treats as free).
EntitlementForUser(ctx context.Context, userID int64) (*Entitlement, error)
// ConfigVersion returns the current global directory version.
// This is the version from the directory_version singleton table.
ConfigVersion(ctx context.Context) (int64, error)
@@ -36,9 +58,14 @@ type NodeStore interface {
ActiveNodeUUIDs(ctx context.Context) ([]string, error)
// CredentialsForNode returns the active data-plane credentials for nodeUUID.
// Returns an empty slice until task 5d creates the connect_credentials table.
CredentialsForNode(ctx context.Context, nodeUUID string) ([]*agentv1.Credential, error)
// PersistCredential upserts a credential row in connect_credentials.
PersistCredential(ctx context.Context, nodeID int64, cred *agentv1.Credential, expiresAt time.Time) error
// DeleteCredential removes the credential for (nodeID, dpUUID).
DeleteCredential(ctx context.Context, nodeID int64, dpUUID string) error
// UserIDByDpUUID maps a data-plane UUID to the owning user's internal ID.
// Returns (0, false, nil) if the dp_uuid is unknown or the user is inactive.
UserIDByDpUUID(ctx context.Context, dpUUID string) (int64, bool, error)
@@ -62,14 +89,17 @@ func NewSQLNodeStore(db *sql.DB) *SQLNodeStore {
// NodeByUUID looks up a node by UUID. Returns (nil, nil) when not found.
func (s *SQLNodeStore) NodeByUUID(ctx context.Context, uuid string) (*NodeRow, error) {
const q = `
SELECT id, uuid, status, reality_pbk, reality_sni, endpoint, hy2_port
SELECT id, uuid, region, name_zh, name_en, tier, status,
reality_pbk, reality_prk, reality_sni, reality_short_id,
endpoint, hy2_port
FROM nodes
WHERE uuid = ?
`
var n NodeRow
err := s.db.QueryRowContext(ctx, q, uuid).Scan(
&n.ID, &n.UUID, &n.Status,
&n.RealityPBK, &n.RealitySNI, &n.Endpoint, &n.Hy2Port,
&n.ID, &n.UUID, &n.Region, &n.NameZH, &n.NameEN, &n.Tier, &n.Status,
&n.RealityPBK, &n.RealityPRK, &n.RealitySNI, &n.RealityShortID,
&n.Endpoint, &n.Hy2Port,
)
if err == sql.ErrNoRows {
return nil, nil
@@ -80,6 +110,77 @@ func (s *SQLNodeStore) NodeByUUID(ctx context.Context, uuid string) (*NodeRow, e
return &n, nil
}
// ListUp returns nodes with status='up', ordered by weight DESC.
func (s *SQLNodeStore) ListUp(ctx context.Context) ([]*NodeRow, error) {
const q = `
SELECT id, uuid, region, name_zh, name_en, tier, status,
reality_pbk, reality_prk, reality_sni, reality_short_id,
endpoint, hy2_port
FROM nodes
WHERE status = 'up'
ORDER BY weight DESC
`
rows, err := s.db.QueryContext(ctx, q)
if err != nil {
return nil, fmt.Errorf("nodes.SQLNodeStore.ListUp: %w", err)
}
defer rows.Close()
var out []*NodeRow
for rows.Next() {
var n NodeRow
if err := rows.Scan(
&n.ID, &n.UUID, &n.Region, &n.NameZH, &n.NameEN, &n.Tier, &n.Status,
&n.RealityPBK, &n.RealityPRK, &n.RealitySNI, &n.RealityShortID,
&n.Endpoint, &n.Hy2Port,
); err != nil {
return nil, fmt.Errorf("nodes.SQLNodeStore.ListUp scan: %w", err)
}
out = append(out, &n)
}
return out, rows.Err()
}
// EntitlementForUser returns the user's dp_uuid and best active plan entitlement.
// Picks the subscription with the latest expires_at; falls back to the 'free' plan
// if the user has no active subscription.
func (s *SQLNodeStore) EntitlementForUser(ctx context.Context, userID int64) (*Entitlement, error) {
// First get dp_uuid from the users table.
var dpUUID string
if err := s.db.QueryRowContext(ctx,
`SELECT dp_uuid FROM users WHERE id = ? AND status = 'active'`, userID,
).Scan(&dpUUID); err == sql.ErrNoRows {
return nil, nil
} else if err != nil {
return nil, fmt.Errorf("nodes.SQLNodeStore.EntitlementForUser: dp_uuid: %w", err)
}
// Look up the best active subscription.
const q = `
SELECT p.code, p.ad_gate, p.daily_minutes, s.expires_at
FROM subscriptions s
JOIN plans p ON p.id = s.plan_id
WHERE s.user_id = ? AND s.expires_at > UTC_TIMESTAMP()
ORDER BY s.expires_at DESC
LIMIT 1
`
e := &Entitlement{DpUUID: dpUUID}
err := s.db.QueryRowContext(ctx, q, userID).Scan(
&e.PlanCode, &e.AdGate, &e.DailyMinutes, &e.ExpiresAt,
)
if err == sql.ErrNoRows {
// No active subscription → free plan defaults.
e.PlanCode = "free"
e.AdGate = true
e.DailyMinutes = sql.NullInt64{Valid: true, Int64: 10}
return e, nil
}
if err != nil {
return nil, fmt.Errorf("nodes.SQLNodeStore.EntitlementForUser: plan: %w", err)
}
return e, nil
}
// ConfigVersion returns the current global directory version.
// Returns 0 with nil error if the directory_version row does not yet exist.
func (s *SQLNodeStore) ConfigVersion(ctx context.Context) (int64, error) {
@@ -115,10 +216,58 @@ func (s *SQLNodeStore) ActiveNodeUUIDs(ctx context.Context) ([]string, error) {
return uuids, rows.Err()
}
// CredentialsForNode returns active credentials for nodeUUID.
// Stub: returns empty slice until task 5d creates the connect_credentials table.
func (s *SQLNodeStore) CredentialsForNode(_ context.Context, _ string) ([]*agentv1.Credential, error) {
return nil, nil
// CredentialsForNode returns active (non-expired) credentials for nodeUUID.
func (s *SQLNodeStore) CredentialsForNode(ctx context.Context, nodeUUID string) ([]*agentv1.Credential, error) {
const q = `
SELECT cc.dp_uuid, cc.protocol, cc.flow, UNIX_TIMESTAMP(cc.expires_at)
FROM connect_credentials cc
JOIN nodes n ON n.id = cc.node_id
WHERE n.uuid = ? AND cc.expires_at > UTC_TIMESTAMP()
`
rows, err := s.db.QueryContext(ctx, q, nodeUUID)
if err != nil {
return nil, fmt.Errorf("nodes.SQLNodeStore.CredentialsForNode: %w", err)
}
defer rows.Close()
var out []*agentv1.Credential
for rows.Next() {
var c agentv1.Credential
if err := rows.Scan(&c.DpUUID, &c.Protocol, &c.Flow, &c.ExpiresAtUnix); err != nil {
return nil, fmt.Errorf("nodes.SQLNodeStore.CredentialsForNode scan: %w", err)
}
out = append(out, &c)
}
return out, rows.Err()
}
// PersistCredential upserts the credential into connect_credentials.
func (s *SQLNodeStore) PersistCredential(ctx context.Context, nodeID int64, cred *agentv1.Credential, expiresAt time.Time) error {
const q = `
INSERT INTO connect_credentials (node_id, dp_uuid, protocol, flow, expires_at)
VALUES (?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
protocol = VALUES(protocol),
flow = VALUES(flow),
expires_at = VALUES(expires_at)
`
if _, err := s.db.ExecContext(ctx, q,
nodeID, cred.DpUUID, int32(cred.Protocol), cred.Flow, expiresAt.UTC(),
); err != nil {
return fmt.Errorf("nodes.SQLNodeStore.PersistCredential: %w", err)
}
return nil
}
// DeleteCredential removes the credential for (nodeID, dpUUID).
func (s *SQLNodeStore) DeleteCredential(ctx context.Context, nodeID int64, dpUUID string) error {
if _, err := s.db.ExecContext(ctx,
`DELETE FROM connect_credentials WHERE node_id = ? AND dp_uuid = ?`,
nodeID, dpUUID,
); err != nil {
return fmt.Errorf("nodes.SQLNodeStore.DeleteCredential: %w", err)
}
return nil
}
// UserIDByDpUUID resolves a data-plane UUID to an active user's internal ID.