feat(server/devices): 设备管理 + 订阅校验中间件 (tsk_x7wrlA87orsY)

实现 server/internal/devices 模块： - handler.go: GET /v1/me/devices 列表、DELETE /v1/me/devices/{id} 移除（chi 路由，挂在 /v1/me 下；JWT 中间件之后）。 - service.go: ListDevices / RegisterIfAbsent（隐式登记，按 plan.max_devices 校验，超限返回双语 DEVICE_LIMIT_EXCEEDED 含上限数字）/ DeleteDevice（事务硬删 + audit_log → 调 CredentialRevoker.RevokeForUser 按用户回收凭证，dp_uuid 模型） / SubscriptionSummary / ResolvePlan / 纯函数 resolveEffectivePlan。 - middleware.go: 订阅校验中间件，解析最高档未过期订阅注入 context； helpers PlanFromCtx / CheckDeviceQuota / RequirePaidTier；预留 60s Redis 缓存开关。 - store.go: devices/users/subscriptions/plans/audit_log 数据访问，按用户行锁串行化登记。 - context.go: user_id / plan 的 context key 与 helper。 - CredentialRevoker 接口（消费侧定义，避免与 nodes 循环依赖）+ NoopRevoker，形状对齐 #5 的 Hub.Push(RevokeCredential)，#5 落地前注入 no-op。测试：service_test.go 15 个单测（trial→pro、过期回落 free、banned 拒绝、最高档/同档最晚到期、UTC 严格边界、设备配额、双语限额、平台/名称归一化）全过； devices_integration_test.go（testcontainers，build tag integration）覆盖注册→connect 隐式登记→list→delete 全链路 + 回收断言 + 403/404。 apierr 增加 Unauthorized/Forbidden/NotFound/AccountBanned； OpenAPI SubscriptionInfo.source 枚举补 free。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-13 12:13:09 +08:00
parent 787151245e
commit 87ccaecddf
11 changed files with 1516 additions and 7 deletions
@@ -0,0 +1,221 @@
+package devices
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+)
+
+// DeviceRow mirrors a `devices` table row.
+type DeviceRow struct {
+	ID        int64
+	UUID      string
+	UserID    int64
+	Name      string
+	Platform  string
+	LastSeen  sql.NullTime
+	CreatedAt time.Time
+}
+
+// effSub is an active-or-expired subscription joined with its plan, used by the
+// pure plan resolver. expiry filtering is performed in Go (resolveEffectivePlan)
+// so the UTC boundary logic is unit-testable without a database.
+type effSub struct {
+	PlanCode     string
+	MaxDevices   int
+	DailyMinutes sql.NullInt64
+	AdGate       bool
+	ExpiresAt    time.Time
+	Source       string
+}
+
+// Store wraps a *sql.DB and exposes the database operations the devices module
+// needs. Methods that take a *sql.Tx run within that transaction.
+type Store struct {
+	db *sql.DB
+}
+
+// NewStore creates a Store backed by the given MySQL connection pool.
+func NewStore(db *sql.DB) *Store { return &Store{db: db} }
+
+// BeginTx starts a transaction at Read Committed isolation. Per-user
+// serialization for device mutations is achieved by locking the users row
+// (SELECT ... FOR UPDATE) inside the transaction.
+func (s *Store) BeginTx(ctx context.Context) (*sql.Tx, error) {
+	return s.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
+}
+
+// --------------------------------------------------------------------------
+// Device queries
+// --------------------------------------------------------------------------
+
+// ListByUser returns all devices for userID ordered by creation time.
+func (s *Store) ListByUser(ctx context.Context, userID int64) ([]DeviceRow, error) {
+	rows, err := s.db.QueryContext(ctx,
+		`SELECT id, uuid, user_id, name, platform, last_seen, created_at
+		 FROM devices WHERE user_id=? ORDER BY created_at ASC`,
+		userID)
+	if err != nil {
+		return nil, fmt.Errorf("store.ListByUser: %w", err)
+	}
+	defer rows.Close()
+
+	var out []DeviceRow
+	for rows.Next() {
+		var d DeviceRow
+		if err := rows.Scan(&d.ID, &d.UUID, &d.UserID, &d.Name, &d.Platform, &d.LastSeen, &d.CreatedAt); err != nil {
+			return nil, fmt.Errorf("store.ListByUser scan: %w", err)
+		}
+		out = append(out, d)
+	}
+	return out, rows.Err()
+}
+
+// findDeviceByUUIDTx looks up a device by UUID with FOR UPDATE inside tx.
+// Returns (nil, nil) when the device does not exist.
+func (s *Store) findDeviceByUUIDTx(ctx context.Context, tx *sql.Tx, uuid string) (*DeviceRow, error) {
+	row := tx.QueryRowContext(ctx,
+		`SELECT id, uuid, user_id, name, platform, last_seen, created_at
+		 FROM devices WHERE uuid=? FOR UPDATE`, uuid)
+	var d DeviceRow
+	if err := row.Scan(&d.ID, &d.UUID, &d.UserID, &d.Name, &d.Platform, &d.LastSeen, &d.CreatedAt); err == sql.ErrNoRows {
+		return nil, nil
+	} else if err != nil {
+		return nil, fmt.Errorf("store.findDeviceByUUIDTx: %w", err)
+	}
+	return &d, nil
+}
+
+// lockUser locks the users row to serialize per-user device mutations and
+// returns the user's status. Returns (false, "", nil) when the user is absent.
+func (s *Store) lockUser(ctx context.Context, tx *sql.Tx, userID int64) (exists bool, status string, err error) {
+	row := tx.QueryRowContext(ctx, `SELECT status FROM users WHERE id=? FOR UPDATE`, userID)
+	if e := row.Scan(&status); e == sql.ErrNoRows {
+		return false, "", nil
+	} else if e != nil {
+		return false, "", fmt.Errorf("store.lockUser: %w", e)
+	}
+	return true, status, nil
+}
+
+// countDevicesTx counts a user's devices inside tx.
+func (s *Store) countDevicesTx(ctx context.Context, tx *sql.Tx, userID int64) (int, error) {
+	var n int
+	if err := tx.QueryRowContext(ctx, `SELECT COUNT(1) FROM devices WHERE user_id=?`, userID).Scan(&n); err != nil {
+		return 0, fmt.Errorf("store.countDevicesTx: %w", err)
+	}
+	return n, nil
+}
+
+// insertDeviceTx inserts a new device row inside tx and returns it.
+func (s *Store) insertDeviceTx(ctx context.Context, tx *sql.Tx, uuid string, userID int64, name, platform string) (*DeviceRow, error) {
+	res, err := tx.ExecContext(ctx,
+		`INSERT INTO devices (uuid, user_id, name, platform, last_seen, created_at)
+		 VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6), UTC_TIMESTAMP(6))`,
+		uuid, userID, name, platform)
+	if err != nil {
+		return nil, fmt.Errorf("store.insertDeviceTx: %w", err)
+	}
+	id, _ := res.LastInsertId()
+	return &DeviceRow{ID: id, UUID: uuid, UserID: userID, Name: name, Platform: platform}, nil
+}
+
+// touchLastSeenTx updates a device's last_seen to now inside tx.
+func (s *Store) touchLastSeenTx(ctx context.Context, tx *sql.Tx, deviceID int64) error {
+	_, err := tx.ExecContext(ctx,
+		`UPDATE devices SET last_seen=UTC_TIMESTAMP(6) WHERE id=?`, deviceID)
+	if err != nil {
+		return fmt.Errorf("store.touchLastSeenTx: %w", err)
+	}
+	return nil
+}
+
+// deleteDeviceTx hard-deletes a device row inside tx.
+func (s *Store) deleteDeviceTx(ctx context.Context, tx *sql.Tx, deviceID int64) error {
+	if _, err := tx.ExecContext(ctx, `DELETE FROM devices WHERE id=?`, deviceID); err != nil {
+		return fmt.Errorf("store.deleteDeviceTx: %w", err)
+	}
+	return nil
+}
+
+// --------------------------------------------------------------------------
+// Subscription / plan queries (for the subscription middleware & /me summary)
+// --------------------------------------------------------------------------
+
+// GetUserStatus returns a user's account status ("active"/"banned").
+// exists is false when no such user row is present.
+func (s *Store) GetUserStatus(ctx context.Context, userID int64) (status string, exists bool, err error) {
+	row := s.db.QueryRowContext(ctx, `SELECT status FROM users WHERE id=?`, userID)
+	if e := row.Scan(&status); e == sql.ErrNoRows {
+		return "", false, nil
+	} else if e != nil {
+		return "", false, fmt.Errorf("store.GetUserStatus: %w", e)
+	}
+	return status, true, nil
+}
+
+// GetSubscriptions returns all subscriptions for userID joined with their plan.
+// Expiry filtering is intentionally left to resolveEffectivePlan.
+func (s *Store) GetSubscriptions(ctx context.Context, userID int64) ([]effSub, error) {
+	rows, err := s.db.QueryContext(ctx,
+		`SELECT p.code, p.max_devices, p.daily_minutes, p.ad_gate, s.expires_at, s.source
+		 FROM subscriptions s JOIN plans p ON p.id=s.plan_id
+		 WHERE s.user_id=?`, userID)
+	if err != nil {
+		return nil, fmt.Errorf("store.GetSubscriptions: %w", err)
+	}
+	defer rows.Close()
+
+	var out []effSub
+	for rows.Next() {
+		var e effSub
+		if err := rows.Scan(&e.PlanCode, &e.MaxDevices, &e.DailyMinutes, &e.AdGate, &e.ExpiresAt, &e.Source); err != nil {
+			return nil, fmt.Errorf("store.GetSubscriptions scan: %w", err)
+		}
+		out = append(out, e)
+	}
+	return out, rows.Err()
+}
+
+// GetFreePlan loads the free plan row (the no-subscription fallback).
+func (s *Store) GetFreePlan(ctx context.Context) (Plan, error) {
+	row := s.db.QueryRowContext(ctx,
+		`SELECT max_devices, daily_minutes, ad_gate FROM plans WHERE code='free'`)
+	var maxDevices int
+	var dailyMinutes sql.NullInt64
+	var adGate bool
+	if err := row.Scan(&maxDevices, &dailyMinutes, &adGate); err != nil {
+		return Plan{}, fmt.Errorf("store.GetFreePlan: %w", err)
+	}
+	p := Plan{
+		PlanCode:   "free",
+		MaxDevices: maxDevices,
+		AdGate:     adGate,
+		Source:     "free",
+	}
+	if dailyMinutes.Valid {
+		m := int(dailyMinutes.Int64)
+		p.DailyMinutes = &m
+	}
+	return p, nil
+}
+
+// --------------------------------------------------------------------------
+// Audit log
+// --------------------------------------------------------------------------
+
+// writeAuditLogTx inserts an audit_log row inside tx.
+func (s *Store) writeAuditLogTx(ctx context.Context, tx *sql.Tx, actor, action, target, metaJSON string) error {
+	if metaJSON == "" {
+		metaJSON = "null"
+	}
+	_, err := tx.ExecContext(ctx,
+		`INSERT INTO audit_log (actor, action, target, meta, at)
+		 VALUES (?, ?, ?, ?, UTC_TIMESTAMP(6))`,
+		actor, action, target, metaJSON)
+	if err != nil {
+		return fmt.Errorf("store.writeAuditLogTx: %w", err)
+	}
+	return nil
+}