feat(routing): connect 读 per-user 档案传入渲染(fail-safe 回退默认)

2026-07-28 08:11:48 +08:00
parent f76aa56929
commit 82b3988974
4 changed files with 230 additions and 7 deletions
@@ -16,6 +16,7 @@ import (
 	"github.com/wangjia/pangolin/server/internal/auth"
 	"github.com/wangjia/pangolin/server/internal/nodes"
 	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+	"github.com/wangjia/pangolin/server/internal/routing"
 )

 const (
@@ -45,13 +46,17 @@ type NodeAPI struct {
 	// privateSplitDomains 私有服务域名(PANGOLIN_PRIVATE_SPLIT_DOMAINS,逗号分隔),
 	// 见 ClientConfigOpts.PrivateSplitDomains;空则不渲染相关规则。
 	privateSplitDomains []string
+	// routingStore 读取用户可配置分流档案(Task 1)。nil = 未注入(旧路径/未迁移
+	// 调用点),ConnectNode 回退到 ?split_cn query 兜底,不产坏配置。
+	routingStore *routing.Store
 }

 // 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, privateSplitDomains []string) *NodeAPI {
+// data-plane healthy — agent gRPC liveness still gates status). routingStore
+// may be nil (per-user 路由档案不生效,ConnectNode 回退到 ?split_cn query)。
+func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, load nodeLoadReader, deriveKey, rulesBaseURL string, privateSplitDomains []string, routingStore *routing.Store) *NodeAPI {
 	return &NodeAPI{store: store, hub: hub, load: load, deriveKey: deriveKey,
-		rulesBaseURL: rulesBaseURL, privateSplitDomains: privateSplitDomains}
+		rulesBaseURL: rulesBaseURL, privateSplitDomains: privateSplitDomains, routingStore: routingStore}
 }

 // dataPlaneHealthy reports the node's last sing-box health (default true when
@@ -316,10 +321,23 @@ func (a *NodeAPI) ConnectNode(w http.ResponseWriter, r *http.Request) {
 	}

 	// 7. Render and return the full sing-box CLIENT config JSON.
-	//    split_cn=1/true → 国内 IP/域名直连(#5);客户端按 smartRoute 偏好传。
+	//    可配置分流(Phase 1):读取用户路由档案,读取/解析失败一律回退 prof=nil
+	//    (fail-safe,不产坏配置)。有档案时国内分流开关由档案决定;无档案(未迁移
+	//    用户/store 未注入)保留 ?split_cn query 兜底旧行为。
+	var prof *routing.Profile
+	if a.routingStore != nil {
+		if p, perr := a.routingStore.Get(r.Context(), uid); perr == nil {
+			prof = p
+		} else {
+			slog.Warn("connect: routing profile load failed, falling back to default", "user", uid, "err", perr)
+		}
+	}
 	splitCN := r.URL.Query().Get("split_cn") == "1" || r.URL.Query().Get("split_cn") == "true"
+	if prof != nil {
+		splitCN = prof.Mode == "rule" && prof.Builtin.ChinaDirect
+	}
 	cfgJSON, renderErr := BuildClientConfig(node, dpUUID, a.deriveKey,
-		ClientConfigOpts{SplitCN: splitCN, RulesBaseURL: a.rulesBaseURL,
+		ClientConfigOpts{Profile: prof, SplitCN: splitCN, RulesBaseURL: a.rulesBaseURL,
 			PrivateSplitDomains: a.privateSplitDomains})
 	if renderErr != nil {
 		slog.Error("connect: build client config failed", "node", nodeUUID, "err", renderErr)
@@ -0,0 +1,205 @@
+package httpapi
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/go-chi/chi/v5"
+	"github.com/redis/go-redis/v9"
+
+	"github.com/wangjia/pangolin/server/internal/codes"
+	"github.com/wangjia/pangolin/server/internal/config"
+	"github.com/wangjia/pangolin/server/internal/nodes"
+	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
+	"github.com/wangjia/pangolin/server/internal/routing"
+	"github.com/wangjia/pangolin/server/internal/store"
+)
+
+// fakeConnectStore implements just the NodeStore methods ConnectNode's happy
+// path touches when the entitlement disables the optional gates (AdGate=false,
+// MaxDevices=0, DailyMB invalid) — everything else panics via the embedded nil
+// interface (未用到即安全，同 nodes_disconnect_test.go 的写法)。
+type fakeConnectStore struct {
+	nodes.NodeStore
+
+	node  *nodes.NodeRow
+	ent   *nodes.Entitlement
+	devDp string
+}
+
+func (f *fakeConnectStore) EntitlementForUser(context.Context, int64) (*nodes.Entitlement, error) {
+	return f.ent, nil
+}
+
+func (f *fakeConnectStore) NodeByUUID(context.Context, string) (*nodes.NodeRow, error) {
+	return f.node, nil
+}
+
+func (f *fakeConnectStore) EnsureDeviceDpUUID(context.Context, int64, string) (string, int64, error) {
+	return f.devDp, 1, nil
+}
+
+func (f *fakeConnectStore) PersistCredential(context.Context, int64, *agentv1.Credential, time.Time) error {
+	return nil
+}
+
+// openRoutingDB opens an in-memory SQLite DB migrated up, for a real
+// routing.Store in tests (same pattern as internal/routing/store_sqlite_test.go).
+func openRoutingDB(t *testing.T) *sql.DB {
+	t.Helper()
+	db, err := store.Open(&config.Config{Driver: "sqlite", DSN: ":memory:"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	if err := store.MigrateUp(db, "sqlite"); err != nil {
+		t.Fatal(err)
+	}
+	return db
+}
+
+// newOnlineHub returns a real *nodes.Hub (backed by miniredis) with nodeUUID
+// registered online, so ConnectNode's a.hub.IsOnline(...) gate passes and
+// Push(...) succeeds without a real Redis deployment.
+func newOnlineHub(t *testing.T, nodeUUID string) *nodes.Hub {
+	t.Helper()
+	mr := miniredis.RunT(t)
+	rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
+	t.Cleanup(func() { rdb.Close() })
+	hub := nodes.NewHub(rdb)
+	_, done := hub.Register(nodeUUID)
+	t.Cleanup(done)
+	return hub
+}
+
+func doConnect(t *testing.T, api *NodeAPI, uid int64) *httptest.ResponseRecorder {
+	t.Helper()
+	req := httptest.NewRequest("POST", "/v1/nodes/node-1/connect", strings.NewReader(`{"device_id":"dev-1"}`))
+	rctx := chi.NewRouteContext()
+	rctx.URLParams.Add("id", "node-1")
+	ctx := context.WithValue(req.Context(), chi.RouteCtxKey, rctx)
+	ctx = context.WithValue(ctx, codes.CtxKeyUserID, uid)
+	rec := httptest.NewRecorder()
+	api.ConnectNode(rec, req.WithContext(ctx))
+	return rec
+}
+
+// TestConnectNode_ReadsRoutingProfile is Task 5's Step 1: a profile with a
+// user rule is stored for the connecting user; connect must read it (via
+// routingStore.Get) and forward it into BuildClientConfig — the returned
+// sing-box config's route.rules must contain that user rule.
+func TestConnectNode_ReadsRoutingProfile(t *testing.T) {
+	const uid = int64(42)
+
+	db := openRoutingDB(t)
+	rst := routing.NewStore(db)
+	p := routing.Default()
+	p.Rules = []routing.Rule{
+		{Type: "domain_suffix", Value: "github.com", Action: "proxy", Enabled: true},
+	}
+	if err := rst.Upsert(context.Background(), uid, p); err != nil {
+		t.Fatal(err)
+	}
+
+	hub := newOnlineHub(t, "node-1")
+	nodeStore := &fakeConnectStore{
+		node: &nodes.NodeRow{ID: 1, UUID: "node-1", Status: "up", Endpoint: "1.2.3.4:443",
+			RealityPBK: "pbk", RealityShortID: "sid", RealitySNI: "www.apple.com"},
+		ent:   &nodes.Entitlement{DpUUID: "acct-dp"}, // AdGate=false, MaxDevices=0, DailyMB invalid → 跳过额外配额查询
+		devDp: "device-dp",
+	}
+	api := NewNodeAPI(nodeStore, hub, nil, "k", "http://x", nil, rst)
+
+	rec := doConnect(t, api, uid)
+	if rec.Code != 200 {
+		t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
+	}
+
+	var cfg map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &cfg); err != nil {
+		t.Fatalf("unmarshal config: %v (body=%s)", err, rec.Body.String())
+	}
+	rules, _ := cfg["route"].(map[string]any)["rules"].([]any)
+	if ruleIndexByDomain(rules, "github.com") < 0 {
+		t.Fatalf("connect config missing user routing rule for github.com; rules=%v", rules)
+	}
+}
+
+// TestConnectNode_NilRoutingStore_FailsSafe: routingStore not injected (nil,
+// e.g. an unmigrated call site) must not error/panic — connect renders the
+// old default (prof=nil), falling back to the ?split_cn query.
+func TestConnectNode_NilRoutingStore_FailsSafe(t *testing.T) {
+	const uid = int64(43)
+
+	hub := newOnlineHub(t, "node-1")
+	nodeStore := &fakeConnectStore{
+		node: &nodes.NodeRow{ID: 1, UUID: "node-1", Status: "up", Endpoint: "1.2.3.4:443",
+			RealityPBK: "pbk", RealityShortID: "sid", RealitySNI: "www.apple.com"},
+		ent:   &nodes.Entitlement{DpUUID: "acct-dp"},
+		devDp: "device-dp",
+	}
+	api := NewNodeAPI(nodeStore, hub, nil, "k", "http://x", nil, nil) // routingStore=nil
+
+	rec := doConnect(t, api, uid)
+	if rec.Code != 200 {
+		t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
+	}
+	var cfg map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &cfg); err != nil {
+		t.Fatalf("unmarshal config: %v (body=%s)", err, rec.Body.String())
+	}
+	// 无档案、无 ?split_cn query → 与既有行为一致:不分流,route 里无 rule_set。
+	if _, ok := cfg["route"].(map[string]any)["rule_set"]; ok {
+		t.Errorf("no profile + no split_cn query should have no rule_set, got %v", cfg["route"])
+	}
+}
+
+// TestConnectNode_RoutingStoreErr_FailsSafe: routingStore.Get erroring (e.g.
+// corrupt profile_json) must not fail the connect — falls back to prof=nil,
+// same as no profile at all, and still honors the ?split_cn query fallback.
+func TestConnectNode_RoutingStoreErr_FailsSafe(t *testing.T) {
+	const uid = int64(44)
+
+	db := openRoutingDB(t)
+	// 写入一条无法反序列化的 profile_json，模拟 Get 出错。
+	if _, err := db.Exec(`INSERT INTO routing_profiles (user_id, profile_json, updated_at) VALUES (?,?,?)`,
+		uid, "{not-json", time.Now().UTC()); err != nil {
+		t.Fatal(err)
+	}
+	rst := routing.NewStore(db)
+
+	hub := newOnlineHub(t, "node-1")
+	nodeStore := &fakeConnectStore{
+		node: &nodes.NodeRow{ID: 1, UUID: "node-1", Status: "up", Endpoint: "1.2.3.4:443",
+			RealityPBK: "pbk", RealityShortID: "sid", RealitySNI: "www.apple.com"},
+		ent:   &nodes.Entitlement{DpUUID: "acct-dp"},
+		devDp: "device-dp",
+	}
+	api := NewNodeAPI(nodeStore, hub, nil, "k", "http://x", nil, rst)
+
+	req := httptest.NewRequest("POST", "/v1/nodes/node-1/connect?split_cn=1", strings.NewReader(`{"device_id":"dev-1"}`))
+	rctx := chi.NewRouteContext()
+	rctx.URLParams.Add("id", "node-1")
+	ctx := context.WithValue(req.Context(), chi.RouteCtxKey, rctx)
+	ctx = context.WithValue(ctx, codes.CtxKeyUserID, uid)
+	rec := httptest.NewRecorder()
+	api.ConnectNode(rec, req.WithContext(ctx))
+
+	if rec.Code != 200 {
+		t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
+	}
+	var cfg map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &cfg); err != nil {
+		t.Fatalf("unmarshal config: %v (body=%s)", err, rec.Body.String())
+	}
+	// prof=nil(读档案出错回退)+ ?split_cn=1 query 兜底 → 仍走国内分流(旧行为)。
+	if _, ok := cfg["route"].(map[string]any)["rule_set"]; !ok {
+		t.Errorf("routing store error should fall back to ?split_cn query behavior (rule_set expected), got %v", cfg["route"])
+	}
+}
@@ -51,7 +51,7 @@ func (f *fakeDisconnectStore) PersistCredential(context.Context, int64, *agentv1

 func doDisconnect(t *testing.T, store *fakeDisconnectStore, body string) int {
 	t.Helper()
-	api := NewNodeAPI(store, nil, nil, "", "", nil) // nil hub:跳过 Push,只验证凭证删除
+	api := NewNodeAPI(store, nil, nil, "", "", nil, nil) // nil hub:跳过 Push,只验证凭证删除
 	req := httptest.NewRequest("POST", "/v1/nodes/node-1/disconnect", strings.NewReader(body))
 	rctx := chi.NewRouteContext()
 	rctx.URLParams.Add("id", "node-1")