diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 536221e..dde58e4 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -286,6 +286,7 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv usageStore := usage.NewStore(sqlDB) usageSvc := usage.NewService(usageStore, rdb, nil, time.Hour) usageHandler := usage.NewUsageHandler(usageSvc) + deviceUsageHandler := usage.NewDeviceUsageHandler(usageSvc) adsHandler := usage.NewAdsUnlockHandler(usageSvc) // ── Account / Plans / Notices ───────────────────────────────────────────── @@ -364,6 +365,7 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv }) protected.Post("/redeem", redeemHandler.ServeHTTP) protected.Get("/usage", usageHandler.ServeHTTP) + protected.Get("/usage/devices", deviceUsageHandler.ServeHTTP) protected.Post("/ads/unlock", adsHandler.ServeHTTP) protected.Get("/plans", accountAPI.ListPlans) protected.Get("/notices", accountAPI.ListNotices) diff --git a/server/internal/nodes/grpc_test.go b/server/internal/nodes/grpc_test.go index f26dd20..664fd38 100644 --- a/server/internal/nodes/grpc_test.go +++ b/server/internal/nodes/grpc_test.go @@ -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]) + } } diff --git a/server/internal/nodes/handler_grpc.go b/server/internal/nodes/handler_grpc.go index 0337fbf..54669be 100644 --- a/server/internal/nodes/handler_grpc.go +++ b/server/internal/nodes/handler_grpc.go @@ -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 } diff --git a/server/internal/nodes/store.go b/server/internal/nodes/store.go index 01f997c..09c651e 100644 --- a/server/internal/nodes/store.go +++ b/server/internal/nodes/store.go @@ -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, diff --git a/server/internal/store/sqlite_migrate_test.go b/server/internal/store/sqlite_migrate_test.go index 62dc53b..7640201 100644 --- a/server/internal/store/sqlite_migrate_test.go +++ b/server/internal/store/sqlite_migrate_test.go @@ -29,8 +29,8 @@ func TestSQLiteMigrateUpDown(t *testing.T) { if dirty { t.Fatalf("schema dirty after MigrateUp") } - if v != 14 { - t.Errorf("version = %d, want 14", v) + if v != 15 { + t.Errorf("version = %d, want 15", v) } // 2. Core tables exist. @@ -38,7 +38,7 @@ func TestSQLiteMigrateUpDown(t *testing.T) { "users", "devices", "plans", "subscriptions", "code_batches", "codes", "usage_daily", "audit_log", "providers", "nodes", "node_events", "directory_version", "provision_idempotency", "replacements", "admins", - "connect_credentials", + "connect_credentials", "usage_device_daily", } { var name string err := db.QueryRow( diff --git a/server/internal/store/sqlite_per_device_test.go b/server/internal/store/sqlite_per_device_test.go new file mode 100644 index 0000000..674b322 --- /dev/null +++ b/server/internal/store/sqlite_per_device_test.go @@ -0,0 +1,293 @@ +package store_test + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/wangjia/pangolin/server/internal/nodes" + "github.com/wangjia/pangolin/server/internal/usage" +) + +// seedDevice inserts a device row with a NULL dp_uuid (awaiting lazy mint). +func seedDevice(t *testing.T, db *sql.DB, id, userID int64, uuid string) { + t.Helper() + if _, err := db.Exec( + `INSERT INTO devices (id, uuid, user_id, name, platform, created_at) + VALUES (?, ?, ?, ?, 'ios', ?)`, + id, uuid, userID, "dev"+uuid, time.Now().UTC()); err != nil { + t.Fatalf("seed device %s: %v", uuid, err) + } +} + +// EnsureDeviceDpUUID mints a per-device dp_uuid, is idempotent for the same +// device, and yields distinct credentials for distinct devices. +func TestSQLite_EnsureDeviceDpUUID(t *testing.T) { + ctx := context.Background() + db := openSQLite(t) + ns := nodes.NewSQLNodeStore(db) + seedUser(t, db, 1) + seedDevice(t, db, 10, 1, "dev-a") + seedDevice(t, db, 11, 1, "dev-b") + + dpA, idA, err := ns.EnsureDeviceDpUUID(ctx, 1, "dev-a") + if err != nil { + t.Fatalf("ensure dev-a: %v", err) + } + if dpA == "" || idA != 10 { + t.Fatalf("dev-a: dp=%q id=%d, want non-empty/10", dpA, idA) + } + + // Idempotent: a second call returns the same dp_uuid (no re-mint). + dpA2, idA2, err := ns.EnsureDeviceDpUUID(ctx, 1, "dev-a") + if err != nil { + t.Fatalf("ensure dev-a again: %v", err) + } + if dpA2 != dpA || idA2 != 10 { + t.Errorf("not idempotent: dp=%q (want %q) id=%d", dpA2, dpA, idA2) + } + + // Distinct device → distinct dp_uuid. + dpB, idB, err := ns.EnsureDeviceDpUUID(ctx, 1, "dev-b") + if err != nil { + t.Fatalf("ensure dev-b: %v", err) + } + if dpB == dpA { + t.Errorf("dev-b dp_uuid collides with dev-a: %q", dpB) + } + if idB != 11 { + t.Errorf("dev-b id=%d, want 11", idB) + } + + // Persisted to devices.dp_uuid. + var stored string + if err := db.QueryRow(`SELECT dp_uuid FROM devices WHERE id=10`).Scan(&stored); err != nil { + t.Fatalf("read stored dp_uuid: %v", err) + } + if stored != dpA { + t.Errorf("stored dp_uuid=%q, want %q", stored, dpA) + } + + // Unknown device → error (not silently minted against nothing). + if _, _, err := ns.EnsureDeviceDpUUID(ctx, 1, "ghost"); err == nil { + t.Errorf("expected error for unknown device") + } +} + +// UserDeviceByDpUUID prefers the per-device credential and falls back to the +// legacy account-level users.dp_uuid (deviceID=0). +func TestSQLite_UserDeviceByDpUUID(t *testing.T) { + ctx := context.Background() + db := openSQLite(t) + ns := nodes.NewSQLNodeStore(db) + seedUser(t, db, 1) // seedUser sets users.dp_uuid='dp-u' + seedDevice(t, db, 10, 1, "dev-a") + + dpA, _, err := ns.EnsureDeviceDpUUID(ctx, 1, "dev-a") + if err != nil { + t.Fatalf("ensure: %v", err) + } + + // Per-device resolution. + uid, did, ok, err := ns.UserDeviceByDpUUID(ctx, dpA) + if err != nil || !ok { + t.Fatalf("resolve device dp: ok=%v err=%v", ok, err) + } + if uid != 1 || did != 10 { + t.Errorf("device dp → user=%d device=%d, want 1/10", uid, did) + } + + // Legacy account fallback (no device dimension). + uid, did, ok, err = ns.UserDeviceByDpUUID(ctx, "dp-u") + if err != nil || !ok { + t.Fatalf("resolve account dp: ok=%v err=%v", ok, err) + } + if uid != 1 || did != 0 { + t.Errorf("account dp → user=%d device=%d, want 1/0", uid, did) + } + + // Unknown dp_uuid. + if _, _, ok, _ := ns.UserDeviceByDpUUID(ctx, "nope"); ok { + t.Errorf("unknown dp_uuid resolved unexpectedly") + } +} + +// AccumulateDeviceUsage sums within (device, date) and keeps devices separate. +func TestSQLite_AccumulateDeviceUsage(t *testing.T) { + ctx := context.Background() + db := openSQLite(t) + ns := nodes.NewSQLNodeStore(db) + seedUser(t, db, 1) + seedDevice(t, db, 10, 1, "dev-a") + seedDevice(t, db, 11, 1, "dev-b") + day := time.Date(2026, 6, 24, 0, 0, 0, 0, time.UTC) + + if err := ns.AccumulateDeviceUsage(ctx, 1, 10, day, 100, 200, 1); err != nil { + t.Fatalf("accum a1: %v", err) + } + if err := ns.AccumulateDeviceUsage(ctx, 1, 10, day, 50, 25, 2); err != nil { + t.Fatalf("accum a2: %v", err) + } + if err := ns.AccumulateDeviceUsage(ctx, 1, 11, day, 7, 8, 9); err != nil { + t.Fatalf("accum b1: %v", err) + } + + var up, down, mins int64 + if err := db.QueryRow( + `SELECT bytes_up, bytes_down, minutes_used FROM usage_device_daily WHERE device_id=10`, + ).Scan(&up, &down, &mins); err != nil { + t.Fatalf("read dev-a: %v", err) + } + if up != 150 || down != 225 || mins != 3 { + t.Errorf("dev-a usage up=%d down=%d mins=%d, want 150/225/3", up, down, mins) + } + + if err := db.QueryRow( + `SELECT bytes_up, bytes_down, minutes_used FROM usage_device_daily WHERE device_id=11`, + ).Scan(&up, &down, &mins); err != nil { + t.Fatalf("read dev-b: %v", err) + } + if up != 7 || down != 8 || mins != 9 { + t.Errorf("dev-b usage up=%d down=%d mins=%d, want 7/8/9", up, down, mins) + } +} + +// AccountDayBytes returns the account's total (up+down) bytes for the day, +// the basis for the GB 综合配额 connect gate. +func TestSQLite_AccountDayBytes(t *testing.T) { + ctx := context.Background() + db := openSQLite(t) + ns := nodes.NewSQLNodeStore(db) + seedUser(t, db, 1) + day := time.Date(2026, 6, 24, 0, 0, 0, 0, time.UTC) + other := time.Date(2026, 6, 23, 0, 0, 0, 0, time.UTC) + + // No usage yet → 0. + if b, err := ns.AccountDayBytes(ctx, 1, day); err != nil || b != 0 { + t.Fatalf("empty: bytes=%d err=%v, want 0/nil", b, err) + } + + if err := ns.AccumulateUsage(ctx, 1, day, 300, 700, 5); err != nil { + t.Fatalf("accum: %v", err) + } + if err := ns.AccumulateUsage(ctx, 1, day, 100, 100, 1); err != nil { + t.Fatalf("accum2: %v", err) + } + if b, err := ns.AccountDayBytes(ctx, 1, day); err != nil || b != 1200 { + t.Errorf("day bytes=%d err=%v, want 1200", b, err) + } + // A different date is isolated. + if b, err := ns.AccountDayBytes(ctx, 1, other); err != nil || b != 0 { + t.Errorf("other day bytes=%d, want 0", b) + } +} + +// DeviceUsageRange aggregates usage_device_daily per device over a window, +// joins devices for display metadata, sums across days within the window, +// excludes out-of-window dates and zero-usage devices, and orders busiest-first. +func TestSQLite_DeviceUsageRange(t *testing.T) { + ctx := context.Background() + db := openSQLite(t) + ns := nodes.NewSQLNodeStore(db) + us := usage.NewStore(db) + seedUser(t, db, 1) + seedDevice(t, db, 10, 1, "dev-a") + seedDevice(t, db, 11, 1, "dev-b") + seedDevice(t, db, 12, 1, "dev-quiet") // no usage → omitted + + from := time.Date(2026, 6, 20, 0, 0, 0, 0, time.UTC) + to := time.Date(2026, 6, 24, 0, 0, 0, 0, time.UTC) + inWindow1 := time.Date(2026, 6, 21, 0, 0, 0, 0, time.UTC) + inWindow2 := time.Date(2026, 6, 23, 0, 0, 0, 0, time.UTC) + outWindow := time.Date(2026, 6, 19, 0, 0, 0, 0, time.UTC) + + // dev-a: two in-window days summed + one out-of-window day excluded. + if err := ns.AccumulateDeviceUsage(ctx, 1, 10, inWindow1, 1000, 2000, 5); err != nil { + t.Fatalf("accum a1: %v", err) + } + if err := ns.AccumulateDeviceUsage(ctx, 1, 10, inWindow2, 500, 500, 3); err != nil { + t.Fatalf("accum a2: %v", err) + } + if err := ns.AccumulateDeviceUsage(ctx, 1, 10, outWindow, 9999, 9999, 99); err != nil { + t.Fatalf("accum a-out: %v", err) + } + // dev-b: smaller total (so it sorts after dev-a). + if err := ns.AccumulateDeviceUsage(ctx, 1, 11, inWindow1, 100, 100, 1); err != nil { + t.Fatalf("accum b1: %v", err) + } + + rows, err := us.DeviceUsageRange(ctx, 1, from, to) + if err != nil { + t.Fatalf("DeviceUsageRange: %v", err) + } + if len(rows) != 2 { + t.Fatalf("got %d device rows, want 2 (dev-quiet omitted)", len(rows)) + } + + // Busiest first: dev-a (4000 bytes) before dev-b (200 bytes). + if rows[0].UUID != "dev-a" { + t.Errorf("rows[0].UUID=%q, want dev-a (busiest first)", rows[0].UUID) + } + a := rows[0] + if a.BytesUp != 1500 || a.BytesDown != 2500 || a.MinutesUsed != 8 { + t.Errorf("dev-a sums up=%d down=%d min=%d, want 1500/2500/8 (out-of-window excluded)", + a.BytesUp, a.BytesDown, a.MinutesUsed) + } + if a.Name != "devdev-a" || a.Platform != "ios" { + t.Errorf("dev-a join meta name=%q platform=%q, want devdev-a/ios", a.Name, a.Platform) + } + if rows[1].UUID != "dev-b" || rows[1].BytesUp != 100 || rows[1].BytesDown != 100 { + t.Errorf("dev-b row wrong: %+v", rows[1]) + } + + // A different user sees none of this account's devices. + if _, err := db.Exec( + `INSERT INTO users (id, uuid, email, pw_hash, dp_uuid, status, created_at) + VALUES (2, 'u2-uuid', 'u2@example.com', 'x', 'dp-u2', 'active', ?)`, + time.Now().UTC()); err != nil { + t.Fatalf("seed user2: %v", err) + } + other, err := us.DeviceUsageRange(ctx, 2, from, to) + if err != nil { + t.Fatalf("DeviceUsageRange user2: %v", err) + } + if len(other) != 0 { + t.Errorf("user2 leaked %d rows, want 0", len(other)) + } +} + +// EntitlementForUser loads daily_mb: free fallback = 500; an active pro +// subscription = 102400 (from the 000015 seed). +func TestSQLite_EntitlementForUser_DailyMB(t *testing.T) { + ctx := context.Background() + db := openSQLite(t) + ns := nodes.NewSQLNodeStore(db) + seedUser(t, db, 1) + + // No subscription → free defaults. + ent, err := ns.EntitlementForUser(ctx, 1) + if err != nil || ent == nil { + t.Fatalf("entitlement: %v", err) + } + if !ent.AdGate || !ent.DailyMB.Valid || ent.DailyMB.Int64 != 500 { + t.Errorf("free: adGate=%v dailyMB=%v/%d, want true/valid/500", + ent.AdGate, ent.DailyMB.Valid, ent.DailyMB.Int64) + } + + // Active pro subscription (plan_id=2 from seed) → daily_mb 102400, no ad gate. + if _, err := db.Exec( + `INSERT INTO subscriptions (user_id, plan_id, expires_at, source, created_at) + VALUES (1, 2, ?, 'code', ?)`, + time.Now().UTC().Add(24*time.Hour), time.Now().UTC()); err != nil { + t.Fatalf("seed subscription: %v", err) + } + ent, err = ns.EntitlementForUser(ctx, 1) + if err != nil || ent == nil { + t.Fatalf("entitlement pro: %v", err) + } + if ent.AdGate || !ent.DailyMB.Valid || ent.DailyMB.Int64 != 102400 { + t.Errorf("pro: adGate=%v dailyMB=%v/%d, want false/valid/102400", + ent.AdGate, ent.DailyMB.Valid, ent.DailyMB.Int64) + } +} diff --git a/server/internal/usage/handler.go b/server/internal/usage/handler.go index 96371ec..85e42da 100644 --- a/server/internal/usage/handler.go +++ b/server/internal/usage/handler.go @@ -69,6 +69,56 @@ func (h *UsageHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { _ = json.NewEncoder(w).Encode(usageResponse{Points: points}) } +// DeviceUsageHandler serves GET /v1/usage/devices?days=N — per-device usage +// totals for the authenticated account ("下分设备" breakdown). +type DeviceUsageHandler struct { + svc *Service +} + +// NewDeviceUsageHandler creates a DeviceUsageHandler. +func NewDeviceUsageHandler(svc *Service) *DeviceUsageHandler { return &DeviceUsageHandler{svc: svc} } + +type deviceUsageResponse struct { + Devices []DeviceUsagePoint `json:"devices"` +} + +// ServeHTTP implements http.Handler. +func (h *DeviceUsageHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + userID := userIDFromContext(r) + if userID == 0 { + apierr.WriteJSON(w, http.StatusUnauthorized, &apierr.Error{ + Code: "UNAUTHORIZED", + MessageZH: "请先登录", + MessageEn: "Authentication required", + }) + return + } + + days := 7 + if v := r.URL.Query().Get("days"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n < 1 || n > 90 { + apierr.WriteJSON(w, http.StatusBadRequest, apierr.ErrBadRequest) + return + } + days = n + } + + devices, apiErr := h.svc.DeviceUsage(r.Context(), userID, days) + if apiErr != nil { + apierr.WriteJSON(w, http.StatusInternalServerError, apiErr) + return + } + + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(deviceUsageResponse{Devices: devices}) +} + // AdsUnlockHandler serves POST /v1/ads/unlock. type AdsUnlockHandler struct { svc *Service diff --git a/server/internal/usage/handler_test.go b/server/internal/usage/handler_test.go new file mode 100644 index 0000000..feba3b4 --- /dev/null +++ b/server/internal/usage/handler_test.go @@ -0,0 +1,46 @@ +package usage + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/wangjia/pangolin/server/internal/codes" +) + +// authedRequest builds a request whose context carries an authenticated user id, +// mirroring what auth.RequireAuth installs upstream. +func authedRequest(method, target string, userID int64) *http.Request { + r := httptest.NewRequest(method, target, nil) + if userID != 0 { + r = r.WithContext(context.WithValue(r.Context(), codes.CtxKeyUserID, userID)) + } + return r +} + +// The DeviceUsageHandler gates on auth, method and the days query param before +// ever touching the service/DB — these branches are verified DB-free. +func TestDeviceUsageHandler_Guards(t *testing.T) { + h := NewDeviceUsageHandler(NewService(nil, nil, nil, 0)) + + cases := []struct { + name string + req *http.Request + status int + }{ + {"no auth → 401", authedRequest(http.MethodGet, "/v1/usage/devices", 0), http.StatusUnauthorized}, + {"wrong method → 405", authedRequest(http.MethodPost, "/v1/usage/devices", 7), http.StatusMethodNotAllowed}, + {"days out of range → 400", authedRequest(http.MethodGet, "/v1/usage/devices?days=999", 7), http.StatusBadRequest}, + {"days non-numeric → 400", authedRequest(http.MethodGet, "/v1/usage/devices?days=abc", 7), http.StatusBadRequest}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + w := httptest.NewRecorder() + h.ServeHTTP(w, c.req) + if w.Code != c.status { + t.Errorf("status=%d, want %d (body=%s)", w.Code, c.status, w.Body.String()) + } + }) + } +} diff --git a/server/internal/usage/service.go b/server/internal/usage/service.go index f814aaf..a2eb4e3 100644 --- a/server/internal/usage/service.go +++ b/server/internal/usage/service.go @@ -85,6 +85,49 @@ func (svc *Service) UsageCurve(ctx context.Context, userID int64, days int) ([]U return points, nil } +// DeviceUsagePoint is one device's aggregated usage over the requested window, +// used by the stats page's per-device ("下分设备") breakdown. +type DeviceUsagePoint struct { + UUID string `json:"uuid"` + Name string `json:"name"` + Platform string `json:"platform"` // ios | android | windows | macos + BytesUp uint64 `json:"bytes_up"` + BytesDown uint64 `json:"bytes_down"` + MinutesUsed int `json:"minutes_used"` +} + +// DeviceUsage returns per-device usage totals for userID over the last `days` +// (UTC), busiest device first. days is clamped to [1, 90]. Devices with no +// usage in the window are omitted (the response is never nil — empty slice). +func (svc *Service) DeviceUsage(ctx context.Context, userID int64, days int) ([]DeviceUsagePoint, *apierr.Error) { + if days < 1 { + days = 7 + } + if days > 90 { + days = 90 + } + + today := utcToday() + from := today.AddDate(0, 0, -(days - 1)) + + rows, err := svc.store.DeviceUsageRange(ctx, userID, from, today) + if err != nil { + return nil, apierr.ErrInternal + } + points := make([]DeviceUsagePoint, 0, len(rows)) + for _, r := range rows { + points = append(points, DeviceUsagePoint{ + UUID: r.UUID, + Name: r.Name, + Platform: r.Platform, + BytesUp: r.BytesUp, + BytesDown: r.BytesDown, + MinutesUsed: r.MinutesUsed, + }) + } + return points, nil +} + // TodaySummary is the /v1/me today_usage block. type TodaySummary struct { MinutesUsed int `json:"minutes_used"` diff --git a/server/internal/usage/store.go b/server/internal/usage/store.go index b808ac7..9464f02 100644 --- a/server/internal/usage/store.go +++ b/server/internal/usage/store.go @@ -107,6 +107,48 @@ func (s *Store) GetUsageRange(ctx context.Context, userID int64, from, to time.T return out, rows.Err() } +// DeviceUsageRow is one device's aggregated usage over a date window, joined to +// the devices table for display metadata. It is the per-device ("下分设备") +// counterpart of DailyUsage's account rollup. +type DeviceUsageRow struct { + UUID string + Name string + Platform string + BytesUp uint64 + BytesDown uint64 + MinutesUsed int +} + +// DeviceUsageRange returns each device's summed usage over [from, to] (inclusive +// UTC dates) for userID, one row per device, ordered by total bytes descending. +// Devices with no usage in the window are omitted. All non-aggregate columns +// are in GROUP BY for MySQL ONLY_FULL_GROUP_BY portability (sqlite tolerant). +func (s *Store) DeviceUsageRange(ctx context.Context, userID int64, from, to time.Time) ([]DeviceUsageRow, error) { + rows, err := s.db.QueryContext(ctx, + `SELECT d.uuid, d.name, d.platform, + SUM(ud.bytes_up), SUM(ud.bytes_down), SUM(ud.minutes_used) + FROM usage_device_daily ud + JOIN devices d ON d.id = ud.device_id + WHERE ud.user_id = ? AND ud.date BETWEEN ? AND ? + GROUP BY ud.device_id, d.uuid, d.name, d.platform + ORDER BY (SUM(ud.bytes_up) + SUM(ud.bytes_down)) DESC`, + userID, from.UTC().Format(dateLayout), to.UTC().Format(dateLayout)) + if err != nil { + return nil, fmt.Errorf("store.DeviceUsageRange: %w", err) + } + defer rows.Close() + + var out []DeviceUsageRow + for rows.Next() { + var r DeviceUsageRow + if err := rows.Scan(&r.UUID, &r.Name, &r.Platform, &r.BytesUp, &r.BytesDown, &r.MinutesUsed); err != nil { + return nil, fmt.Errorf("store.DeviceUsageRange scan: %w", err) + } + out = append(out, r) + } + return out, rows.Err() +} + // GetDay returns the usage_daily row for (userID, day), or nil if none exists. func (s *Store) GetDay(ctx context.Context, userID int64, day time.Time) (*DailyUsage, error) { var u DailyUsage diff --git a/server/migrations/mysql/000015_per_device_usage.down.sql b/server/migrations/mysql/000015_per_device_usage.down.sql new file mode 100644 index 0000000..ea988b1 --- /dev/null +++ b/server/migrations/mysql/000015_per_device_usage.down.sql @@ -0,0 +1,4 @@ +ALTER TABLE plans DROP COLUMN daily_mb; +DROP TABLE IF EXISTS usage_device_daily; +DROP INDEX idx_devices_dp_uuid ON devices; +ALTER TABLE devices DROP COLUMN dp_uuid; diff --git a/server/migrations/mysql/000015_per_device_usage.up.sql b/server/migrations/mysql/000015_per_device_usage.up.sql new file mode 100644 index 0000000..aae2fb9 --- /dev/null +++ b/server/migrations/mysql/000015_per_device_usage.up.sql @@ -0,0 +1,26 @@ +-- 每设备归因 + GB 综合配额（todo #5 Phase 2） +-- +-- ① devices.dp_uuid：每设备独立的数据面凭证（替代账户级 users.dp_uuid 作为出网凭证）。 +-- 唯一索引而非内联 UNIQUE，与 SQLite 迁移对齐（NULL 不冲突，存量设备待回填）。 +-- ② usage_device_daily：按设备的每日用量；usage_daily 仍作账户综合 rollup。 +-- ③ plans.daily_mb：每日综合流量配额（MB，与 daily_minutes 并存，按账户综合卡）。 +-- 用 MB 而非 GB 以支持免费版 500MB。free 500 / pro 100GB / team 200GB（可调）。 + +ALTER TABLE devices ADD COLUMN dp_uuid CHAR(36) NULL; +CREATE UNIQUE INDEX idx_devices_dp_uuid ON devices (dp_uuid); + +CREATE TABLE usage_device_daily ( + user_id BIGINT UNSIGNED NOT NULL, + device_id BIGINT UNSIGNED NOT NULL, + date DATE NOT NULL, + bytes_up BIGINT UNSIGNED NOT NULL DEFAULT 0, + bytes_down BIGINT UNSIGNED NOT NULL DEFAULT 0, + minutes_used INT NOT NULL DEFAULT 0, + PRIMARY KEY (device_id, date), + INDEX idx_usage_device_user_date (user_id, date) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + +ALTER TABLE plans ADD COLUMN daily_mb INT NULL; +UPDATE plans SET daily_mb = 500 WHERE code = 'free'; +UPDATE plans SET daily_mb = 102400 WHERE code = 'pro'; +UPDATE plans SET daily_mb = 204800 WHERE code = 'team'; diff --git a/server/migrations/sqlite/000015_per_device_usage.down.sql b/server/migrations/sqlite/000015_per_device_usage.down.sql new file mode 100644 index 0000000..f947571 --- /dev/null +++ b/server/migrations/sqlite/000015_per_device_usage.down.sql @@ -0,0 +1,5 @@ +ALTER TABLE plans DROP COLUMN daily_mb; +DROP INDEX IF EXISTS idx_usage_device_user_date; +DROP TABLE IF EXISTS usage_device_daily; +DROP INDEX IF EXISTS idx_devices_dp_uuid; +ALTER TABLE devices DROP COLUMN dp_uuid; diff --git a/server/migrations/sqlite/000015_per_device_usage.up.sql b/server/migrations/sqlite/000015_per_device_usage.up.sql new file mode 100644 index 0000000..8ed5aaa --- /dev/null +++ b/server/migrations/sqlite/000015_per_device_usage.up.sql @@ -0,0 +1,27 @@ +-- 每设备归因 + GB 综合配额（todo #5 Phase 2） +-- +-- ① devices.dp_uuid：每设备独立的数据面凭证（替代账户级 users.dp_uuid 作为出网凭证）。 +-- SQLite ADD COLUMN 不支持 UNIQUE 约束 → 列加完再建唯一索引（NULL 视为相异，存量设备 +-- NULL 不冲突，待回填）。 +-- ② usage_device_daily：按设备的每日用量；usage_daily 仍作账户综合 rollup。 +-- ③ plans.daily_mb：每日综合流量配额（MB，与 daily_minutes 并存，按账户综合卡）。 +-- 用 MB 而非 GB 以支持免费版 500MB。free 500 / pro 100GB / team 200GB（可调）。 + +ALTER TABLE devices ADD COLUMN dp_uuid TEXT NULL; +CREATE UNIQUE INDEX idx_devices_dp_uuid ON devices (dp_uuid); + +CREATE TABLE usage_device_daily ( + user_id INTEGER NOT NULL, + device_id INTEGER NOT NULL, + date DATE NOT NULL, + bytes_up INTEGER NOT NULL DEFAULT 0, + bytes_down INTEGER NOT NULL DEFAULT 0, + minutes_used INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (device_id, date) +); +CREATE INDEX idx_usage_device_user_date ON usage_device_daily (user_id, date); + +ALTER TABLE plans ADD COLUMN daily_mb INTEGER NULL; +UPDATE plans SET daily_mb = 500 WHERE code = 'free'; +UPDATE plans SET daily_mb = 102400 WHERE code = 'pro'; +UPDATE plans SET daily_mb = 204800 WHERE code = 'team';