← Back to DevBytes
Developer Tutorial July 07, 2026 14 min read

Designing a URL Shortener for 100 Million Users

Understanding URL Shorteners at Scale

A URL shortener transforms long web addresses into compact, easily shareable links. Services like Bitly, TinyURL, and Twitter's t.co process billions of redirects monthly. Building one for 100 million users demands careful architectural planning — every millisecond of latency, every database write, and every cache miss compounds at this scale. This tutorial walks through designing a production-grade URL shortener capable of handling 100 million active users, covering the full stack from hashing algorithms to deployment topology.

Core Requirements

Before writing code, define the system's functional and non-functional requirements:

Traffic Estimation

For 100 million monthly active users, assuming each user creates 2 short URLs per month and visits 10 short URLs per month:

These numbers inform every design decision that follows.

Choosing a Hashing Strategy

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

Option 1: Hash + Base62 Encoding

Generate a hash of the long URL, then encode it in Base62 (characters a-z, A-Z, 0-9) for compactness. MD5 produces 128 bits; truncating to 7 characters gives 62^7 ≈ 3.5 trillion combinations — sufficient for 12 billion URLs with low collision probability.

import hashlib

BASE62_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"

def to_base62(num: int) -> str:
    """Convert an integer to a Base62 string."""
    if num == 0:
        return BASE62_ALPHABET[0]
    chars = []
    base = len(BASE62_ALPHABET)
    while num > 0:
        num, rem = divmod(num, base)
        chars.append(BASE62_ALPHABET[rem])
    return ''.join(reversed(chars))

def generate_short_key(long_url: str, counter: int) -> str:
    """Generate a unique short key using a counter and URL hash."""
    # Combine counter with URL fingerprint for uniqueness
    fingerprint = hashlib.md5(long_url.encode()).hexdigest()
    combined = f"{counter}:{fingerprint}"
    hash_bytes = hashlib.sha256(combined.encode()).digest()
    hash_int = int.from_bytes(hash_bytes[:8], 'big')
    return to_base62(hash_int)

Option 2: Distributed Counter-Based ID Generation

Use a distributed ID generator (like Twitter Snowflake) to produce unique 64-bit integers, then convert to Base62. This avoids hash collisions entirely and provides ordering guarantees.

import time
import os

# Snowflake-inspired ID generator
EPOCH = 1704067200000  # Custom epoch (Jan 1, 2024 in ms)
MACHINE_ID_BITS = 10
SEQUENCE_BITS = 12

MAX_MACHINE_ID = (1 << MACHINE_ID_BITS) - 1
MAX_SEQUENCE = (1 << SEQUENCE_BITS) - 1

MACHINE_ID_SHIFT = SEQUENCE_BITS
TIMESTAMP_SHIFT = SEQUENCE_BITS + MACHINE_ID_BITS

class SnowflakeGenerator:
    def __init__(self, machine_id: int):
        if machine_id > MAX_MACHINE_ID:
            raise ValueError(f"Machine ID must be between 0 and {MAX_MACHINE_ID}")
        self.machine_id = machine_id
        self.sequence = 0
        self.last_timestamp = -1
    
    def _current_millis(self) -> int:
        return int(time.time() * 1000)
    
    def next_id(self) -> int:
        now = self._current_millis()
        
        if now < self.last_timestamp:
            raise Exception("Clock moved backwards — refusing to generate ID")
        
        if now == self.last_timestamp:
            self.sequence = (self.sequence + 1) & MAX_SEQUENCE
            if self.sequence == 0:
                # Sequence exhausted, wait for next millisecond
                while now <= self.last_timestamp:
                    now = self._current_millis()
        else:
            self.sequence = 0
        
        self.last_timestamp = now
        
        id_value = ((now - EPOCH) << TIMESTAMP_SHIFT) | \
                   (self.machine_id << MACHINE_ID_SHIFT) | \
                   self.sequence
        return id_value

# Usage
generator = SnowflakeGenerator(machine_id=int(os.environ.get('MACHINE_ID', '1')))
unique_id = generator.next_id()
short_key = to_base62(unique_id)
print(f"Short key: {short_key}")  # Example: 1aZk9xR

Handling Collisions

Even with 3.5 trillion possible keys, collisions are statistically possible at scale. Implement a collision resolution strategy:

def create_short_url(long_url: str, custom_alias: str = None) -> str:
    """Create a short URL with collision handling."""
    
    if custom_alias:
        # Validate custom alias
        if len(custom_alias) < 4 or len(custom_alias) > 16:
            raise ValueError("Custom alias must be 4-16 characters")
        if not all(c in BASE62_ALPHABET for c in custom_alias):
            raise ValueError("Alias must contain only Base62 characters")
        key = custom_alias
    else:
        # Auto-generate key with collision retry logic
        max_retries = 5
        for attempt in range(max_retries):
            # Add entropy for each retry
            counter = generator.next_id()
            key = to_base62(counter)
            
            # Check if key exists in database
            if not database.key_exists(key):
                break
        else:
            raise Exception("Failed to generate unique key after maximum retries")
    
    # Store mapping in database
    database.insert_mapping(key, long_url)
    return key

Database Design

Schema for Primary Storage

Choose a database that supports high write throughput and horizontal scaling. PostgreSQL with partitioning or Cassandra are excellent choices. Here's a schema using PostgreSQL with declarative partitioning:

-- Partitioned table by hash of the short key
CREATE TABLE url_mappings (
    short_key     VARCHAR(16) NOT NULL,
    long_url      TEXT NOT NULL,
    created_at    TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    expires_at    TIMESTAMP WITH TIME ZONE,
    user_id       BIGINT,
    click_count   BIGINT DEFAULT 0,
    PRIMARY KEY (short_key, created_at)
) PARTITION BY HASH (short_key);

-- Create 64 partitions for even distribution
CREATE TABLE url_mappings_0  PARTITION OF url_mappings FOR VALUES WITH (MODULUS 64, REMAINDER 0);
CREATE TABLE url_mappings_1  PARTITION OF url_mappings FOR VALUES WITH (MODULUS 64, REMAINDER 1);
-- ... create partitions 2 through 62
CREATE TABLE url_mappings_63 PARTITION OF url_mappings FOR VALUES WITH (MODULUS 64, REMAINDER 63);

-- Index for fast lookups
CREATE INDEX idx_short_key ON url_mappings (short_key);

-- Separate analytics table to avoid updating hot rows
CREATE TABLE click_events (
    short_key   VARCHAR(16) NOT NULL,
    timestamp   TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    user_agent  TEXT,
    ip_address  INET,
    referrer    TEXT,
    country     VARCHAR(2)
) PARTITION BY RANGE (timestamp);

-- Monthly partitions for click events
CREATE TABLE click_events_2024_01 PARTITION OF click_events
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE click_events_2024_02 PARTITION OF click_events
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
-- Continue for all months needed

Why Separate Click Events from URL Mappings

Updating a click_count column on the url_mappings table creates write contention on hot rows. Popular URLs could receive thousands of clicks per second, causing row-level locking issues. Instead, append click events to a separate time-series table and compute aggregated counts asynchronously using batch processing or materialized views.

-- Materialized view for pre-computed click counts
CREATE MATERIALIZED VIEW url_click_stats AS
SELECT 
    short_key,
    COUNT(*) AS total_clicks,
    COUNT(DISTINCT ip_address) AS unique_visitors,
    MAX(timestamp) AS last_clicked
FROM click_events
GROUP BY short_key;

-- Refresh every 5 minutes using a scheduled job
-- REFRESH MATERIALIZED VIEW CONCURRENTLY url_click_stats;

API Design

REST Endpoints

Keep the API minimal and fast. Every endpoint must return within strict time budgets:

from fastapi import FastAPI, HTTPException, Request, Query
from fastapi.responses import RedirectResponse, JSONResponse
from pydantic import BaseModel, HttpUrl, validator
import uvicorn

app = FastAPI()

class ShortenRequest(BaseModel):
    url: HttpUrl
    custom_alias: str | None = None
    expiration_days: int | None = None
    
    @validator('custom_alias')
    def validate_alias(cls, v):
        if v and len(v) < 4:
            raise ValueError('Custom alias must be at least 4 characters')
        return v

class ShortenResponse(BaseModel):
    short_key: str
    short_url: str
    original_url: str
    expires_at: str | None

@app.post("/api/v1/shorten", response_model=ShortenResponse, status_code=201)
async def create_short_url(request: ShortenRequest):
    """Create a shortened URL."""
    try:
        short_key = create_short_url(
            long_url=str(request.url),
            custom_alias=request.custom_alias
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    
    return {
        "short_key": short_key,
        "short_url": f"https://short.example.com/{short_key}",
        "original_url": str(request.url),
        "expires_at": None  # Set based on expiration_days if provided
    }

@app.get("/{short_key}")
async def redirect_to_url(short_key: str, request: Request):
    """Redirect a short URL to its original destination."""
    # Check cache first
    long_url = cache.get(short_key)
    
    if long_url is None:
        # Fall back to database
        long_url = database.get_long_url(short_key)
        if long_url is None:
            raise HTTPException(status_code=404, detail="Short URL not found")
        # Populate cache with TTL
        cache.set(short_key, long_url, ttl=3600)  # 1 hour
    
    # Asynchronously log the click event
    await click_logger.log(
        short_key=short_key,
        user_agent=request.headers.get('User-Agent', ''),
        ip_address=request.client.host,
        referrer=request.headers.get('Referer', '')
    )
    
    return RedirectResponse(url=long_url, status_code=301)

Rate Limiting Middleware

At 100 million users, API abuse is inevitable. Implement token-bucket rate limiting per IP and per API key:

import time
from collections import defaultdict

class TokenBucket:
    """Token bucket algorithm for rate limiting."""
    
    def __init__(self, rate: int, capacity: int):
        self.rate = rate          # Tokens per second
        self.capacity = capacity  # Maximum burst size
        self.tokens = capacity
        self.last_refill = time.monotonic()
    
    def consume(self, tokens: int = 1) -> bool:
        """Try to consume tokens. Returns True if allowed."""
        now = time.monotonic()
        elapsed = now - self.last_refill
        
        # Refill tokens based on elapsed time
        self.tokens = min(
            self.capacity,
            self.tokens + elapsed * self.rate
        )
        self.last_refill = now
        
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

class RateLimiter:
    def __init__(self):
        self.buckets: dict[str, TokenBucket] = {}
        # Cleanup old entries periodically
        self.last_cleanup = time.monotonic()
    
    def is_allowed(self, key: str, rate: int = 100, capacity: int = 200) -> bool:
        """Check if request is allowed for given key."""
        now = time.monotonic()
        
        # Periodic cleanup of stale buckets
        if now - self.last_cleanup > 300:  # Every 5 minutes
            stale_keys = [
                k for k, b in self.buckets.items()
                if now - b.last_refill > 3600
            ]
            for k in stale_keys:
                del self.buckets[k]
            self.last_cleanup = now
        
        if key not in self.buckets:
            self.buckets[key] = TokenBucket(rate=rate, capacity=capacity)
        
        return self.buckets[key].consume()

# Usage in FastAPI middleware
rate_limiter = RateLimiter()

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_ip = request.client.host
    
    if not rate_limiter.is_allowed(client_ip, rate=100, capacity=200):
        return JSONResponse(
            status_code=429,
            content={"detail": "Too many requests. Please slow down."}
        )
    
    response = await call_next(request)
    return response

Caching Architecture

Multi-Tier Caching Strategy

For 100 million users generating millions of redirects daily, caching is critical. A single Redis cluster may become a bottleneck. Use a multi-tier approach:

from functools import lru_cache
import redis
import hashlib

class MultiTierCache:
    def __init__(self, redis_cluster_nodes: list[str]):
        # L1: In-process LRU cache (fastest path)
        self.l1_cache = {}
        self.l1_max_size = 10000
        
        # L2: Redis Cluster connection pool
        self.redis_cluster = redis.RedisCluster(
            host=redis_cluster_nodes[0],
            port=6379,
            decode_responses=True
        )
    
    def _get_redis_shard(self, key: str) -> int:
        """Determine which Redis shard holds this key."""
        return int(hashlib.md5(key.encode()).hexdigest(), 16) % 16384
    
    def get(self, short_key: str) -> str | None:
        # Check L1 first
        if short_key in self.l1_cache:
            return self.l1_cache[short_key]
        
        # Check L2 Redis Cluster
        long_url = self.redis_cluster.get(short_key)
        if long_url:
            # Promote to L1
            self._promote_to_l1(short_key, long_url)
            return long_url
        
        return None  # Cache miss — caller must check database
    
    def set(self, short_key: str, long_url: str, ttl: int = 3600):
        # Write through to L2
        self.redis_cluster.setex(short_key, ttl, long_url)
        # Also populate L1
        self._promote_to_l1(short_key, long_url)
    
    def _promote_to_l1(self, short_key: str, long_url: str):
        """Promote an entry to L1, evicting oldest if needed."""
        if len(self.l1_cache) >= self.l1_max_size:
            # Evict a random entry (simple approach; production would use LRU properly)
            oldest_key = next(iter(self.l1_cache))
            del self.l1_cache[oldest_key]
        self.l1_cache[short_key] = long_url
    
    def invalidate(self, short_key: str):
        """Invalidate a cached entry at all tiers."""
        self.l1_cache.pop(short_key, None)
        self.redis_cluster.delete(short_key)

Handling High Availability and Failover

Deployment Topology

Design for zero-downtime deployments and automatic failover:

# Example Kubernetes deployment configuration (partial)
# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: url-shortener-api
spec:
  replicas: 6
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 2
      maxUnavailable: 0
  template:
    spec:
      containers:
      - name: app
        image: url-shortener:latest
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secrets
              key: url
        - name: REDIS_CLUSTER_NODES
          value: "redis-0.redis-service,redis-1.redis-service,redis-2.redis-service"
        - name: MACHINE_ID
          valueFrom:
            fieldRef:
              fieldPath: metadata.name  # Unique per pod for Snowflake
        resources:
          requests:
            cpu: "2"
            memory: "4Gi"
          limits:
            cpu: "4"
            memory: "8Gi"
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5

Health Check and Graceful Degradation

@app.get("/health")
async def health_check():
    """Comprehensive health check for load balancers and monitoring."""
    health_status = {
        "status": "healthy",
        "timestamp": time.time(),
        "checks": {}
    }
    
    # Check database connectivity
    try:
        database.ping()
        health_status["checks"]["database"] = "ok"
    except Exception as e:
        health_status["checks"]["database"] = f"degraded: {str(e)}"
        health_status["status"] = "degraded"
    
    # Check Redis connectivity
    try:
        cache.redis_cluster.ping()
        health_status["checks"]["redis"] = "ok"
    except Exception as e:
        health_status["checks"]["redis"] = f"degraded: {str(e)}"
        health_status["status"] = "degraded"
    
    # Check Snowflake generator health
    try:
        test_id = generator.next_id()
        health_status["checks"]["id_generator"] = "ok"
    except Exception as e:
        health_status["checks"]["id_generator"] = f"failing: {str(e)}"
        health_status["status"] = "unhealthy"
    
    status_code = 200 if health_status["status"] == "healthy" else 503
    return JSONResponse(content=health_status, status_code=status_code)

Analytics Pipeline

Asynchronous Click Processing

For 100 million users, synchronous click logging on every redirect would slow responses. Use a message queue for decoupled processing:

import asyncio
from kafka import KafkaProducer
import json

class ClickEventProducer:
    def __init__(self, bootstrap_servers: list[str]):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            compression_type='gzip',
            batch_size=16384,
            linger_ms=10,  # Wait up to 10ms to batch messages
            acks=1  # Leader acknowledgment is sufficient for analytics
        )
    
    def send(self, short_key: str, user_agent: str, ip_address: str, referrer: str):
        """Fire-and-forget click event to Kafka."""
        event = {
            "short_key": short_key,
            "user_agent": user_agent[:512],  # Truncate for sanity
            "ip_address": ip_address,
            "referrer": referrer[:2048],
            "timestamp": int(time.time() * 1000)
        }
        # Asynchronous send — don't block the redirect response
        self.producer.send('click-events', value=event)

# Kafka consumer for batch processing
class ClickEventConsumer:
    def __init__(self, bootstrap_servers: list[str]):
        from kafka import KafkaConsumer
        self.consumer = KafkaConsumer(
            'click-events',
            bootstrap_servers=bootstrap_servers,
            group_id='click-processor-group',
            max_poll_records=500,
            enable_auto_commit=True
        )
    
    def process_batch(self):
        """Process a batch of click events and insert into database."""
        batch = []
        for message in self.consumer:
            event = json.loads(message.value)
            batch.append(event)
            
            if len(batch) >= 500:
                self._insert_batch(batch)
                batch = []
    
    def _insert_batch(self, events: list[dict]):
        """Batch insert click events into the database."""
        import psycopg2
        conn = psycopg2.connect(os.environ['DATABASE_URL'])
        cursor = conn.cursor()
        
        # Use COPY protocol for maximum throughput
        from io import StringIO
        buffer = StringIO()
        for event in events:
            buffer.write(
                f"{event['short_key']}\t"
                f"{event['timestamp']}\t"
                f"{event['user_agent']}\t"
                f"{event['ip_address']}\t"
                f"{event['referrer']}\n"
            )
        buffer.seek(0)
        
        cursor.copy_from(
            buffer,
            'click_events',
            columns=('short_key', 'timestamp', 'user_agent', 'ip_address', 'referrer'),
            sep='\t'
        )
        conn.commit()
        cursor.close()
        conn.close()

Security Considerations

Preventing Malicious URLs

Short URLs are inherently opaque, making them attractive for phishing. Implement multiple layers of protection:

import re
import requests
from urllib.parse import urlparse

class URLValidator:
    # Known phishing and malware domains (would be much larger in production)
    BLOCKED_DOMAINS = {
        'phish.example.com', 'malware.example.net'
    }
    
    # Patterns commonly found in phishing URLs
    SUSPICIOUS_PATTERNS = [
        r'\.php\?redirect=',
        r'open-redirect',
        r'url=http',
    ]
    
    @classmethod
    def validate(cls, url: str) -> tuple[bool, str | None]:
        """Validate a URL before shortening. Returns (is_valid, reason)."""
        parsed = urlparse(url)
        
        # Must be HTTP or HTTPS
        if parsed.scheme not in ('http', 'https'):
            return False, "Only HTTP and HTTPS URLs are allowed"
        
        # Check against blocked domains
        if parsed.netloc.lower() in cls.BLOCKED_DOMAINS:
            return False, "This domain is blocked"
        
        # Check for suspicious patterns
        for pattern in cls.SUSPICIOUS_PATTERNS:
            if re.search(pattern, url, re.IGNORECASE):
                return False, "URL contains suspicious patterns"
        
        # Optional: Check against Safe Browsing API
        # safe_browsing_result = check_safe_browsing(url)
        # if not safe_browsing_result['safe']:
        #     return False, "URL flagged by Safe Browsing"
        
        return True, None
    
    @classmethod
    def normalize_url(cls, url: str) -> str:
        """Normalize URL to prevent duplicate encodings."""
        parsed = urlparse(url)
        # Lowercase scheme and netloc
        normalized = f"{parsed.scheme.lower()}://{parsed.netloc.lower()}"
        # Remove default ports
        if (parsed.scheme == 'https' and ':443' in parsed.netloc) or \
           (parsed.scheme == 'http' and ':80' in parsed.netloc):
            normalized = normalized.replace(':443', '').replace(':80', '')
        # Add path, preserving case
        normalized += parsed.path or '/'
        if parsed.query:
            normalized += f"?{parsed.query}"
        return normalized

Abuse Prevention

Implement exponential backoff for suspicious request patterns and CAPTCHA challenges for anomalous traffic:

class AbuseDetectionMiddleware:
    def __init__(self):
        self.suspicious_ips: dict[str, list[float]] = defaultdict(list)
        self.blocked_ips: dict[str, float] = {}
        self.window_size = 60  # 60 seconds
    
    def record_request(self, ip: str) -> bool:
        """Record a request. Returns True if request should be allowed."""
        now = time.time()
        
        # Check if IP is currently blocked
        if ip in self.blocked_ips:
            if now < self.blocked_ips[ip]:
                return False  # Still blocked
            else:
                del self.blocked_ips[ip]  # Block expired
        
        # Clean old records
        self.suspicious_ips[ip] = [
            t for t in self.suspicious_ips[ip]
            if now - t < self.window_size
        ]
        
        self.suspicious_ips[ip].append(now)
        
        # If too many requests in window, block the IP
        if len(self.suspicious_ips[ip]) > 200:  # 200 requests per minute
            self.blocked_ips[ip] = now + 300  # Block for 5 minutes
            return False
        
        return True

Best Practices Summary

Conclusion

Designing a URL shortener for 100 million users requires thinking far beyond a simple hash function and database lookup. The system must handle millions of writes and redirects daily while maintaining sub-50ms latency, surviving infrastructure failures gracefully, and protecting users from malicious links. The architecture outlined here — combining Snowflake-style ID generation, Base62 encoding, PostgreSQL partitioning, multi-tier Redis caching, Kafka-based analytics, and comprehensive security validation — provides a battle-tested blueprint. Start with the core redirect path and iteratively add caching layers, analytics, and security features as traffic grows. Measure everything, optimize bottlenecks as they appear, and always design with the assumption that your most popular URL will receive thousands of simultaneous clicks. Build for that peak, and the system will serve 100 million users with confidence.

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles