Token Management Best Practices

What You’ll Build

A production-grade Token management layer — covering persistence storage selection, multi-user isolation, concurrent refresh race conditions, expiry degradation, proactive revocation, and secure storage. This ensures your SaaS application works correctly even after Token rotation and service restarts.

Token Lifecycle

Token	Validity	Rotation Rules
Access Token	1 hour (JWT)	New one issued on each refresh
Refresh Token	90 days	Invalidated on each use, new Refresh Token issued simultaneously

Core Principles

Violating any of the following will cause users to silently lose their session — no error message, just a sudden “please log in again.”

1. Always persist after refresh — The onTokenRefresh callback is your only chance to save the new Token. Miss it, and the old Refresh Token is already invalidated while the new Token is lost — the user must re-authorize. 2. Never use in-memory-only storage — Service restart = all user Tokens lost = everyone must log in again. Tokens must be persisted to disk (DB / Redis / file). 3. Rotation must be atomic — After each refreshToken() call, the old Access Token and Refresh Token are both invalidated. Saving the new Token and discarding the old one must happen in the same transaction.

Step 1: Choose a Storage Strategy

Approach	Use Case	Advantages	Disadvantages
DrizzleTokenStore (TS) / SQLAlchemyTokenStore (Python)	Projects with existing relational DB	Zero config, transaction-safe, built-in schema	Requires DB connection
Redis	High-concurrency SaaS, session management	High performance, native TTL, atomic operations	Requires Redis instance, persistence config
Custom DB (Prisma / Mongoose / etc.)	Projects with non-Drizzle ORMs	Full schema control	Requires implementing CRUD yourself
Encrypted file	CLI tools, single-user scripts	No external dependencies	Not suitable for multi-user, no concurrency control

For 80% of use cases, the Contrib DrizzleTokenStore or SQLAlchemyTokenStore will work. You only need a custom implementation for special requirements (Redis sessions, custom encryption).

Using Contrib Storage

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),
)

Using Redis Custom Storage

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: Multi-User Token Management

SaaS applications typically maintain independent Tokens for each user. The core pattern: use userId as the key, isolate reads and writes.

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: Concurrent Refresh Race Conditions

Problem: Two requests simultaneously discover the Access Token is expired and both attempt to refresh. The first succeeds, but the second receives invalid_grant because the Refresh Token has already been rotated. Solution: Use a mutex to ensure only one refresh operation runs at a time per user.

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

The Python version re-checks if the Token was already refreshed by another coroutine after acquiring the lock (double-check pattern), avoiding redundant refreshes. The TypeScript version achieves the same effect by sharing a single Promise.

For multi-instance deployments (multiple pods / processes), in-process Map / dict locks aren’t sufficient. You need a distributed lock:

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 Expiry Degradation

When the Refresh Token also expires (90 days without use) or is revoked, the SDK throws AuthExpired. The only recovery path is guiding the user to re-authorize.

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: "Authorization expired. Please reconnect your Profy account.",
        },
        { 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": "Authorization expired. Please reconnect your Profy account.",
            },
        )

Frontend handling of the auth_expired response:

Frontend Degradation Component

"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">
        Your Profy authorization has expired. You need to reconnect to continue using the service.
      </p>
      <a
        href={authUrl}
        className="mt-2 inline-block rounded bg-amber-600 px-4 py-2 text-sm text-white"
      >
        Reconnect Profy
      </a>
    </div>
  );
}

Step 5: Token Revocation

When users proactively disconnect (revoke authorization) or you need to clean up unused Tokens, call the revocation API.

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)

After revocation, the user’s Access Token and Refresh Token are immediately invalidated. Subsequent API calls will receive 401. Make sure all in-flight requests are completed before revoking.

Typical revocation triggers:

User clicks “Disconnect Profy” in your App settings
User deletes their account
Admin bulk-cleans inactive authorizations
Abnormal access patterns detected

Step 6: Secure Storage

Encryption at Rest

Tokens are equivalent to user credentials and should be stored encrypted in the database:

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()

Security Checklist

Measure	Description
Encrypted storage	Tokens stored as ciphertext in DB / Redis
No logging	Never log Token values (use `userId` or first 8 characters instead)
HTTPS only	Tokens transmitted only over HTTPS
HttpOnly Cookie	If using cookies to pass Token references, set `HttpOnly` + `Secure` + `SameSite`
Key rotation	Rotate encryption keys periodically, supporting multi-key decryption (encrypt with new key, decrypt with both old and new)
Least privilege	Only request necessary OAuth scopes (e.g., `events:write`)

Anti-Patterns

The following are common Token management mistakes in production environments — each one has caused real production incidents.

1. Storing Tokens in Frontend localStorage

// ❌ Any XSS attack can steal Tokens
localStorage.setItem("profy_token", JSON.stringify(token));

Tokens should only exist on the server side. The frontend references them via HttpOnly cookies or Session IDs.

2. Ignoring the onTokenRefresh Callback

// ❌ New Token is lost after SDK auto-refresh
const profy = new ProfyApp({
  clientId: "...",
  clientSecret: "...",
  // onTokenRefresh not set → new Token isn't persisted
});

After refresh, the old Refresh Token is already invalidated. If the new Token isn’t saved, the next refresh will fail and the user will be forced to re-authorize.

3. Using an Already-Rotated Refresh Token

// ❌ Cached an old Token, still using the old one after refresh
const cachedToken = memoryCache.get(userId); // Might be stale
await profy.refreshToken(cachedToken.refreshToken); // invalid_grant

Always read the latest Token from persistent storage — don’t rely on in-memory caches.

4. Not Handling Concurrent Refreshes

// ❌ Two concurrent requests each refresh → the second one will always fail
async function getToken(userId: string) {
  const stored = await db.getToken(userId);
  if (isTokenExpired(stored)) {
    return await profy.refreshToken(stored.refreshToken); // No lock
  }
  return stored;
}

Refer to the mutex pattern in Step 3.

Production Checklist

Verify each item before going live:

Next Steps

Next.js Full-Stack Integration

Build a complete OAuth + billing Next.js application from scratch

Python FastAPI Integration

Python SDK + SQLAlchemy Token persistence

Webhook Event Handling

Receive and process real-time events pushed by the Profy platform

SDK Complete Guide

Dual-language API reference and Token persistence

Get Started

Authentication & Keys

API Guides

SDK

Integrations

Cookbook

Reference

Token Management Best Practices

What You’ll Build

Token Lifecycle

Core Principles

Step 1: Choose a Storage Strategy

Using Contrib Storage

Using Redis Custom Storage

Step 2: Multi-User Token Management

Step 3: Concurrent Refresh Race Conditions

Step 4: Token Expiry Degradation

Step 5: Token Revocation

Step 6: Secure Storage

Encryption at Rest

Security Checklist

Anti-Patterns

1. Storing Tokens in Frontend localStorage

2. Ignoring the onTokenRefresh Callback

3. Using an Already-Rotated Refresh Token

4. Not Handling Concurrent Refreshes

Production Checklist

Next Steps

Next.js Full-Stack Integration

Python FastAPI Integration

Webhook Event Handling

SDK Complete Guide

​What You’ll Build

​Token Lifecycle

​Core Principles

​Step 1: Choose a Storage Strategy

​Using Contrib Storage

​Using Redis Custom Storage

​Step 2: Multi-User Token Management

​Step 3: Concurrent Refresh Race Conditions

​Step 4: Token Expiry Degradation

​Step 5: Token Revocation

​Step 6: Secure Storage

​Encryption at Rest

​Security Checklist

​Anti-Patterns

​1. Storing Tokens in Frontend localStorage

​2. Ignoring the onTokenRefresh Callback

​3. Using an Already-Rotated Refresh Token

​4. Not Handling Concurrent Refreshes

​Production Checklist

​Next Steps

Next.js Full-Stack Integration

Python FastAPI Integration

Webhook Event Handling

SDK Complete Guide

What You’ll Build

Token Lifecycle

Core Principles

Step 1: Choose a Storage Strategy

Using Contrib Storage

Using Redis Custom Storage

Step 2: Multi-User Token Management

Step 3: Concurrent Refresh Race Conditions

Step 4: Token Expiry Degradation

Step 5: Token Revocation

Step 6: Secure Storage

Encryption at Rest

Security Checklist

Anti-Patterns

1. Storing Tokens in Frontend localStorage

2. Ignoring the onTokenRefresh Callback

3. Using an Already-Rotated Refresh Token

4. Not Handling Concurrent Refreshes

Production Checklist

Next Steps