feat(server/stats): stats-overhaul Phase2 — 每设备归因 + GB 综合配额

服务端记账从「账户」细到「每设备」(每设备独立 dp_uuid),配额单位分钟→GB 按账户综合卡控。000015_per_device_usage 迁移(mysql+sqlite 双份):devices.dp_uuid + usage_device_daily 表 + plans.daily_mb。handler_grpc 按 dp_uuid 回映射 (user_id,device_id) 双写账户+每设备;usage 服务/handler 暴露 /v1/usage(/devices)。含 sqlite_per_device / usage handler 测试。注:此迁移 prod 已 migrate up 运行、部署二进制已内嵌;本次补提交使 git 与线上一致。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-28 18:12:00 +08:00
parent 4f9d2d2cf3
commit 636a3bbf2f
14 changed files with 778 additions and 8 deletions
@@ -37,6 +37,9 @@ type mockNodeStore struct {
 	configVer     int64
 	usageAccum    []mockUsageEntry
 	usersByDpUUID map[string]int64
+	// devicesByDpUUID maps a per-device dp_uuid → {userID, deviceID}.
+	devicesByDpUUID  map[string][2]int64
+	deviceUsageAccum []mockDeviceUsageEntry
 }

 type mockUsageEntry struct {
@@ -47,6 +50,15 @@ type mockUsageEntry struct {
 	Minutes   int64
 }

+type mockDeviceUsageEntry struct {
+	UserID    int64
+	DeviceID  int64
+	Date      time.Time
+	BytesUp   int64
+	BytesDown int64
+	Minutes   int64
+}
+
 func (m *mockNodeStore) NodeByUUID(_ context.Context, uuid string) (*nodes.NodeRow, error) {
 	if uuid == m.nodeUUID {
 		return &nodes.NodeRow{
@@ -105,6 +117,44 @@ func (m *mockNodeStore) AccumulateUsage(_ context.Context, userID int64, date ti
 	return nil
 }

+func (m *mockNodeStore) EnsureDeviceDpUUID(_ context.Context, _ int64, _ string) (string, int64, error) {
+	return "", 0, nil
+}
+
+func (m *mockNodeStore) AccountDayBytes(_ context.Context, _ int64, _ time.Time) (int64, error) {
+	return 0, nil
+}
+
+func (m *mockNodeStore) UserDeviceByDpUUID(_ context.Context, dpUUID string) (int64, int64, bool, error) {
+	if m.devicesByDpUUID != nil {
+		if ud, ok := m.devicesByDpUUID[dpUUID]; ok {
+			return ud[0], ud[1], true, nil
+		}
+	}
+	if uid, ok := m.usersByDpUUID[dpUUID]; ok {
+		return uid, 0, true, nil // legacy account credential, no device dimension
+	}
+	return 0, 0, false, nil
+}
+
+func (m *mockNodeStore) AccumulateDeviceUsage(_ context.Context, userID, deviceID int64, date time.Time,
+	bytesUp, bytesDown, minutes int64,
+) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.deviceUsageAccum = append(m.deviceUsageAccum,
+		mockDeviceUsageEntry{userID, deviceID, date, bytesUp, bytesDown, minutes})
+	return nil
+}
+
+func (m *mockNodeStore) deviceUsageLog() []mockDeviceUsageEntry {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	out := make([]mockDeviceUsageEntry, len(m.deviceUsageAccum))
+	copy(out, m.deviceUsageAccum)
+	return out
+}
+
 func (m *mockNodeStore) usageLog() []mockUsageEntry {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -850,4 +900,51 @@ func TestReportUsage_Accumulates(t *testing.T) {
 		t.Errorf("bytes_up=%d bytes_down=%d, want 100/200",
 			log[0].BytesUp, log[0].BytesDown)
 	}
+	// Account-level dp_uuid → no per-device attribution (deviceID 0).
+	if dev := b.store.deviceUsageLog(); len(dev) != 0 {
+		t.Errorf("device usage should be empty for account credential, got %d", len(dev))
+	}
+}
+
+// TestReportUsage_PerDevice verifies a per-device dp_uuid is dual-written: account
+// rollup (usage_daily) AND per-device attribution (usage_device_daily).
+func TestReportUsage_PerDevice(t *testing.T) {
+	const nodeUUID = "test-node-usage-dev"
+	b := newTestServer(t, 1, nodeUUID)
+	ctx := context.Background()
+	// dp-dev1 resolves to user 101, device 55 (per-device credential).
+	b.store.devicesByDpUUID = map[string][2]int64{"dp-dev1": {101, 55}}
+
+	_, _, conn := enrollNode(t, b, nodeUUID)
+	client := agentv1.NewAgentServiceClient(conn)
+	if _, err := client.Register(ctx, &agentv1.RegisterRequest{NodeUUID: nodeUUID}); err != nil {
+		t.Fatalf("Register: %v", err)
+	}
+
+	now := time.Now()
+	if _, err := client.ReportUsage(ctx, &agentv1.UsageReport{
+		NodeUUID:        nodeUUID,
+		WindowStartUnix: now.Add(-time.Minute).Unix(),
+		WindowEndUnix:   now.Unix(),
+		Entries: []*agentv1.UsageEntry{
+			{DpUUID: "dp-dev1", BytesUp: 100, BytesDown: 200, SessionMinutes: 3},
+		},
+	}); err != nil {
+		t.Fatalf("ReportUsage: %v", err)
+	}
+
+	// Account rollup still recorded.
+	acc := b.store.usageLog()
+	if len(acc) != 1 || acc[0].UserID != 101 || acc[0].BytesUp != 100 {
+		t.Fatalf("account usage wrong: %+v", acc)
+	}
+	// Per-device usage recorded with device attribution.
+	dev := b.store.deviceUsageLog()
+	if len(dev) != 1 {
+		t.Fatalf("device usage has %d entries, want 1", len(dev))
+	}
+	if dev[0].UserID != 101 || dev[0].DeviceID != 55 ||
+		dev[0].BytesUp != 100 || dev[0].BytesDown != 200 || dev[0].Minutes != 3 {
+		t.Errorf("device usage wrong: %+v", dev[0])
+	}
 }
@@ -293,7 +293,7 @@ func (h *Handler) ReportUsage(ctx context.Context, req *agentv1.UsageReport) (*a
 		if entry.DpUUID == "" {
 			continue
 		}
-		userID, found, err := h.store.UserIDByDpUUID(ctx, entry.DpUUID)
+		userID, deviceID, found, err := h.store.UserDeviceByDpUUID(ctx, entry.DpUUID)
 		if err != nil {
 			slog.Warn("nodes.Handler.ReportUsage: dp_uuid lookup failed",
 				"dp_uuid", entry.DpUUID, "err", err)
@@ -302,12 +302,23 @@ func (h *Handler) ReportUsage(ctx context.Context, req *agentv1.UsageReport) (*a
 		if !found {
 			continue
 		}
+		// Account-level rollup (always).
 		if err := h.store.AccumulateUsage(ctx, userID, date,
 			entry.BytesUp, entry.BytesDown, entry.SessionMinutes,
 		); err != nil {
 			slog.Warn("nodes.Handler.ReportUsage: accumulate failed",
 				"user_id", userID, "err", err)
 		}
+		// Per-device attribution (only when the dp_uuid maps to a registered device;
+		// deviceID==0 means a legacy account-level credential — no device dimension).
+		if deviceID > 0 {
+			if err := h.store.AccumulateDeviceUsage(ctx, userID, deviceID, date,
+				entry.BytesUp, entry.BytesDown, entry.SessionMinutes,
+			); err != nil {
+				slog.Warn("nodes.Handler.ReportUsage: device accumulate failed",
+					"user_id", userID, "device_id", deviceID, "err", err)
+			}
+		}
 	}
 	return &agentv1.UsageAck{}, nil
 }
@@ -7,6 +7,7 @@ import (
 	"time"

 	dbx "github.com/wangjia/pangolin/server/internal/db"
+	"github.com/wangjia/pangolin/server/internal/idgen"
 	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
 )

@@ -33,7 +34,8 @@ type Entitlement struct {
 	PlanCode     string
 	AdGate       bool // true = free plan, require ad unlock + minute quota
 	DailyMinutes sql.NullInt64
-	ExpiresAt    sql.NullTime // latest subscription expiry (nil = trial/active)
+	DailyMB      sql.NullInt64 // 每日综合流量配额(MB, 按账户综合卡); NULL = 不限
+	ExpiresAt    sql.NullTime  // latest subscription expiry (nil = trial/active)
 }

 // NodeStore is the persistence interface used by the nodes domain handlers.
@@ -71,10 +73,31 @@ type NodeStore interface {
 	// Returns (0, false, nil) if the dp_uuid is unknown or the user is inactive.
 	UserIDByDpUUID(ctx context.Context, dpUUID string) (int64, bool, error)

+	// EnsureDeviceDpUUID returns the per-device data-plane UUID for (userID,
+	// deviceUUID), minting + persisting one (devices.dp_uuid) if the device has
+	// none yet. Also returns the device's internal id. Each device gets its own
+	// dp_uuid so the node reports per-device traffic counters (todo #5 Phase 2).
+	EnsureDeviceDpUUID(ctx context.Context, userID int64, deviceUUID string) (dpUUID string, deviceID int64, err error)
+
+	// UserDeviceByDpUUID resolves a data-plane UUID to (userID, deviceID). It
+	// prefers the per-device credential (devices.dp_uuid); failing that it falls
+	// back to the account-level users.dp_uuid (deviceID=0) for legacy compat.
+	// Returns ok=false when unknown or the user is inactive.
+	UserDeviceByDpUUID(ctx context.Context, dpUUID string) (userID, deviceID int64, ok bool, err error)
+
 	// AccumulateUsage adds bytes and minutes to usage_daily for userID on date.
 	// Uses INSERT … ON DUPLICATE KEY UPDATE (idempotent within a day).
 	AccumulateUsage(ctx context.Context, userID int64, date time.Time,
 		bytesUp, bytesDown int64, minutes int64) error
+
+	// AccountDayBytes returns the account's total bytes (up+down) for userID on
+	// date — the basis for the GB 综合配额 connect gate. 0 when no usage yet.
+	AccountDayBytes(ctx context.Context, userID int64, date time.Time) (int64, error)
+
+	// AccumulateDeviceUsage adds bytes/minutes to usage_device_daily for
+	// (deviceID, date); user_id is carried for per-account rollups/queries.
+	AccumulateDeviceUsage(ctx context.Context, userID, deviceID int64, date time.Time,
+		bytesUp, bytesDown int64, minutes int64) error
 }

 // SQLNodeStore implements NodeStore against a SQL database (MySQL or SQLite).
@@ -159,7 +182,7 @@ func (s *SQLNodeStore) EntitlementForUser(ctx context.Context, userID int64) (*E

 	// Look up the best active subscription.
 	const q = `
-		SELECT p.code, p.ad_gate, p.daily_minutes, s.expires_at
+		SELECT p.code, p.ad_gate, p.daily_minutes, p.daily_mb, s.expires_at
 		FROM subscriptions s
 		JOIN plans p ON p.id = s.plan_id
 		WHERE s.user_id = ? AND s.expires_at > ?
@@ -168,13 +191,14 @@ func (s *SQLNodeStore) EntitlementForUser(ctx context.Context, userID int64) (*E
 	`
 	e := &Entitlement{DpUUID: dpUUID}
 	err := s.db.QueryRowContext(ctx, q, userID, time.Now().UTC()).Scan(
-		&e.PlanCode, &e.AdGate, &e.DailyMinutes, &e.ExpiresAt,
+		&e.PlanCode, &e.AdGate, &e.DailyMinutes, &e.DailyMB, &e.ExpiresAt,
 	)
 	if err == sql.ErrNoRows {
-		// No active subscription → free plan defaults.
+		// No active subscription → free plan defaults (mirrors the free plan seed).
 		e.PlanCode = "free"
 		e.AdGate = true
 		e.DailyMinutes = sql.NullInt64{Valid: true, Int64: 10}
+		e.DailyMB = sql.NullInt64{Valid: true, Int64: 500}
 		return e, nil
 	}
 	if err != nil {
@@ -288,6 +312,106 @@ func (s *SQLNodeStore) UserIDByDpUUID(ctx context.Context, dpUUID string) (int64
 	return userID, true, nil
 }

+// EnsureDeviceDpUUID returns (and lazily mints) the per-device dp_uuid for the
+// (userID, deviceUUID) pair. Minting uses a guarded UPDATE so concurrent connects
+// for the same device converge on a single dp_uuid (the loser's UPDATE is a no-op
+// and the re-read returns the winner's value).
+func (s *SQLNodeStore) EnsureDeviceDpUUID(ctx context.Context, userID int64, deviceUUID string) (string, int64, error) {
+	var deviceID int64
+	var dpUUID sql.NullString
+	err := s.db.QueryRowContext(ctx,
+		`SELECT id, dp_uuid FROM devices WHERE user_id = ? AND uuid = ? LIMIT 1`,
+		userID, deviceUUID,
+	).Scan(&deviceID, &dpUUID)
+	if err == sql.ErrNoRows {
+		return "", 0, fmt.Errorf("nodes.SQLNodeStore.EnsureDeviceDpUUID: device %q not found for user %d", deviceUUID, userID)
+	}
+	if err != nil {
+		return "", 0, fmt.Errorf("nodes.SQLNodeStore.EnsureDeviceDpUUID: %w", err)
+	}
+	if dpUUID.Valid && dpUUID.String != "" {
+		return dpUUID.String, deviceID, nil
+	}
+	minted := idgen.NewString()
+	if _, err := s.db.ExecContext(ctx,
+		`UPDATE devices SET dp_uuid = ? WHERE id = ? AND (dp_uuid IS NULL OR dp_uuid = '')`,
+		minted, deviceID,
+	); err != nil {
+		return "", 0, fmt.Errorf("nodes.SQLNodeStore.EnsureDeviceDpUUID: mint: %w", err)
+	}
+	// Re-read to resolve any concurrent mint race deterministically.
+	if err := s.db.QueryRowContext(ctx,
+		`SELECT dp_uuid FROM devices WHERE id = ?`, deviceID,
+	).Scan(&dpUUID); err != nil {
+		return "", 0, fmt.Errorf("nodes.SQLNodeStore.EnsureDeviceDpUUID: reread: %w", err)
+	}
+	return dpUUID.String, deviceID, nil
+}
+
+// UserDeviceByDpUUID resolves a dp_uuid to (userID, deviceID), preferring the
+// per-device credential and falling back to the legacy account-level users.dp_uuid.
+func (s *SQLNodeStore) UserDeviceByDpUUID(ctx context.Context, dpUUID string) (int64, int64, bool, error) {
+	var userID, deviceID int64
+	err := s.db.QueryRowContext(ctx,
+		`SELECT d.user_id, d.id FROM devices d
+		 JOIN users u ON u.id = d.user_id
+		 WHERE d.dp_uuid = ? AND u.status = 'active' LIMIT 1`, dpUUID,
+	).Scan(&userID, &deviceID)
+	if err == nil {
+		return userID, deviceID, true, nil
+	}
+	if err != sql.ErrNoRows {
+		return 0, 0, false, fmt.Errorf("nodes.SQLNodeStore.UserDeviceByDpUUID: device: %w", err)
+	}
+	// Legacy fallback: account-level credential, no device attribution.
+	err = s.db.QueryRowContext(ctx,
+		`SELECT id FROM users WHERE dp_uuid = ? AND status = 'active' LIMIT 1`, dpUUID,
+	).Scan(&userID)
+	if err == sql.ErrNoRows {
+		return 0, 0, false, nil
+	}
+	if err != nil {
+		return 0, 0, false, fmt.Errorf("nodes.SQLNodeStore.UserDeviceByDpUUID: user: %w", err)
+	}
+	return userID, 0, true, nil
+}
+
+// AccumulateDeviceUsage adds bytes/minutes to usage_device_daily for (deviceID, date).
+func (s *SQLNodeStore) AccumulateDeviceUsage(
+	ctx context.Context, userID, deviceID int64, date time.Time,
+	bytesUp, bytesDown int64, minutes int64,
+) error {
+	q := `
+		INSERT INTO usage_device_daily (user_id, device_id, date, bytes_up, bytes_down, minutes_used)
+		VALUES (?, ?, ?, ?, ?, ?)
+		` + s.dialect.Upsert([]string{"device_id", "date"},
+		"bytes_up = bytes_up + EXCLUDED.bytes_up",
+		"bytes_down = bytes_down + EXCLUDED.bytes_down",
+		"minutes_used = minutes_used + EXCLUDED.minutes_used")
+	if _, err := s.db.ExecContext(ctx, q,
+		userID, deviceID, date.Format("2006-01-02"), bytesUp, bytesDown, minutes,
+	); err != nil {
+		return fmt.Errorf("nodes.SQLNodeStore.AccumulateDeviceUsage: %w", err)
+	}
+	return nil
+}
+
+// AccountDayBytes returns the account's total bytes (up+down) for the day.
+func (s *SQLNodeStore) AccountDayBytes(ctx context.Context, userID int64, date time.Time) (int64, error) {
+	var total sql.NullInt64
+	err := s.db.QueryRowContext(ctx,
+		`SELECT bytes_up + bytes_down FROM usage_daily WHERE user_id = ? AND date = ?`,
+		userID, date.Format("2006-01-02"),
+	).Scan(&total)
+	if err == sql.ErrNoRows {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, fmt.Errorf("nodes.SQLNodeStore.AccountDayBytes: %w", err)
+	}
+	return total.Int64, nil
+}
+
 // AccumulateUsage adds bytes/minutes to usage_daily for the given user and date.
 func (s *SQLNodeStore) AccumulateUsage(
 	ctx context.Context, userID int64, date time.Time,