Token 管理最佳实践

你将构建什么

一个生产级的 Token 管理层——覆盖持久化存储选型、多用户隔离、并发刷新竞争、过期降级、主动撤销和安全存储，确保你的 SaaS 应用在 Token 轮换和服务重启后仍然正常工作。

Token 生命周期

Token	有效期	轮换规则
Access Token	1 小时（JWT）	每次刷新签发新的
Refresh Token	90 天	每次使用即失效，同时签发新 Refresh Token

核心原则

违反以下任何一条，都会导致用户静默掉线——没有错误提示，只有突然的「请重新登录」。

1. 刷新后必须持久化 — onTokenRefresh 回调是你保存新 Token 的唯一时机。错过它，旧 Refresh Token 已失效，新 Token 丢失，用户必须重新授权。 2. 禁止纯内存存储 — 服务重启 = 所有用户 Token 丢失 = 全体重新登录。Token 必须落盘（DB / Redis / 文件）。 3. 轮换必须原子化 — 每次 refreshToken() 调用后，旧的 Access Token 和 Refresh Token 同时失效。保存新 Token 和废弃旧 Token 必须在同一事务中完成。

Step 1: 选择存储策略

方案	适用场景	优势	劣势
DrizzleTokenStore (TS) / SQLAlchemyTokenStore (Python)	已有关系型 DB 的项目	零配置、事务安全、自带 schema	需要 DB 连接
Redis	高并发 SaaS、Session 管理	高性能、天然 TTL、原子操作	需要 Redis 实例、持久化配置
自定义 DB（Prisma / Mongoose / etc.）	已有非 Drizzle ORM	完全掌控 schema	需要自行实现 CRUD
加密文件	CLI 工具、单用户脚本	无外部依赖	不适合多用户、无并发控制

80% 的场景直接用 Contrib 的 DrizzleTokenStore 或 SQLAlchemyTokenStore 即可。只有特殊需求（Redis Session、自定义加密）才需要自行实现。

使用 Contrib 存储

import { ProfyApp } from "@profy-ai/sdk";
import { DrizzleTokenStore } from "@profy-ai/sdk/contrib/drizzle";
import { drizzle } from "drizzle-orm/postgres-js";
import postgres from "postgres";

const db = drizzle(postgres(process.env.DATABASE_URL!));
const tokenStore = new DrizzleTokenStore(db);

const profy = new ProfyApp({
  clientId: process.env.PROFY_APP_ID!,
  clientSecret: process.env.PROFY_APP_SECRET!,
  onTokenRefresh: async (userId, newToken) => {
    await tokenStore.save(userId, newToken);
  },
});

from profy_sdk import ProfyApp
from profy_sdk.contrib.sqlalchemy import SQLAlchemyTokenStore
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(os.environ["DATABASE_URL"])
token_store = SQLAlchemyTokenStore(engine)

profy = ProfyApp(
    client_id=os.environ["PROFY_APP_ID"],
    client_secret=os.environ["PROFY_APP_SECRET"],
    on_token_refresh=lambda user_id, new_token: token_store.save(user_id, new_token),
)

使用 Redis 自定义存储

import { ProfyApp, type OAuthToken } from "@profy-ai/sdk";
import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL!);
const TOKEN_PREFIX = "profy:token:";
const TOKEN_TTL = 90 * 24 * 60 * 60; // 90 days

async function saveToken(userId: string, token: OAuthToken) {
  await redis.set(
    `${TOKEN_PREFIX}${userId}`,
    JSON.stringify(token),
    "EX",
    TOKEN_TTL
  );
}

async function getToken(userId: string): Promise<OAuthToken | null> {
  const raw = await redis.get(`${TOKEN_PREFIX}${userId}`);
  return raw ? JSON.parse(raw) : null;
}

async function deleteToken(userId: string) {
  await redis.del(`${TOKEN_PREFIX}${userId}`);
}

const profy = new ProfyApp({
  clientId: process.env.PROFY_APP_ID!,
  clientSecret: process.env.PROFY_APP_SECRET!,
  onTokenRefresh: saveToken,
});

import json
import os
from redis.asyncio import Redis
from profy_sdk import ProfyApp

redis = Redis.from_url(os.environ["REDIS_URL"])
TOKEN_PREFIX = "profy:token:"
TOKEN_TTL = 90 * 24 * 60 * 60

async def save_token(user_id: str, token: dict):
    await redis.set(
        f"{TOKEN_PREFIX}{user_id}",
        json.dumps(token),
        ex=TOKEN_TTL,
    )

async def get_token(user_id: str) -> dict | None:
    raw = await redis.get(f"{TOKEN_PREFIX}{user_id}")
    return json.loads(raw) if raw else None

async def delete_token(user_id: str):
    await redis.delete(f"{TOKEN_PREFIX}{user_id}")

profy = ProfyApp(
    client_id=os.environ["PROFY_APP_ID"],
    client_secret=os.environ["PROFY_APP_SECRET"],
    on_token_refresh=save_token,
)

Step 2: 多用户 Token 管理

SaaS 应用通常为每个用户维护独立的 Token。核心模式：以 userId 为键，隔离读写。

import { isTokenExpired, AuthExpired } from "@profy-ai/sdk";

class TokenManager {
  constructor(
    private profy: ProfyApp,
    private store: {
      get: (userId: string) => Promise<OAuthToken | null>;
      save: (userId: string, token: OAuthToken) => Promise<void>;
      delete: (userId: string) => Promise<void>;
    }
  ) {}

  async getValidToken(userId: string): Promise<string> {
    const stored = await this.store.get(userId);
    if (!stored) {
      throw new AuthExpired("No token found for user");
    }

    if (!isTokenExpired(stored)) {
      return stored.accessToken;
    }

    const refreshed = await this.profy.refreshToken(stored.refreshToken);
    await this.store.save(userId, refreshed);
    return refreshed.accessToken;
  }

  async revokeUser(userId: string) {
    const stored = await this.store.get(userId);
    if (stored) {
      await this.profy.revokeToken(stored.accessToken);
      await this.store.delete(userId);
    }
  }
}

from profy_sdk import ProfyApp, AuthExpired

class TokenManager:
    def __init__(self, profy: ProfyApp, store):
        self.profy = profy
        self.store = store

    async def get_valid_token(self, user_id: str) -> str:
        stored = await self.store.get(user_id)
        if not stored:
            raise AuthExpired("No token found for user")

        if not stored.is_expired:
            return stored.access_token

        refreshed = await self.profy.refresh_token(stored.refresh_token)
        await self.store.save(user_id, refreshed)
        return refreshed.access_token

    async def revoke_user(self, user_id: str):
        stored = await self.store.get(user_id)
        if stored:
            await self.profy.revoke_token(stored.access_token)
            await self.store.delete(user_id)

Step 3: 并发刷新竞争

问题：两个请求同时发现 Access Token 过期，都尝试刷新。第一个成功，第二个因 Refresh Token 已轮换而收到 invalid_grant。 解决方案：用互斥锁保证同一用户同一时刻只有一个刷新操作。

import { isTokenExpired } from "@profy-ai/sdk";

class SafeTokenManager {
  private refreshLocks = new Map<string, Promise<OAuthToken>>();

  constructor(
    private profy: ProfyApp,
    private store: {
      get: (userId: string) => Promise<OAuthToken | null>;
      save: (userId: string, token: OAuthToken) => Promise<void>;
    }
  ) {}

  async getValidToken(userId: string): Promise<string> {
    const stored = await this.store.get(userId);
    if (!stored) throw new Error("No token found");

    if (!isTokenExpired(stored)) {
      return stored.accessToken;
    }

    const refreshed = await this.refreshWithLock(userId, stored.refreshToken);
    return refreshed.accessToken;
  }

  private async refreshWithLock(
    userId: string,
    refreshToken: string
  ): Promise<OAuthToken> {
    const existing = this.refreshLocks.get(userId);
    if (existing) return existing;

    const promise = this.doRefresh(userId, refreshToken).finally(() => {
      this.refreshLocks.delete(userId);
    });

    this.refreshLocks.set(userId, promise);
    return promise;
  }

  private async doRefresh(
    userId: string,
    refreshToken: string
  ): Promise<OAuthToken> {
    const refreshed = await this.profy.refreshToken(refreshToken);
    await this.store.save(userId, refreshed);
    return refreshed;
  }
}

import asyncio
from profy_sdk import ProfyApp

class SafeTokenManager:
    def __init__(self, profy: ProfyApp, store):
        self.profy = profy
        self.store = store
        self._refresh_locks: dict[str, asyncio.Lock] = {}
        self._refresh_tasks: dict[str, asyncio.Task] = {}

    def _get_lock(self, user_id: str) -> asyncio.Lock:
        if user_id not in self._refresh_locks:
            self._refresh_locks[user_id] = asyncio.Lock()
        return self._refresh_locks[user_id]

    async def get_valid_token(self, user_id: str) -> str:
        stored = await self.store.get(user_id)
        if not stored:
            raise Exception("No token found")

        if not stored.is_expired:
            return stored.access_token

        return await self._refresh_with_lock(user_id, stored.refresh_token)

    async def _refresh_with_lock(self, user_id: str, refresh_token: str) -> str:
        lock = self._get_lock(user_id)
        async with lock:
            fresh = await self.store.get(user_id)
            if fresh and not fresh.is_expired:
                return fresh.access_token

            refreshed = await self.profy.refresh_token(refresh_token)
            await self.store.save(user_id, refreshed)
            return refreshed.access_token

Python 版本在获取锁后重新检查 Token 是否已被其他协程刷新（double-check pattern），避免重复刷新。TypeScript 版本通过共享同一个 Promise 实现相同效果。

对于多实例部署（多 Pod / 多进程），进程内的 Map / dict 锁不够。需要分布式锁：

import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL!);
const LOCK_TTL = 10; // seconds

async function refreshWithDistributedLock(
  userId: string,
  refreshToken: string
): Promise<OAuthToken> {
  const lockKey = `profy:refresh_lock:${userId}`;
  const acquired = await redis.set(lockKey, "1", "EX", LOCK_TTL, "NX");

  if (!acquired) {
    await new Promise((r) => setTimeout(r, 1000));
    const fresh = await getToken(userId);
    if (fresh && !isTokenExpired(fresh)) return fresh;
    throw new Error("Refresh in progress by another instance");
  }

  try {
    const refreshed = await profy.refreshToken(refreshToken);
    await saveToken(userId, refreshed);
    return refreshed;
  } finally {
    await redis.del(lockKey);
  }
}

import asyncio
import redis.asyncio as aioredis

redis_client = aioredis.from_url(os.environ["REDIS_URL"])
LOCK_TTL = 10  # seconds

async def refresh_with_distributed_lock(
    user_id: str, refresh_token: str
) -> dict:
    lock_key = f"profy:refresh_lock:{user_id}"
    acquired = await redis_client.set(lock_key, "1", ex=LOCK_TTL, nx=True)

    if not acquired:
        await asyncio.sleep(1)
        fresh = await get_token(user_id)
        if fresh and not is_token_expired(fresh):
            return fresh
        raise Exception("Refresh in progress by another instance")

    try:
        refreshed = await profy.refresh_token(refresh_token)
        await save_token(user_id, refreshed)
        return refreshed
    finally:
        await redis_client.delete(lock_key)

Step 4: Token 过期降级

当 Refresh Token 也过期（90 天未使用）或被撤销时，SDK 抛出 AuthExpired。此时唯一的恢复路径是引导用户重新授权。

import { AuthExpired } from "@profy-ai/sdk";

export async function POST(request: NextRequest) {
  try {
    const token = await tokenManager.getValidToken(userId);
    const result = await profy.reportEvent("generate", { token });
    return NextResponse.json(result);
  } catch (error) {
    if (error instanceof AuthExpired) {
      await tokenManager.revokeUser(userId);

      return NextResponse.json(
        {
          error: "auth_expired",
          reauthorizeUrl: buildAuthUrl(),
          message: "授权已过期，请重新连接袋袋账号",
        },
        { status: 401 }
      );
    }
    throw error;
  }
}

from profy_sdk import AuthExpired
from fastapi import HTTPException

@app.post("/api/report")
async def report_event(user_id: str, event_name: str):
    try:
        token = await token_manager.get_valid_token(user_id)
        result = await profy.report_event(event_name, token=token)
        return result
    except AuthExpired:
        await token_manager.revoke_user(user_id)

        raise HTTPException(
            status_code=401,
            detail={
                "error": "auth_expired",
                "reauthorize_url": build_auth_url(),
                "message": "授权已过期，请重新连接袋袋账号",
            },
        )

前端处理 auth_expired 响应：

前端降级组件

"use client";

import { useState } from "react";

function ReauthorizePrompt({ authUrl }: { authUrl: string }) {
  return (
    <div className="rounded-lg border border-amber-200 bg-amber-50 p-4">
      <p className="text-sm text-amber-800">
        你的袋袋授权已过期，需要重新连接才能继续使用。
      </p>
      <a
        href={authUrl}
        className="mt-2 inline-block rounded bg-amber-600 px-4 py-2 text-sm text-white"
      >
        重新连接袋袋
      </a>
    </div>
  );
}

Step 5: Token 撤销

用户主动断开连接（取消授权）或你需要清理不再使用的 Token 时，调用撤销接口。

async function disconnectUser(userId: string) {
  const stored = await tokenStore.get(userId);
  if (!stored) return;

  await profy.revokeToken(stored.accessToken);
  await tokenStore.delete(userId);
}

async def disconnect_user(user_id: str):
    stored = await token_store.get(user_id)
    if not stored:
        return

    await profy.revoke_token(stored.access_token)
    await token_store.delete(user_id)

撤销后，该用户的 Access Token 和 Refresh Token 立即失效。后续 API 调用会收到 401。确保在撤销前已完成所有进行中的请求。

撤销的典型触发场景：

用户在你的 App 设置页点击「断开袋袋连接」
用户删除账号
管理员批量清理不活跃授权
检测到异常访问模式

Step 6: 安全存储

静态加密

Token 等同于用户凭证，在数据库中应加密存储：

import { createCipheriv, createDecipheriv, randomBytes } from "crypto";

const ALGORITHM = "aes-256-gcm";
const KEY = Buffer.from(process.env.TOKEN_ENCRYPTION_KEY!, "hex"); // 32 bytes

function encrypt(plaintext: string): string {
  const iv = randomBytes(16);
  const cipher = createCipheriv(ALGORITHM, KEY, iv);
  const encrypted = Buffer.concat([
    cipher.update(plaintext, "utf8"),
    cipher.final(),
  ]);
  const tag = cipher.getAuthTag();
  return [iv.toString("hex"), encrypted.toString("hex"), tag.toString("hex")].join(":");
}

function decrypt(ciphertext: string): string {
  const [ivHex, encHex, tagHex] = ciphertext.split(":");
  const decipher = createDecipheriv(
    ALGORITHM,
    KEY,
    Buffer.from(ivHex, "hex")
  );
  decipher.setAuthTag(Buffer.from(tagHex, "hex"));
  return decipher.update(encHex, "hex", "utf8") + decipher.final("utf8");
}

import os
from cryptography.fernet import Fernet

KEY = os.environ["TOKEN_ENCRYPTION_KEY"].encode()
fernet = Fernet(KEY)

def encrypt(plaintext: str) -> str:
    return fernet.encrypt(plaintext.encode()).decode()

def decrypt(ciphertext: str) -> str:
    return fernet.decrypt(ciphertext.encode()).decode()

安全清单

措施	说明
加密存储	Token 在 DB / Redis 中以密文存储
禁止日志输出	日志中不打印 Token 值（用 `userId` 或 Token 前 8 位替代）
HTTPS Only	Token 传输仅通过 HTTPS
HttpOnly Cookie	如果用 cookie 传递 Token 引用，设置 `HttpOnly` + `Secure` + `SameSite`
密钥轮换	加密密钥定期轮换，支持多密钥解密（新密钥加密、新旧密钥均可解密）
最小权限	OAuth scope 只申请必要权限（如 `events:write`）

反模式

以下是生产环境中常见的 Token 管理错误，每一条都曾导致真实的线上事故。

1. 前端 localStorage 存储 Token

// ❌ 任何 XSS 攻击都能窃取 Token
localStorage.setItem("profy_token", JSON.stringify(token));

Token 应该只存在于服务端。前端通过 HttpOnly Cookie 或 Session ID 引用。

2. 忽略 onTokenRefresh 回调

// ❌ SDK 自动刷新后新 Token 丢失
const profy = new ProfyApp({
  clientId: "...",
  clientSecret: "...",
  // onTokenRefresh 未设置 → 新 Token 没有持久化
});

刷新后旧 Refresh Token 已失效。如果新 Token 没有保存，下次刷新会失败，用户被迫重新授权。

3. 使用已轮换的 Refresh Token

// ❌ 缓存了旧 Token，刷新后仍用旧的
const cachedToken = memoryCache.get(userId); // 可能是过时的
await profy.refreshToken(cachedToken.refreshToken); // invalid_grant

每次从持久化存储读取最新 Token，不要依赖内存缓存。

4. 不处理并发刷新

// ❌ 两个并发请求各自刷新 → 第二个一定失败
async function getToken(userId: string) {
  const stored = await db.getToken(userId);
  if (isTokenExpired(stored)) {
    return await profy.refreshToken(stored.refreshToken); // 无锁
  }
  return stored;
}

参照 Step 3 的互斥锁模式。

生产环境清单

上线前逐项确认：

下一步

Next.js 全栈集成

从零构建完整的 OAuth + 计费 Next.js 应用

Python FastAPI 集成

Python SDK + SQLAlchemy Token 持久化

Webhook 事件监听

接收并处理袋袋平台推送的实时事件

SDK 完整指南

双语言 API 详解与 Token 持久化

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

Token 管理最佳实践

你将构建什么

Token 生命周期

核心原则

Step 1: 选择存储策略

使用 Contrib 存储

使用 Redis 自定义存储

Step 2: 多用户 Token 管理

Step 3: 并发刷新竞争

Step 4: Token 过期降级

Step 5: Token 撤销

Step 6: 安全存储

静态加密

安全清单

反模式

1. 前端 localStorage 存储 Token

2. 忽略 onTokenRefresh 回调

3. 使用已轮换的 Refresh Token

4. 不处理并发刷新

生产环境清单

下一步

Next.js 全栈集成

Python FastAPI 集成

Webhook 事件监听

SDK 完整指南

​你将构建什么

​Token 生命周期

​核心原则

​Step 1: 选择存储策略

​使用 Contrib 存储

​使用 Redis 自定义存储

​Step 2: 多用户 Token 管理

​Step 3: 并发刷新竞争

​Step 4: Token 过期降级

​Step 5: Token 撤销

​Step 6: 安全存储

​静态加密

​安全清单

​反模式

​1. 前端 localStorage 存储 Token

​2. 忽略 onTokenRefresh 回调

​3. 使用已轮换的 Refresh Token

​4. 不处理并发刷新

​生产环境清单

​下一步

Next.js 全栈集成

Python FastAPI 集成

Webhook 事件监听

SDK 完整指南

你将构建什么

Token 生命周期

核心原则

Step 1: 选择存储策略

使用 Contrib 存储

使用 Redis 自定义存储

Step 2: 多用户 Token 管理

Step 3: 并发刷新竞争

Step 4: Token 过期降级

Step 5: Token 撤销

Step 6: 安全存储

静态加密

安全清单

反模式

1. 前端 localStorage 存储 Token

2. 忽略 onTokenRefresh 回调

3. 使用已轮换的 Refresh Token

4. 不处理并发刷新

生产环境清单

下一步