/database-patterns

name: database-patterns
description: Use when designing PostgreSQL + Redis data models, indexes, caching strategies, JSONB usage, tiered storage, or cache consistency contracts.

Database Patterns

Core Principles

• **PostgreSQL Primary** — Relational data, transactions, complex queries
• **Redis Secondary** — Caching, sessions, real-time data
• **Index-First Design** — Design queries before indexes
• **JSONB Sparingly** — Structured data prefers columns
• **Cache-Aside Default** — Read-through, write-around
• **Tiered Storage** — Hot/Warm/Cold data separation
• **No backwards compatibility** — Migrate data, don't keep legacy schemas

---

PostgreSQL

Data Type Selection

| Use Case | Type | Avoid | |----------|------|-------| | Primary Key | `UUID` / `BIGSERIAL` | `INT` (range limits) | | Timestamps | `TIMESTAMPTZ` | `TIMESTAMP` (no timezone) | | Money | `NUMERIC(19,4)` | `FLOAT` (precision loss) | | Status | `TEXT` + CHECK | `INT` (unreadable) | | Semi-structured | `JSONB` | `JSON` (no indexing) | | Full-text | `TSVECTOR` | `LIKE '%..%'` |

Schema Design

-- Use UUID for distributed-friendly IDs
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

CREATE TABLE users (
  id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  email TEXT UNIQUE NOT NULL,
  name TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'active'
    CHECK (status IN ('active', 'inactive', 'suspended')),
  metadata JSONB DEFAULT '{}',
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Updated timestamp trigger
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
  NEW.updated_at = NOW();
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER users_updated_at
  BEFORE UPDATE ON users
  FOR EACH ROW
  EXECUTE FUNCTION update_updated_at();

Indexing Strategy

-- B-Tree: Equality, range, sorting (default)
CREATE INDEX idx_users_email ON users(email);

-- Composite: Leftmost prefix rule
-- Supports: (user_id), (user_id, created_at)
-- Does NOT support: (created_at) alone
CREATE INDEX idx_orders_user_date ON orders(user_id, created_at DESC);

-- Partial: Reduce index size
CREATE INDEX idx_active_users ON users(email)
  WHERE status = 'active';

-- GIN for JSONB: Containment queries
CREATE INDEX idx_metadata ON users USING GIN (metadata jsonb_path_ops);

-- Expression: Specific JSONB field
CREATE INDEX idx_user_role ON users ((metadata->>'role'));

-- Full-text search
CREATE INDEX idx_search ON products USING GIN (to_tsvector('english', name || ' ' || description));

JSONB Usage

-- Good: Dynamic attributes, rarely queried fields
CREATE TABLE products (
  id UUID PRIMARY KEY,
  name TEXT NOT NULL,
  price NUMERIC(19,4) NOT NULL,
  category TEXT NOT NULL,           -- Extracted: frequently queried
  attributes JSONB DEFAULT '{}'     -- Dynamic: color, size, specs
);

-- Query with containment
SELECT * FROM products
WHERE category = 'electronics'              -- B-Tree index
  AND attributes @> '{"brand": "Apple"}';   -- GIN index

-- Query specific field
SELECT * FROM products
WHERE attributes->>'color' = 'black';       -- Expression index

-- Update JSONB field
UPDATE products
SET attributes = attributes || '{"featured": true}'
WHERE id = '...';

Query Optimization

-- Always use EXPLAIN ANALYZE
EXPLAIN ANALYZE
SELECT u.*, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON o.user_id = u.id
WHERE u.status = 'active'
GROUP BY u.id
ORDER BY u.created_at DESC
LIMIT 20;

-- Watch for:
-- ❌ Seq Scan on large tables → Add index
-- ❌ Sort → Use index for ordering
-- ❌ Nested Loop with many rows → Consider JOIN order
-- ❌ Hash Join on huge tables → Add indexes

Connection Pooling

// PgBouncer or built-in pool
import { Pool } from 'pg';

const pool = new Pool({
  max: 20,                      // Max connections
  idleTimeoutMillis: 30000,     // Close idle connections
  connectionTimeoutMillis: 2000, // Fail fast
});

// Connection count formula:
// connections = (cores * 2) + effective_spindle_count
// Usually 10-30 is enough

---

Redis

Data Structure Selection

| Use Case | Structure | Example | |----------|-----------|---------| | Cache objects | String | `user:123` → JSON | | Counters | String + INCR | `views:article:456` | | Sessions | Hash | `session:abc` → {userId, ...} | | Leaderboards | Sorted Set | `scores` → {userId: score} | | Queues | List/Stream | `tasks` → LPUSH/RPOP | | Unique sets | Set | `online_users` | | Real-time | Pub/Sub/Stream | Notifications |

Key Naming

# Format: <entity>:<id>:<attribute>
user:123:profile
user:123:settings
order:456:items
session:abc123

# Use colons for hierarchy
# Enables pattern matching with SCAN
SCAN 0 MATCH "user:*:profile" COUNT 100

TTL Strategy

const TTL = {
  SESSION: 24 * 60 * 60,      // 24 hours
  CACHE: 15 * 60,             // 15 minutes
  RATE_LIMIT: 60,             // 1 minute
  LOCK: 30,                   // 30 seconds
};

// Set with TTL
await redis.set(`cache:user:${id}`, JSON.stringify(user), 'EX', TTL.CACHE);

// Check TTL
const remaining = await redis.ttl(`cache:user:${id}`);

---

Caching Patterns

Cache-Aside (Lazy Loading)

async function getUser(id: string): Promise<User> {
  const cacheKey = `user:${id}`;

  // 1. Check cache
  const cached = await redis.get(cacheKey);
  if (cached) {
    return JSON.parse(cached);
  }

  // 2. Cache miss → Query database
  const user = await db.user.findUnique({ where: { id } });
  if (!user) {
    throw new NotFoundError('User not found');
  }

  // 3. Populate cache
  await redis.set(cacheKey, JSON.stringify(user), 'EX', 900);

  return user;
}

Write-Through

async function updateUser(id: string, data: UpdateInput): Promise<User> {
  // 1. Update database
  const user = await db.user.update({
    where: { id },
    data,
  });

  // 2. Update cache immediately
  await redis.set(`user:${i