merge: maestro/tsk_x7wrlA87orsY [devices + 订阅校验中间件] (tsk_VRzw-af__qWx)

手动合并 tsk_x7wrlA87orsY（设备管理 + 订阅校验中间件）到 main：冲突解决： - server/internal/apierr/apierr.go：保留 tsk_GXDoc3Cs07Rn 版本（New/StatusFor/ Middleware/ErrConflict/改善文档），并入 tsk_x7wrlA87orsY 新增的 ErrAccountBanned 及对应 StatusFor case（→ 403）。新增文件（来自 tsk_x7wrlA87orsY）： - server/internal/devices/doc.go package 文档（替换占位 stub） - server/internal/devices/context.go CtxKeyUserID / Plan / WithPlan / PlanFromCtx - server/internal/devices/handler.go GET /v1/me/devices · DELETE /v1/me/devices/{id} - server/internal/devices/middleware.go SubscriptionMiddleware · CheckDeviceQuota · RequirePaidTier - server/internal/devices/service.go RegisterIfAbsent / DeleteDevice / ResolvePlan + 纯函数 resolveEffectivePlan - server/internal/devices/store.go MySQL 数据访问层 - server/internal/devices/service_test.go 15 个单测（全通过） - server/internal/devices/devices_integration_test.go testcontainers 集成测试 OpenAPI 更新（来自 tsk_x7wrlA87orsY）： - server/api/openapi.yaml：SubscriptionInfo.source 枚举补 free - design/server/openapi.yaml：SubscriptionInfo.source 枚举补 admin, free 测试：go build ./... ✓；go test ./internal/apierr/... ✓（8 tests）； go test ./internal/devices/... ✓（15 tests）。 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-13 15:09:51 +08:00
parent 30e73b31c2
commit d88c1ae647
11 changed files with 1501 additions and 7 deletions
@@ -0,0 +1,385 @@
+package devices
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/wangjia/pangolin/server/internal/apierr"
+)
+
+// CredentialRevoker abstracts the node-side credential recall triggered when a
+// device is removed.
+//
+// The data-plane credential model is per-user (users.dp_uuid, doc/02 §3.2): the
+// node side has no device dimension, so "revoke a device" is realised as
+// "recall/re-issue the owning user's credential". The interface is defined here
+// (consumer side) to avoid an import cycle with the nodes package; module #5's
+// nodes.Hub will satisfy it, with RevokeForUser fronting Hub.Push(RevokeCredential).
+type CredentialRevoker interface {
+	// RevokeForUser instructs node agents to recall (and lazily re-issue) the
+	// data-plane credential for userID. reason is a short machine tag, e.g.
+	// "device_deleted", used for audit/telemetry on the node side.
+	RevokeForUser(ctx context.Context, userID int64, reason string) error
+}
+
+// NoopRevoker is the default CredentialRevoker used until module #5 lands.
+// It records the last call so callers/tests can assert on it.
+type NoopRevoker struct {
+	Calls []RevokeCall
+}
+
+// RevokeCall captures the arguments of a RevokeForUser invocation.
+type RevokeCall struct {
+	UserID int64
+	Reason string
+}
+
+// RevokeForUser implements CredentialRevoker by recording the call.
+func (n *NoopRevoker) RevokeForUser(_ context.Context, userID int64, reason string) error {
+	n.Calls = append(n.Calls, RevokeCall{UserID: userID, Reason: reason})
+	return nil
+}
+
+// Service implements the devices business logic.
+type Service struct {
+	store   *Store
+	revoker CredentialRevoker
+}
+
+// NewService creates a Service. If revoker is nil a NoopRevoker is used so the
+// module can be wired and tested before module #5 (nodes.Hub) is available.
+func NewService(store *Store, revoker CredentialRevoker) *Service {
+	if revoker == nil {
+		revoker = &NoopRevoker{}
+	}
+	return &Service{store: store, revoker: revoker}
+}
+
+// Device is the API representation of a registered device.
+type Device struct {
+	UUID     string  `json:"uuid"`
+	Name     string  `json:"name"`
+	Platform string  `json:"platform"`
+	LastSeen *string `json:"last_seen"` // RFC 3339 UTC; null when never seen
+}
+
+func toAPIDevice(d DeviceRow) Device {
+	out := Device{UUID: d.UUID, Name: d.Name, Platform: d.Platform}
+	if d.LastSeen.Valid {
+		s := d.LastSeen.Time.UTC().Format(time.RFC3339)
+		out.LastSeen = &s
+	}
+	return out
+}
+
+// ListDevices returns the user's devices.
+func (svc *Service) ListDevices(ctx context.Context, userID int64) ([]Device, *apierr.Error) {
+	rows, err := svc.store.ListByUser(ctx, userID)
+	if err != nil {
+		return nil, apierr.ErrInternal
+	}
+	out := make([]Device, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, toAPIDevice(r))
+	}
+	return out, nil
+}
+
+// RegisterInput carries the inputs for implicit device registration, called by
+// nodes.connect when a client presents a device_id.
+type RegisterInput struct {
+	UserID     int64
+	DeviceUUID string
+	Name       string // client-reported, may come from UA; truncated to 64 runes
+	Platform   string // ios | android | windows | macos
+	MaxDevices int    // plan cap, from the resolved Plan (PlanFromCtx)
+}
+
+// RegisterIfAbsent registers a device on first sight and refreshes last_seen on
+// subsequent sights. The device count is checked against MaxDevices before
+// inserting a brand-new device. Per-user serialization is achieved by locking
+// the users row for the duration of the transaction.
+func (svc *Service) RegisterIfAbsent(ctx context.Context, in RegisterInput) (*Device, *apierr.Error) {
+	uuid := strings.TrimSpace(in.DeviceUUID)
+	if uuid == "" {
+		return nil, apierr.ErrBadRequest
+	}
+	platform, ok := normalizePlatform(in.Platform)
+	if !ok {
+		return nil, apierr.ErrBadRequest
+	}
+	name := normalizeName(in.Name, platform)
+
+	tx, err := svc.store.BeginTx(ctx)
+	if err != nil {
+		return nil, apierr.ErrInternal
+	}
+	committed := false
+	defer func() {
+		if !committed {
+			_ = tx.Rollback()
+		}
+	}()
+
+	// Lock the owning user to serialize concurrent registrations.
+	exists, status, err := svc.store.lockUser(ctx, tx, in.UserID)
+	if err != nil {
+		return nil, apierr.ErrInternal
+	}
+	if !exists {
+		return nil, apierr.ErrUnauthorized
+	}
+	if status == "banned" {
+		return nil, apierr.ErrAccountBanned
+	}
+
+	existing, err := svc.store.findDeviceByUUIDTx(ctx, tx, uuid)
+	if err != nil {
+		return nil, apierr.ErrInternal
+	}
+	if existing != nil {
+		if existing.UserID != in.UserID {
+			// UUID is client-generated; a collision across users is treated as
+			// a conflict rather than silently rebinding the device.
+			return nil, apierr.ErrForbidden
+		}
+		if err := svc.store.touchLastSeenTx(ctx, tx, existing.ID); err != nil {
+			return nil, apierr.ErrInternal
+		}
+		if err := tx.Commit(); err != nil {
+			return nil, apierr.ErrInternal
+		}
+		committed = true
+		d := toAPIDevice(*existing)
+		return &d, nil
+	}
+
+	// New device: enforce the plan device cap.
+	count, err := svc.store.countDevicesTx(ctx, tx, in.UserID)
+	if err != nil {
+		return nil, apierr.ErrInternal
+	}
+	if in.MaxDevices > 0 && count >= in.MaxDevices {
+		return nil, errDeviceLimit(in.MaxDevices)
+	}
+
+	row, err := svc.store.insertDeviceTx(ctx, tx, uuid, in.UserID, name, platform)
+	if err != nil {
+		return nil, apierr.ErrInternal
+	}
+	if err := svc.store.writeAuditLogTx(ctx, tx,
+		fmt.Sprintf("user:%d", in.UserID), "device.register", "device:"+uuid,
+		deviceAuditMeta(name, platform)); err != nil {
+		_ = err // audit failure must not abort the business transaction
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, apierr.ErrInternal
+	}
+	committed = true
+
+	d := toAPIDevice(*row)
+	return &d, nil
+}
+
+// DeleteDevice hard-deletes a device (transactionally, with an audit_log entry)
+// and then triggers per-user credential recall on the node side.
+//
+//   - device not found            → 404 NOT_FOUND
+//   - device owned by another user → 403 FORBIDDEN (does not delete)
+func (svc *Service) DeleteDevice(ctx context.Context, userID int64, deviceUUID string) *apierr.Error {
+	uuid := strings.TrimSpace(deviceUUID)
+	if uuid == "" {
+		return apierr.ErrBadRequest
+	}
+
+	tx, err := svc.store.BeginTx(ctx)
+	if err != nil {
+		return apierr.ErrInternal
+	}
+	committed := false
+	defer func() {
+		if !committed {
+			_ = tx.Rollback()
+		}
+	}()
+
+	dev, err := svc.store.findDeviceByUUIDTx(ctx, tx, uuid)
+	if err != nil {
+		return apierr.ErrInternal
+	}
+	if dev == nil {
+		return apierr.ErrNotFound
+	}
+	if dev.UserID != userID {
+		return apierr.ErrForbidden
+	}
+
+	if err := svc.store.deleteDeviceTx(ctx, tx, dev.ID); err != nil {
+		return apierr.ErrInternal
+	}
+	if err := svc.store.writeAuditLogTx(ctx, tx,
+		fmt.Sprintf("user:%d", userID), "device.delete", "device:"+uuid,
+		deviceAuditMeta(dev.Name, dev.Platform)); err != nil {
+		_ = err // audit failure must not abort the delete
+	}
+	if err := tx.Commit(); err != nil {
+		return apierr.ErrInternal
+	}
+	committed = true
+
+	// Recall the user's data-plane credential. Per the dp_uuid model this is a
+	// per-user operation (no device dimension on the node side). Best-effort:
+	// the device is already removed; a transient revoke failure is reconciled
+	// by the node agent's periodic resync, so we do not fail the request.
+	if err := svc.revoker.RevokeForUser(ctx, userID, "device_deleted"); err != nil {
+		_ = err
+	}
+	return nil
+}
+
+// SubscriptionInfo is the /v1/me subscription summary provided by this module.
+type SubscriptionInfo struct {
+	PlanCode  string  `json:"plan_code"`
+	ExpiresAt *string `json:"expires_at"` // RFC 3339 UTC; null for free fallback
+	Source    string  `json:"source"`
+}
+
+// SubscriptionSummary resolves the user's effective plan and returns the /v1/me
+// subscription summary.
+func (svc *Service) SubscriptionSummary(ctx context.Context, userID int64) (*SubscriptionInfo, *apierr.Error) {
+	p, apiErr := svc.ResolvePlan(ctx, userID)
+	if apiErr != nil {
+		return nil, apiErr
+	}
+	info := &SubscriptionInfo{PlanCode: p.PlanCode, Source: p.Source}
+	if p.ExpiresAt != nil {
+		s := p.ExpiresAt.UTC().Format(time.RFC3339)
+		info.ExpiresAt = &s
+	}
+	return info, nil
+}
+
+// ResolvePlan loads the user's status + subscriptions and resolves the effective
+// plan. It is the shared entry point used by SubscriptionMiddleware and
+// SubscriptionSummary.
+func (svc *Service) ResolvePlan(ctx context.Context, userID int64) (Plan, *apierr.Error) {
+	status, exists, err := svc.store.GetUserStatus(ctx, userID)
+	if err != nil {
+		return Plan{}, apierr.ErrInternal
+	}
+	if !exists {
+		return Plan{}, apierr.ErrUnauthorized
+	}
+	subs, err := svc.store.GetSubscriptions(ctx, userID)
+	if err != nil {
+		return Plan{}, apierr.ErrInternal
+	}
+	free, err := svc.store.GetFreePlan(ctx)
+	if err != nil {
+		return Plan{}, apierr.ErrInternal
+	}
+	return resolveEffectivePlan(time.Now().UTC(), status, subs, free)
+}
+
+// resolveEffectivePlan is the pure plan-resolution rule (no I/O), unit-tested
+// directly:
+//
+//   - banned account                         → ErrAccountBanned
+//   - among subscriptions with expires_at>now, pick the highest tier;
+//     ties broken by the latest expiry
+//   - no active subscription                 → free fallback (ExpiresAt nil)
+//
+// now must be in UTC. The boundary is strict: expires_at == now counts as expired.
+func resolveEffectivePlan(now time.Time, status string, subs []effSub, free Plan) (Plan, *apierr.Error) {
+	if status == "banned" {
+		return Plan{}, apierr.ErrAccountBanned
+	}
+
+	var best *effSub
+	for i := range subs {
+		s := &subs[i]
+		if !s.ExpiresAt.After(now) {
+			continue // expired (strict UTC boundary)
+		}
+		if best == nil {
+			best = s
+			continue
+		}
+		bt, st := planTier(best.PlanCode), planTier(s.PlanCode)
+		if st > bt || (st == bt && s.ExpiresAt.After(best.ExpiresAt)) {
+			best = s
+		}
+	}
+
+	if best == nil {
+		return free, nil
+	}
+
+	exp := best.ExpiresAt.UTC()
+	p := Plan{
+		PlanCode:   best.PlanCode,
+		ExpiresAt:  &exp,
+		MaxDevices: best.MaxDevices,
+		AdGate:     best.AdGate,
+		Source:     best.Source,
+	}
+	if best.DailyMinutes.Valid {
+		m := int(best.DailyMinutes.Int64)
+		p.DailyMinutes = &m
+	}
+	return p, nil
+}
+
+// --------------------------------------------------------------------------
+// helpers
+// --------------------------------------------------------------------------
+
+// normalizePlatform validates/canonicalizes a platform string against the
+// devices.platform ENUM (ios/android/windows/macos). Note: the OpenAPI Device
+// schema also lists "linux", but the DB ENUM (migration #1) does not include
+// it, so "linux" is rejected until the schema is widened.
+func normalizePlatform(s string) (string, bool) {
+	switch strings.ToLower(strings.TrimSpace(s)) {
+	case "ios":
+		return "ios", true
+	case "android":
+		return "android", true
+	case "windows":
+		return "windows", true
+	case "macos":
+		return "macos", true
+	}
+	return "", false
+}
+
+// normalizeName trims the client-reported name, falls back to the platform when
+// empty, and truncates to the devices.name VARCHAR(64) limit (by rune).
+func normalizeName(name, platform string) string {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		name = platform
+	}
+	r := []rune(name)
+	if len(r) > 64 {
+		name = string(r[:64])
+	}
+	return name
+}
+
+// deviceAuditMeta builds a compact JSON meta blob for audit_log.
+func deviceAuditMeta(name, platform string) string {
+	b, _ := json.Marshal(map[string]string{"name": name, "platform": platform})
+	return string(b)
+}
+
+// errDeviceLimit builds a bilingual device-cap error that embeds the limit.
+func errDeviceLimit(max int) *apierr.Error {
+	return &apierr.Error{
+		Code:      "DEVICE_LIMIT_EXCEEDED",
+		MessageZH: fmt.Sprintf("设备数量已达上限（%d 台），请先在其他设备上退出后重试", max),
+		MessageEn: fmt.Sprintf("Device limit reached (%d devices). Please remove another device and try again.", max),
+	}
+}