Upstash

# SDK for JavaScript/TypeScript
npm install @upstash/redis

# SDK for Python
pip install upstash-redis

// lib/redis.ts
import { Redis } from '@upstash/redis'

// Option 1: With environment variables
export const redis = Redis.fromEnv()

// Option 2: With explicit config
export const redis = new Redis({
  url: process.env.UPSTASH_REDIS_REST_URL!,
  token: process.env.UPSTASH_REDIS_REST_TOKEN!,
})

// Option 3: For Edge Runtime
export const redis = new Redis({
  url: 'https://xxx.upstash.io',
  token: 'AXxxxx',
  automaticDeserialization: true, // Automatic JSON parse
})

# .env.local
UPSTASH_REDIS_REST_URL=https://xxx.upstash.io
UPSTASH_REDIS_REST_TOKEN=AXxxxx

import { redis } from '@/lib/redis'

// STRING operations
await redis.set('user:1:name', 'John Doe')
const name = await redis.get('user:1:name') // "John Doe"

// SET with TTL (expiration)
await redis.set('session:abc123', { userId: 1 }, { ex: 3600 }) // 1 hour

// SET with conditions
await redis.set('lock:resource', 'locked', { nx: true }) // Only if it does not exist
await redis.set('config:version', '2.0', { xx: true }) // Only if it already exists

// GET with typing
interface User {
  id: number
  name: string
  email: string
}
const user = await redis.get<User>('user:1') // Typed!

// MGET/MSET - multiple keys
await redis.mset({
  'key1': 'value1',
  'key2': 'value2',
  'key3': 'value3',
})
const values = await redis.mget('key1', 'key2', 'key3')

// Increment/Decrement
await redis.incr('page:views')
await redis.incrby('user:1:points', 10)
await redis.decr('inventory:item:123')

// Append
await redis.append('log:today', '\n2024-01-15: User logged in')

// String length
const length = await redis.strlen('user:1:bio')

// HSET - set hash fields
await redis.hset('user:1', {
  name: 'John Doe',
  email: 'john@example.com',
  age: 30,
  verified: true,
})

// HGET - get a single field
const email = await redis.hget('user:1', 'email')

// HMGET - get multiple fields
const [name, email] = await redis.hmget('user:1', 'name', 'email')

// HGETALL - get the entire hash
const user = await redis.hgetall<User>('user:1')

// HINCRBY - increment a numeric value
await redis.hincrby('user:1', 'age', 1)

// HDEL - delete a field
await redis.hdel('user:1', 'temporary_field')

// HEXISTS - check if a field exists
const hasEmail = await redis.hexists('user:1', 'email')

// HKEYS/HVALS - keys and values
const keys = await redis.hkeys('user:1')
const values = await redis.hvals('user:1')

// LPUSH/RPUSH - add elements
await redis.lpush('queue:tasks', 'task1', 'task2')
await redis.rpush('queue:tasks', 'task3')

// LPOP/RPOP - retrieve and remove an element
const task = await redis.lpop('queue:tasks')
const lastTask = await redis.rpop('queue:tasks')

// LRANGE - get a range
const tasks = await redis.lrange('queue:tasks', 0, -1) // All elements
const firstFive = await redis.lrange('queue:tasks', 0, 4)

// LLEN - list length
const queueLength = await redis.llen('queue:tasks')

// LINDEX - element at a given index
const secondTask = await redis.lindex('queue:tasks', 1)

// LSET - set element at a given index
await redis.lset('queue:tasks', 0, 'updated_task')

// LTRIM - trim the list
await redis.ltrim('notifications', 0, 99) // Keep the last 100

// SADD - add elements
await redis.sadd('tags:post:1', 'javascript', 'react', 'typescript')

// SMEMBERS - all elements
const tags = await redis.smembers('tags:post:1')

// SISMEMBER - check if an element belongs to the set
const hasTag = await redis.sismember('tags:post:1', 'react')

// SCARD - number of elements
const tagCount = await redis.scard('tags:post:1')

// SREM - remove an element
await redis.srem('tags:post:1', 'deprecated')

// SINTER - intersection of sets
const commonTags = await redis.sinter('tags:post:1', 'tags:post:2')

// SUNION - union of sets
const allTags = await redis.sunion('tags:post:1', 'tags:post:2')

// SDIFF - difference of sets
const uniqueTags = await redis.sdiff('tags:post:1', 'tags:post:2')

// ZADD - add with score
await redis.zadd('leaderboard', {
  score: 100,
  member: 'player1',
})
await redis.zadd('leaderboard',
  { score: 150, member: 'player2' },
  { score: 80, member: 'player3' }
)

// ZRANGE - get a range (ascending)
const topPlayers = await redis.zrange('leaderboard', 0, 9, {
  rev: true, // Descending (highest first)
  withScores: true,
})

// ZSCORE - get the score of a member
const score = await redis.zscore('leaderboard', 'player1')

// ZRANK - position in the ranking
const rank = await redis.zrank('leaderboard', 'player1')
const rankFromTop = await redis.zrevrank('leaderboard', 'player1')

// ZINCRBY - increment score
await redis.zincrby('leaderboard', 10, 'player1')

// ZRANGEBYSCORE - range by score
const playersAbove100 = await redis.zrangebyscore('leaderboard', 100, '+inf')

// ZREMRANGEBYRANK - remove a range
await redis.zremrangebyrank('leaderboard', 0, 9) // Remove the weakest 10

import { redis } from '@/lib/redis'

// Save a JSON object
await redis.json.set('product:1', '$', {
  name: 'MacBook Pro',
  price: 2499,
  specs: {
    cpu: 'M3 Pro',
    ram: '18GB',
    storage: '512GB',
  },
  tags: ['laptop', 'apple', 'pro'],
})

// Get the entire object
const product = await redis.json.get('product:1')

// Get a specific path
const price = await redis.json.get('product:1', '$.price')
const cpu = await redis.json.get('product:1', '$.specs.cpu')

// Update a nested field
await redis.json.set('product:1', '$.price', 2299)
await redis.json.set('product:1', '$.specs.ram', '36GB')

// JSON array operations
await redis.json.arrappend('product:1', '$.tags', 'new-tag')
await redis.json.arrlen('product:1', '$.tags')

// Numeric operations
await redis.json.numincrby('product:1', '$.price', -100) // Discount

// Pipeline - multiple operations in a single request
const pipeline = redis.pipeline()

pipeline.set('key1', 'value1')
pipeline.set('key2', 'value2')
pipeline.incr('counter')
pipeline.hset('user:1', { lastActive: Date.now() })

const results = await pipeline.exec()
// results = [['OK', null], ['OK', null], [1, null], [1, null]]

// Multi/Exec - atomic transaction
const tx = redis.multi()

tx.decrby('account:A:balance', 100)
tx.incrby('account:B:balance', 100)

await tx.exec() // Atomic fund transfer

npm install @upstash/ratelimit

import { Ratelimit } from '@upstash/ratelimit'
import { Redis } from '@upstash/redis'

// Create a rate limiter
const ratelimit = new Ratelimit({
  redis: Redis.fromEnv(),
  limiter: Ratelimit.slidingWindow(10, '10 s'), // 10 requests per 10 seconds
  analytics: true, // Enable analytics dashboard
  prefix: '@upstash/ratelimit',
})

// Middleware for Next.js
export async function middleware(request: NextRequest) {
  const ip = request.headers.get('x-forwarded-for') ?? '127.0.0.1'

  const { success, limit, remaining, reset } = await ratelimit.limit(ip)

  if (!success) {
    return new NextResponse('Too Many Requests', {
      status: 429,
      headers: {
        'X-RateLimit-Limit': limit.toString(),
        'X-RateLimit-Remaining': remaining.toString(),
        'X-RateLimit-Reset': reset.toString(),
        'Retry-After': Math.ceil((reset - Date.now()) / 1000).toString(),
      },
    })
  }

  return NextResponse.next()
}

// Fixed Window - the simplest
const fixedWindow = new Ratelimit({
  redis,
  limiter: Ratelimit.fixedWindow(100, '1 h'), // 100/hour
})

// Sliding Window - more fair
const slidingWindow = new Ratelimit({
  redis,
  limiter: Ratelimit.slidingWindow(10, '10 s'),
})

// Token Bucket - burst-friendly
const tokenBucket = new Ratelimit({
  redis,
  limiter: Ratelimit.tokenBucket(10, '1 s', 30), // 10 tokens/sec, max 30
})

// Cached - more efficient for high traffic
const cached = new Ratelimit({
  redis,
  limiter: Ratelimit.cachedFixedWindow(100, '10 s'),
  ephemeralCache: new Map(), // In-memory cache
})

// Rate limit per authenticated user
export async function POST(request: Request) {
  const session = await getSession()
  const identifier = session?.user?.id ?? 'anonymous'

  const { success, remaining } = await ratelimit.limit(identifier)

  if (!success) {
    return Response.json(
      { error: 'Rate limit exceeded' },
      { status: 429 }
    )
  }

  // Add info to response headers
  const response = await handleRequest(request)
  response.headers.set('X-RateLimit-Remaining', remaining.toString())

  return response
}

// Different limits for different tiers
const limits = {
  free: new Ratelimit({
    redis,
    limiter: Ratelimit.slidingWindow(100, '1 d'),
  }),
  pro: new Ratelimit({
    redis,
    limiter: Ratelimit.slidingWindow(10000, '1 d'),
  }),
  enterprise: new Ratelimit({
    redis,
    limiter: Ratelimit.slidingWindow(1000000, '1 d'),
  }),
}

export async function POST(request: Request) {
  const user = await getUser()
  const limiter = limits[user.tier] || limits.free

  const { success } = await limiter.limit(user.id)
  // ...
}

npm install @upstash/qstash

import { Client } from '@upstash/qstash'

const qstash = new Client({
  token: process.env.QSTASH_TOKEN!,
})

// Send a message to an endpoint
await qstash.publishJSON({
  url: 'https://my-app.com/api/process',
  body: {
    userId: '123',
    action: 'send-email',
    data: { template: 'welcome' },
  },
})

// With delay
await qstash.publishJSON({
  url: 'https://my-app.com/api/reminder',
  body: { userId: '123' },
  delay: '10m', // 10 minutes
})

// With retry
await qstash.publishJSON({
  url: 'https://my-app.com/api/webhook',
  body: { event: 'order.created' },
  retries: 5,
  callback: 'https://my-app.com/api/callback',
  failureCallback: 'https://my-app.com/api/failure',
})

// One-time scheduled message
await qstash.publishJSON({
  url: 'https://my-app.com/api/report',
  body: { type: 'weekly' },
  notBefore: Math.floor(Date.now() / 1000) + 86400, // In 24 hours
})

// Recurring schedule (cron)
const schedule = await qstash.schedules.create({
  destination: 'https://my-app.com/api/daily-report',
  cron: '0 9 * * *', // Every day at 9:00 UTC
  body: JSON.stringify({ type: 'daily' }),
})

// List schedules
const schedules = await qstash.schedules.list()

// Delete a schedule
await qstash.schedules.delete(schedule.scheduleId)

// app/api/process/route.ts
import { Receiver } from '@upstash/qstash'
import { NextResponse } from 'next/server'

const receiver = new Receiver({
  currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!,
  nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!,
})

export async function POST(request: Request) {
  const signature = request.headers.get('upstash-signature')!
  const body = await request.text()

  // Verify that the message comes from QStash
  const isValid = await receiver.verify({
    signature,
    body,
    url: process.env.VERCEL_URL + '/api/process',
  })

  if (!isValid) {
    return new NextResponse('Invalid signature', { status: 401 })
  }

  const data = JSON.parse(body)

  // Process the message
  await processMessage(data)

  return new NextResponse('OK')
}

// Send multiple messages at once
const messages = users.map(user => ({
  url: 'https://my-app.com/api/notify',
  body: JSON.stringify({ userId: user.id }),
}))

await qstash.batchJSON(messages)

// Send to multiple endpoints
await qstash.publishJSON({
  url: [
    'https://my-app.com/api/analytics',
    'https://my-app.com/api/logging',
    'https://webhook.site/xxx',
  ],
  body: { event: 'user.signup', userId: '123' },
})

// Create a topic
await qstash.topics.create({ name: 'user-events' })

// Add an endpoint to the topic
await qstash.topics.addEndpoint({
  topic: 'user-events',
  endpoint: 'https://service-a.com/webhook',
})
await qstash.topics.addEndpoint({
  topic: 'user-events',
  endpoint: 'https://service-b.com/webhook',
})

// Publish to the topic - all subscribers will receive the message
await qstash.publishJSON({
  topic: 'user-events',
  body: { event: 'user.created', userId: '123' },
})

npm install @upstash/kafka

import { Kafka } from '@upstash/kafka'

const kafka = new Kafka({
  url: process.env.UPSTASH_KAFKA_REST_URL!,
  username: process.env.UPSTASH_KAFKA_REST_USERNAME!,
  password: process.env.UPSTASH_KAFKA_REST_PASSWORD!,
})

const producer = kafka.producer()

// Send a single message
await producer.produce('orders', {
  value: JSON.stringify({
    orderId: '123',
    items: ['item1', 'item2'],
    total: 99.99,
  }),
})

// With a partition key
await producer.produce('user-events', {
  key: 'user:123', // All events for this user go to the same partition
  value: JSON.stringify({
    event: 'page_view',
    page: '/products',
  }),
})

// Batch produce
await producer.produceMany([
  { topic: 'logs', value: 'Log entry 1' },
  { topic: 'logs', value: 'Log entry 2' },
  { topic: 'logs', value: 'Log entry 3' },
])

const consumer = kafka.consumer()

// Consume messages
const messages = await consumer.consume({
  consumerGroupId: 'my-group',
  instanceId: 'instance-1',
  topics: ['orders'],
  autoOffsetReset: 'earliest',
})

for (const message of messages) {
  console.log('Topic:', message.topic)
  console.log('Partition:', message.partition)
  console.log('Offset:', message.offset)
  console.log('Value:', message.value)

  // Process the message
  await processOrder(JSON.parse(message.value))
}

// Commit offsets
await consumer.commit({
  consumerGroupId: 'my-group',
  instanceId: 'instance-1',
  topics: [
    {
      topic: 'orders',
      partitions: [
        { partition: 0, offset: messages[messages.length - 1].offset },
      ],
    },
  ],
})

// app/api/cache/route.ts
import { Redis } from '@upstash/redis'

export const runtime = 'edge' // Runs on Edge

const redis = Redis.fromEnv()

export async function GET(request: Request) {
  const { searchParams } = new URL(request.url)
  const key = searchParams.get('key')

  if (!key) {
    return Response.json({ error: 'Key required' }, { status: 400 })
  }

  const cached = await redis.get(key)

  if (cached) {
    return Response.json({ data: cached, source: 'cache' })
  }

  // Fetch fresh data
  const data = await fetchFreshData(key)
  await redis.set(key, data, { ex: 3600 })

  return Response.json({ data, source: 'fresh' })
}

// Vercel KV is built on top of Upstash Redis
import { kv } from '@vercel/kv'

// Identical API
await kv.set('key', 'value')
const value = await kv.get('key')
await kv.hset('hash', { field: 'value' })

// worker.ts
import { Redis } from '@upstash/redis/cloudflare'

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const redis = new Redis({
      url: env.UPSTASH_REDIS_REST_URL,
      token: env.UPSTASH_REDIS_REST_TOKEN,
    })

    const visits = await redis.incr('page:visits')

    return new Response(`This page has been visited ${visits} times`)
  },
}

// lib/session.ts
import { Redis } from '@upstash/redis'
import { cookies } from 'next/headers'
import { nanoid } from 'nanoid'

const redis = Redis.fromEnv()

interface Session {
  userId: string
  email: string
  createdAt: number
}

export async function createSession(userId: string, email: string) {
  const sessionId = nanoid(32)
  const session: Session = {
    userId,
    email,
    createdAt: Date.now(),
  }

  await redis.set(`session:${sessionId}`, session, {
    ex: 60 * 60 * 24 * 7, // 7 days
  })

  const cookieStore = await cookies()
  cookieStore.set('session', sessionId, {
    httpOnly: true,
    secure: process.env.NODE_ENV === 'production',
    sameSite: 'lax',
    maxAge: 60 * 60 * 24 * 7,
  })

  return session
}

export async function getSession(): Promise<Session | null> {
  const cookieStore = await cookies()
  const sessionId = cookieStore.get('session')?.value

  if (!sessionId) return null

  const session = await redis.get<Session>(`session:${sessionId}`)

  if (!session) return null

  // Refresh TTL on access
  await redis.expire(`session:${sessionId}`, 60 * 60 * 24 * 7)

  return session
}

export async function deleteSession() {
  const cookieStore = await cookies()
  const sessionId = cookieStore.get('session')?.value

  if (sessionId) {
    await redis.del(`session:${sessionId}`)
    cookieStore.delete('session')
  }
}

// lib/cache.ts
import { Redis } from '@upstash/redis'

const redis = Redis.fromEnv()

interface CacheOptions {
  staleTime?: number // Time after which data is considered "stale"
  maxAge?: number    // Maximum time to live
}

export async function cachedFetch<T>(
  key: string,
  fetcher: () => Promise<T>,
  options: CacheOptions = {}
): Promise<{ data: T; stale: boolean }> {
  const { staleTime = 60, maxAge = 3600 } = options

  // Check cache
  const cached = await redis.get<{
    data: T
    timestamp: number
  }>(key)

  const now = Date.now()

  if (cached) {
    const age = (now - cached.timestamp) / 1000

    if (age < staleTime) {
      // Fresh - return from cache
      return { data: cached.data, stale: false }
    }

    if (age < maxAge) {
      // Stale - return from cache, refresh in the background
      refreshInBackground(key, fetcher, maxAge)
      return { data: cached.data, stale: true }
    }
  }

  // No cache or expired - fetch fresh
  const data = await fetcher()
  await redis.set(key, { data, timestamp: now }, { ex: maxAge })

  return { data, stale: false }
}

async function refreshInBackground<T>(
  key: string,
  fetcher: () => Promise<T>,
  maxAge: number
) {
  try {
    const data = await fetcher()
    await redis.set(key, { data, timestamp: Date.now() }, { ex: maxAge })
  } catch (error) {
    console.error('Background refresh failed:', error)
  }
}