package devices import ( "context" "encoding/json" "fmt" "strings" "time" "github.com/wangjia/pangolin/server/internal/apierr" ) // CredentialRevoker recalls a device's data-plane credential (its dp_uuid) on the // node side when the device is cleared. Per migration 000015 the credential is // per-device, so revocation targets the dp_uuid. Defined consumer-side to avoid an // import cycle; nodes.Service satisfies it (push Revoke to the holding node(s) + // drop the connect_credentials row). Best-effort: node resync reconciles misses. type CredentialRevoker interface { RevokeDevice(ctx context.Context, dpUUID string) error } // NoopRevoker is the default CredentialRevoker (tests / before nodes wiring). // It records the dp_uuids passed so callers/tests can assert on it. type NoopRevoker struct { Calls []string } // RevokeDevice implements CredentialRevoker by recording the dp_uuid. func (n *NoopRevoker) RevokeDevice(_ context.Context, dpUUID string) error { n.Calls = append(n.Calls, dpUUID) return nil } // SessionPort provides per-device session reads/revocations. Satisfied by // sessions.Store; injected to keep packages decoupled. type SessionPort interface { LastLoginByDevice(ctx context.Context, userID int64) (map[int64]time.Time, error) RevokeByDevice(ctx context.Context, userID, deviceID int64) ([]string, error) HasActiveSession(ctx context.Context, userID, deviceID int64) (bool, error) ActiveSessionDeviceIDs(ctx context.Context, userID int64) (map[int64]bool, error) } // JTIRevoker drops a refresh JTI from the Redis whitelist. Satisfied by // auth.TokenManager (so a force-logged-out device's tokens stop refreshing). type JTIRevoker interface { Revoke(ctx context.Context, jti string) error } // Service implements the devices business logic. type Service struct { store *Store revoker CredentialRevoker sessions SessionPort // nil until wired (P2/P3) jtiRevoker JTIRevoker // nil until wired (P3) } // NewService creates a Service. If revoker is nil a NoopRevoker is used so the // module can be wired and tested before nodes is available. func NewService(store *Store, revoker CredentialRevoker) *Service { if revoker == nil { revoker = &NoopRevoker{} } return &Service{store: store, revoker: revoker} } // SetSessionPort / SetJTIRevoker / SetCredentialRevoker wire collaborators after // construction (main keeps devices, sessions, auth and nodes decoupled). func (svc *Service) SetSessionPort(s SessionPort) { svc.sessions = s } func (svc *Service) SetJTIRevoker(r JTIRevoker) { svc.jtiRevoker = r } func (svc *Service) SetCredentialRevoker(r CredentialRevoker) { if r != nil { svc.revoker = r } } // 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 ClientVersion string `json:"client_version,omitempty"` // 000016 Online bool `json:"online"` // last_seen within onlineWindow (data-plane active) LastLogin *string `json:"last_login"` // most recent session created_at; null when none } // onlineWindow: a device is "online" if its last_seen is within this window. // last_seen is bumped by connect、用量上报,以及 ~15s 的会话轮询(SessionActive)—— // 后者让「在线」准确反映 app 正在运行。窗口取 90s(覆盖几次丢失的 15s 轮询), // app 一关 ~90s 内转离线。 const onlineWindow = 90 * time.Second // staleWindow: devices not seen within this window are "stale" — excluded from the // plan-cap count and best-effort pruned on login. Covers free-plan reinstall churn // (each reinstall churns the device UUID, leaving zombie rows) without locking users out. const staleWindow = 30 * 24 * time.Hour // DeviceLimitStatus reports the device-cap state. Over is true when the account's // active device count exceeds the plan cap; Devices lists the active set so the // client can present "remove a device" UX. type DeviceLimitStatus struct { Over bool `json:"over"` MaxDevices int `json:"max_devices"` Devices []Device `json:"devices,omitempty"` } 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 out.Online = time.Since(d.LastSeen.Time) < onlineWindow } if d.ClientVersion.Valid { out.ClientVersion = d.ClientVersion.String } return out } // ListDevices returns the user's devices with online status (from last_seen) and // last-login time (most recent session, when a LastLoginSource is wired). 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 } var lastLogin map[int64]time.Time var activeSet map[int64]bool if svc.sessions != nil { if m, e := svc.sessions.LastLoginByDevice(ctx, userID); e == nil { lastLogin = m } // 有活跃会话集:online 须同时满足「last_seen 新」+「会话未撤销」, // 被强退的设备没有活跃会话 → 立即离线(不必等 last_seen 窗口过期)。 if m, e := svc.sessions.ActiveSessionDeviceIDs(ctx, userID); e == nil { activeSet = m } } out := make([]Device, 0, len(rows)) for _, r := range rows { d := toAPIDevice(r) if activeSet != nil && !activeSet[r.ID] { d.Online = false // 无未撤销会话 → 强制离线(activeSet 为 nil 时不干预,fail-open) } if t, ok := lastLogin[r.ID]; ok { s := t.UTC().Format(time.RFC3339) d.LastLogin = &s } out = append(out, d) } 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 | linux ClientVersion string // app version, stored/refreshed on devices.client_version MaxDevices int // plan cap; 0 = no cap } // 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. // Returns the internal device id (for session binding), the API device, or an error. func (svc *Service) RegisterIfAbsent(ctx context.Context, in RegisterInput) (int64, *Device, *apierr.Error) { uuid := strings.TrimSpace(in.DeviceUUID) if uuid == "" { return 0, nil, apierr.ErrBadRequest } platform, ok := normalizePlatform(in.Platform) if !ok { return 0, nil, apierr.ErrBadRequest } name := normalizeName(in.Name, platform) tx, err := svc.store.BeginTx(ctx) if err != nil { return 0, 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 0, nil, apierr.ErrInternal } if !exists { return 0, nil, apierr.ErrUnauthorized } if status == "banned" { return 0, nil, apierr.ErrAccountBanned } // 按 (user,uuid) 查:唯一键是 UNIQUE(user_id,uuid)(migration 21),同一物理设备 // 在别的账号名下的行与本次注册无关 —— 同机换账号各自成行,不再互相 403(F3)。 existing, err := svc.store.findDeviceByUserUUIDTx(ctx, tx, in.UserID, uuid) if err != nil { return 0, nil, apierr.ErrInternal } if existing != nil { if err := svc.store.touchLastSeenTx(ctx, tx, existing.ID, in.ClientVersion); err != nil { return 0, nil, apierr.ErrInternal } if err := tx.Commit(); err != nil { return 0, nil, apierr.ErrInternal } committed = true d := toAPIDevice(*existing) return existing.ID, &d, nil } // New device: enforce the plan device cap. count, err := svc.store.countDevicesTx(ctx, tx, in.UserID) if err != nil { return 0, nil, apierr.ErrInternal } if in.MaxDevices > 0 && count >= in.MaxDevices { return 0, nil, errDeviceLimit(in.MaxDevices) } row, err := svc.store.insertDeviceTx(ctx, tx, uuid, in.UserID, name, platform, in.ClientVersion) if err != nil { return 0, 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 0, nil, apierr.ErrInternal } committed = true d := toAPIDevice(*row) return row.ID, &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 → 404 NOT_FOUND (user-scoped lookup; invisible) func (svc *Service) DeleteDevice(ctx context.Context, userID int64, deviceUUID string) *apierr.Error { uuid := strings.TrimSpace(deviceUUID) if uuid == "" { return apierr.ErrBadRequest } // Resolve (non-tx) so sessions can be revoked BEFORE the delete tx: SQLite // (_txlock=immediate) holds a write lock for the tx, so a session write on // another pool connection would deadlock against it. Lookup is user-scoped // (UNIQUE(user_id,uuid)) — other users' rows with the same uuid are invisible, // so "not mine" and "not found" are both 404. dev, err := svc.store.FindByUserUUID(ctx, userID, uuid) if err != nil { return apierr.ErrInternal } if dev == nil { return apierr.ErrNotFound } // Revoke the device's sessions (drop their refresh JTIs from Redis) while the // rows still exist; the device delete then cascades them away. svc.revokeDeviceSessions(ctx, userID, dev.ID) tx, err := svc.store.BeginTx(ctx) if err != nil { return apierr.ErrInternal } committed := false defer func() { if !committed { _ = tx.Rollback() } }() 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 device's data-plane credential so a live connection is dropped // too. Best-effort: the device row is already gone; a transient failure is // reconciled by node resync, so we never fail the request. (Sessions were // already revoked above, before the cascade removed their rows.) if dev.DpUUID.Valid && dev.DpUUID.String != "" { if err := svc.revoker.RevokeDevice(ctx, dev.DpUUID.String); err != nil { _ = err } } return nil } // ForceLogout revokes a device's login sessions (kicks it offline; the device // stays in the list). The device's data-plane credential is left intact so the // user can simply log in again. // // - device not found → 404 NOT_FOUND // - device owned by another user → 404 NOT_FOUND (user-scoped lookup; invisible) func (svc *Service) ForceLogout(ctx context.Context, userID int64, deviceUUID string) *apierr.Error { uuid := strings.TrimSpace(deviceUUID) if uuid == "" { return apierr.ErrBadRequest } dev, err := svc.store.FindByUserUUID(ctx, userID, uuid) if err != nil { return apierr.ErrInternal } if dev == nil { return apierr.ErrNotFound } svc.revokeDeviceSessions(ctx, userID, dev.ID) return nil } // RenameDevice sets a device's display name (ownership-checked). Empty name → // 400; name is trimmed + truncated to 64 runes. // // - device not found → 404 // - device owned by another user → 404 (user-scoped lookup; invisible) func (svc *Service) RenameDevice(ctx context.Context, userID int64, deviceUUID, rawName string) *apierr.Error { uuid := strings.TrimSpace(deviceUUID) name := strings.TrimSpace(rawName) if uuid == "" || name == "" { return apierr.ErrBadRequest } if r := []rune(name); len(r) > 64 { name = string(r[:64]) } dev, err := svc.store.FindByUserUUID(ctx, userID, uuid) if err != nil { return apierr.ErrInternal } if dev == nil { return apierr.ErrNotFound } if err := svc.store.UpdateName(ctx, dev.ID, name); err != nil { return apierr.ErrInternal } return nil } // SessionActive reports whether the device (by UUID) still has a live login session // for the user — false after a force-logout elsewhere. The client polls this (~15s) // and logs out on false → near-instant remote kick. Fail-open (return true) on a // missing session port / blank or unknown / not-own device, so transient ambiguity // never spuriously logs a user out. func (svc *Service) SessionActive(ctx context.Context, userID int64, deviceUUID string) (bool, *apierr.Error) { if svc.sessions == nil || strings.TrimSpace(deviceUUID) == "" { return true, nil } dev, err := svc.store.FindByUserUUID(ctx, userID, deviceUUID) if err != nil { return false, apierr.ErrInternal } if dev == nil { // 本用户名下无此设备行 = 本设备被「移除」(DeleteDevice 删行)。已登录的客户端在 // 登录时必然注册过自己的设备,轮询自身 device_id 却查无此行,只能是被移除 → 视为 // 会话失效,让其登出(否则被移除的设备永远收到 active=true,不退出,只表现为数据面 // 被断→「节点异常」)。查找按 (user,uuid) 作用域,他人账号下的同 uuid 行不可见, // 不存在旧「非本人设备 fail-safe」分支。 return false, nil } active, err := svc.sessions.HasActiveSession(ctx, userID, dev.ID) if err != nil { return false, apierr.ErrInternal } // 仅活跃会话才刷 last_seen(=app 在运行且未被踢)。被强退设备的轮询不刷, // 否则那一刷会把它顶成「在线」直到窗口过期(本问题的根因)。 if active { _ = svc.store.TouchLastSeen(ctx, dev.ID) } return active, nil } // revokeDeviceSessions marks all of a device's sessions revoked and drops their // refresh JTIs from the Redis whitelist. Best-effort. func (svc *Service) revokeDeviceSessions(ctx context.Context, userID, deviceID int64) { if svc.sessions == nil { return } jtis, err := svc.sessions.RevokeByDevice(ctx, userID, deviceID) if err != nil { return } if svc.jtiRevoker != nil { for _, jti := range jtis { _ = svc.jtiRevoker.Revoke(ctx, jti) } } } // 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) } // CheckDeviceLimit prunes stale devices (best-effort churn cleanup), resolves the // plan cap, and reports whether the account's active device count exceeds it. Used // by the login flow (non-hard-reject gate) and the connect backstop. When over, the // active device list is included so the client can present "remove a device" UX. // The device is already registered by the time this runs, so "over" means // activeCount > cap. MaxDevices==0 means uncapped (never over). func (svc *Service) CheckDeviceLimit(ctx context.Context, userID int64) (*DeviceLimitStatus, *apierr.Error) { cutoff := time.Now().UTC().Add(-staleWindow) if _, err := svc.store.PruneStaleDevices(ctx, userID, cutoff); err != nil { _ = err // prune failure must never block login/connect } plan, apiErr := svc.ResolvePlan(ctx, userID) if apiErr != nil { return nil, apiErr } active, err := svc.store.CountActiveDevices(ctx, userID, cutoff) if err != nil { return nil, apierr.ErrInternal } st := &DeviceLimitStatus{MaxDevices: plan.MaxDevices} st.Over = plan.MaxDevices > 0 && active > plan.MaxDevices if st.Over { devs, apiErr := svc.ListDevices(ctx, userID) if apiErr != nil { return nil, apiErr } st.Devices = devs } return st, nil } // 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 case "linux": return "linux", 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), } }