Preguntas de entrevista de Redis — 35 con código y respuestas reales

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# String — contador de vistas
r.incr('post:42:views')

# Hash — perfil de usuario
r.hset('user:1001', mapping={
    'name': 'Martín',
    'plan': 'pro',
    'interviews_done': 3
})
print(r.hget('user:1001', 'plan'))  # 'pro'

# List — cola FIFO
r.rpush('jobs:queue', 'job:a', 'job:b')
job = r.blpop('jobs:queue', timeout=5)

# Set — seguidores
r.sadd('user:1001:following', 'user:2', 'user:3')
r.sadd('user:2:following', 'user:1001', 'user:4')

# Usuarios que los dos siguen
common = r.sinter('user:1001:following', 'user:2:following')

# Sorted Set — leaderboard
r.zadd('ranking:global', {'user:1001': 9850, 'user:2': 7200})
top3 = r.zrevrange('ranking:global', 0, 2, withscores=True)

# MAL: actualizar un campo en JSON requiere leer-parsear-escribir
raw = r.get('user:1001')
user = json.loads(raw)
user['interviews_done'] += 1
r.set('user:1001', json.dumps(user))

# BIEN: con Hash es atómico
r.hincrby('user:1001', 'interviews_done', 1)

import time

# Feed cronológico de un usuario — Sorted Set con timestamp como score
def add_to_feed(user_id: str, activity_id: str):
    key = f'feed:{user_id}'
    score = time.time()
    r.zadd(key, {activity_id: score})
    # Mantener sólo los últimos 200 items
    r.zremrangebyrank(key, 0, -201)

def get_feed(user_id: str, page: int = 0, per_page: int = 20):
    key = f'feed:{user_id}'
    start = page * per_page
    end = start + per_page - 1
    return r.zrevrange(key, start, end, withscores=True)

# Productor
r.xadd('events:interviews', {
    'user_id': '1001',
    'event': 'session_started',
    'company': 'Mercado Libre',
    'ts': str(time.time())
})

# Consumidor con grupo
r.xgroup_create('events:interviews', 'analytics-group', id='0', mkstream=True)

while True:
    messages = r.xreadgroup(
        groupname='analytics-group',
        consumername='worker-1',
        streams={'events:interviews': '>'},
        count=10,
        block=2000  # ms
    )
    for stream, entries in (messages or []):
        for entry_id, fields in entries:
            process(fields)
            r.xack('events:interviews', 'analytics-group', entry_id)

# redis.conf — configuración recomendada para producción
save 900 1       # RDB: si hubo 1 cambio en 900s
save 300 10      # RDB: si hubo 10 cambios en 300s
save 60 10000    # RDB: si hubo 10000 cambios en 60s

appendonly yes
appendfsync everysec
auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

# Cómo validar el estado de persistencia desde código
info = r.info('persistence')
print(f"RDB last save: {info['rdb_last_save_time']}")
print(f"AOF enabled: {info['aof_enabled']}")
print(f"AOF pending rewrite: {info['aof_rewrite_scheduled']}")

# Pub/Sub — notificaciones en tiempo real
import threading

def subscriber():
    sub = redis.Redis(decode_responses=True)
    p = sub.pubsub()
    p.subscribe('notifications:user:1001')
    for message in p.listen():
        if message['type'] == 'message':
            print(f"Notificación: {message['data']}")

# En otro thread/proceso
thread = threading.Thread(target=subscriber, daemon=True)
thread.start()

r.publish('notifications:user:1001', 'Nueva oferta de trabajo disponible')

import threading
from functools import lru_cache

local_cache = {}

def setup_cache_invalidation_listener(redis_client):
    """Listener que corre en background thread"""
    def listen():
        sub = redis_client.pubsub()
        sub.subscribe('cache:invalidations')
        for message in sub.listen():
            if message['type'] == 'message':
                key = message['data']
                local_cache.pop(key, None)
                print(f"Cache local invalidado: {key}")
    
    thread = threading.Thread(target=listen, daemon=True)
    thread.start()

def get_user(user_id: str) -> dict:
    cache_key = f'user:{user_id}'
    
    # L1: caché local en memoria
    if cache_key in local_cache:
        return local_cache[cache_key]
    
    # L2: Redis
    raw = r.get(cache_key)
    if raw:
        user = json.loads(raw)
        local_cache[cache_key] = user
        return user
    
    # L3: base de datos
    user = db.fetch_user(user_id)
    r.setex(cache_key, 3600, json.dumps(user))
    local_cache[cache_key] = user
    return user

def update_user(user_id: str, data: dict):
    db.update_user(user_id, data)
    cache_key = f'user:{user_id}'
    r.delete(cache_key)
    # Notificar a todas las instancias de la app
    r.publish('cache:invalidations', cache_key)

from redis.sentinel import Sentinel

sentinel = Sentinel(
    [('sentinel-1', 26379), ('sentinel-2', 26379), ('sentinel-3', 26379)],
    socket_timeout=0.1
)

# El cliente se conecta siempre al master actual
master = sentinel.master_for('mymaster', socket_timeout=0.1, decode_responses=True)
# Para lecturas puede usar réplicas
slave = sentinel.slave_for('mymaster', socket_timeout=0.1, decode_responses=True)

master.set('key', 'value')
print(slave.get('key'))

Nodo A: slots 0-5460
Nodo B: slots 5461-10922
Nodo C: slots 10923-16383

from redis.cluster import RedisCluster

# El cliente maneja el routing automáticamente
rc = RedisCluster(
    host='node-1',
    port=7000,
    decode_responses=True
)

rc.set('user:1001', 'data')  # va al nodo que corresponde según el hash slot

# Hash tags: forzar que múltiples keys estén en el mismo slot
rc.set('{user:1001}.profile', 'data')
rc.set('{user:1001}.sessions', 'data')
# Ambas van al slot de "user:1001" — necesario para operaciones multi-key

# Monitorear el estado del cluster
redis-cli --cluster check node-1:7000

# Ver la distribución de slots
redis-cli -h node-1 -p 7000 CLUSTER NODES

def get_product(product_id: str) -> dict:
    key = f'product:{product_id}'
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    product = db.get_product(product_id)
    r.setex(key, 3600, json.dumps(product))
    return product

def update_product(product_id: str, data: dict):
    db.update_product(product_id, data)
    key = f'product:{product_id}'
    r.setex(key, 3600, json.dumps(data))  # siempre fresco

def update_product_fast(product_id: str, data: dict):
    key = f'product:{product_id}'
    r.setex(key, 3600, json.dumps(data))
    # Encolar para persistencia asíncrona
    r.rpush('db:writes:queue', json.dumps({'table': 'products', 'id': product_id, 'data': data}))

import uuid

def get_with_lock(key: str, regenerate_fn, ttl: int = 3600):
    value = r.get(key)
    if value:
        return json.loads(value)
    
    lock_key = f'lock:{key}'
    lock_value = str(uuid.uuid4())
    
    # Intentar tomar el lock (NX = solo si no existe)
    acquired = r.set(lock_key, lock_value, nx=True, ex=10)
    
    if acquired:
        try:
            data = regenerate_fn()
            r.setex(key, ttl, json.dumps(data))
            return data
        finally:
            # Liberar el lock sólo si somos nosotros quien lo tiene
            lua_script = """
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
            """
            r.eval(lua_script, 1, lock_key, lock_value)
    else:
        # Otro worker está regenerando, esperar y reintentar
        import time
        time.sleep(0.1)
        return get_with_lock(key, regenerate_fn, ttl)

import math
import random
import time

def get_with_xfetch(key: str, regenerate_fn, ttl: int = 3600, beta: float = 1.0):
    """
    Probabilidad de revalidar aumenta a medida que se acerca la expiración.
    No necesita locks.
    """
    data = r.get(f'data:{key}')
    expiry = r.get(f'expiry:{key}')
    
    if data and expiry:
        remaining = float(expiry) - time.time()
        recompute_time = r.get(f'recompute_time:{key}') or '1'
        
        # Decidir probabilísticamente si recomputar anticipadamente
        if remaining - beta * float(recompute_time) * math.log(random.random()) > 0:
            return json.loads(data)
    
    start = time.time()
    new_data = regenerate_fn()
    recompute_time = time.time() - start
    
    r.setex(f'data:{key}', ttl, json.dumps(new_data))
    r.set(f'expiry:{key}', time.time() + ttl)
    r.set(f'recompute_time:{key}', recompute_time)
    
    return new_data

def get_user_safe(user_id: str):
    key = f'user:{user_id}'
    cached = r.get(key)
    
    if cached is not None:
        if cached == 'NULL':
            return None  # sabemos que no existe
        return json.loads(cached)
    
    user = db.get_user(user_id)
    if user:
        r.setex(key, 3600, json.dumps(user))
    else:
        r.setex(key, 300, 'NULL')  # TTL corto para nulos
    
    return user

# Con RedisBloom (módulo) o implementación propia
# Antes de consultar DB, verificar si el ID podría existir

from pybloom_live import BloomFilter

# Inicializar con todos los IDs existentes (en startup o batch)
bloom = BloomFilter(capacity=1000000, error_rate=0.01)
for user_id in db.get_all_user_ids():
    bloom.add(user_id)

def get_user_with_bloom(user_id: str):
    if user_id not in bloom:
        return None  # definitivamente no existe — sin DB query
    
    key = f'user:{user_id}'
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    user = db.get_user(user_id)
    if user:
        r.setex(key, 3600, json.dumps(user))
        bloom.add(user_id)
    return user

def is_allowed_fixed_window(user_id: str, limit: int = 100, window_seconds: int = 60) -> bool:
    key = f'ratelimit:{user_id}:{int(time.time()) // window_seconds}'
    current = r.incr(key)
    if current == 1:
        r.expire(key, window_seconds * 2)
    return current <= limit

def is_allowed_sliding_window(user_id: str, limit: int = 100, window_ms: int = 60000) -> bool:
    key = f'ratelimit:log:{user_id}'
    now = int(time.time() * 1000)
    window_start = now - window_ms
    
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)   # eliminar requests viejos
    pipe.zadd(key, {str(now) + str(uuid.uuid4()): now})  # agregar el request actual
    pipe.zcard(key)                                # contar requests en la ventana
    pipe.expire(key, window_ms // 1000 + 1)
    results = pipe.execute()
    
    return results[2] <= limit

-- token_bucket.lua
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])      -- tokens por segundo
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

local last_time = tonumber(redis.call('hget', key, 'last_time') or now)
local tokens = tonumber(redis.call('hget', key, 'tokens') or capacity)

-- Calcular tokens generados desde la última vez
local elapsed = now - last_time
local new_tokens = math.min(capacity, tokens + elapsed * rate)

if new_tokens >= requested then
    redis.call('hset', key, 'tokens', new_tokens - requested, 'last_time', now)
    redis.call('expire', key, math.ceil(capacity / rate) + 1)
    return 1  -- permitido
else
    redis.call('hset', key, 'tokens', new_tokens, 'last_time', now)
    return 0  -- rechazado
end

with open('token_bucket.lua', 'r') as f:
    script = f.read()

token_bucket = r.register_script(script)

def is_allowed_token_bucket(user_id: str, capacity: int = 100, rate: float = 10) -> bool:
    result = token_bucket(
        keys=[f'bucket:{user_id}'],
        args=[capacity, rate, time.time(), 1]
    )
    return bool(result)

def check_rate_limits(user_id: str, ip: str, endpoint: str) -> tuple[bool, str]:
    """
    Retorna (is_allowed, reason)
    """
    limits = [
        # (key, limit, window)
        (f'rl:ip:{ip}', 1000, 60),           # 1000 req/min por IP
        (f'rl:user:{user_id}', 100, 60),      # 100 req/min por usuario
        (f'rl:user:{user_id}:{endpoint}', 10, 60),  # 10 req/min por endpoint
    ]
    
    pipe = r.pipeline()
    now = int(time.time())
    
    for key, limit, window in limits:
        window_key = f'{key}:{now // window}'
        pipe.incr(window_key)
        pipe.expire(window_key, window * 2)
    
    results = pipe.execute()
    counts = results[::2]  # cada segundo resultado es del expire, ignoramos
    
    for i, (key, limit, window) in enumerate(limits):
        if counts[i] > limit:
            return False, f'Rate limit exceeded: {key}'
    
    return True, 'ok'

import uuid
import time

class RedisLock:
    def __init__(self, redis_client, key: str, timeout: int = 10):
        self.r = redis_client
        self.key = f'lock:{key}'
        self.timeout = timeout
        self.token = str(uuid.uuid4())
        
    def acquire(self, blocking: bool = True, blocking_timeout: float = 5.0) -> bool:
        deadline = time.time() + blocking_timeout
        while True:
            acquired = self.r.set(
                self.key,
                self.token,
                nx=True,   # solo si no existe
                ex=self.timeout
            )
            if acquired:
                return True
            if not blocking or time.time() >= deadline:
                return False
            time.sleep(0.01)
    
    def release(self):
        """Libera el lock SOLO si somos el dueño — usa Lua para atomicidad"""
        lua = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        return self.r.eval(lua, 1, self.key, self.token)
    
    def __enter__(self):
        if not self.acquire():
            raise TimeoutError(f'No se pudo adquirir el lock {self.key}')
        return self
    
    def __exit__(self, *args):
        self.release()

# Uso
with RedisLock(r, 'invoice:42:process', timeout=30) as lock:
    process_invoice(42)

# Con la librería redlock-py
from redlock import Redlock

dlm = Redlock([
    {"host": "redis-1", "port": 6379},
    {"host": "redis-2", "port": 6379},
    {"host": "redis-3", "port": 6379},
])

lock = dlm.lock("my-resource", 10000)  # TTL en ms
if lock:
    try:
        # sección crítica
        do_work()
    finally:
        dlm.unlock(lock)

import time

# Sin pipeline: N round-trips
start = time.time()
for i in range(1000):
    r.set(f'key:{i}', f'value:{i}')
print(f"Sin pipeline: {time.time() - start:.3f}s")

# Con pipeline: 1 round-trip
start = time.time()
pipe = r.pipeline(transaction=False)
for i in range(1000):
    pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()
print(f"Con pipeline: {time.time() - start:.3f}s")
# Típicamente 10-50x más rápido en red con latencia

# Pipeline en Node.js
const pipeline = redis.pipeline();
for (let i = 0; i < 1000; i++) {
  pipeline.set(`key:${i}`, `value:${i}`);
}
await pipeline.exec();

# MULTI/EXEC: todos los comandos entre MULTI y EXEC se ejecutan atómicamente
pipe = r.pipeline(transaction=True)  # transaction=True habilita MULTI/EXEC

pipe.multi()
pipe.incr('counter:views')
pipe.incr('counter:unique_users')
pipe.expire('counter:views', 3600)
results = pipe.execute()

# En Node.js
const results = await redis
  .multi()
  .incr('counter:views')
  .incr('counter:unique_users')
  .expire('counter:views', 3600)
  .exec();

def transfer_points(from_user: str, to_user: str, amount: int, retries: int = 3):
    """
    Transferir puntos entre usuarios de forma segura con optimistic locking
    """
    from_key = f'points:{from_user}'
    to_key = f'points:{to_user}'
    
    for attempt in range(retries):
        with r.pipeline() as pipe:
            try:
                # WATCH: si alguna de estas keys cambia antes del EXEC, la transacción falla
                pipe.watch(from_key, to_key)
                
                from_points = int(pipe.get(from_key) or 0)
                if from_points < amount:
                    raise ValueError("Puntos insuficientes")
                
                # Iniciar la transacción
                pipe.multi()
                pipe.decrby(from_key, amount)
                pipe.incrby(to_key, amount)
                pipe.execute()
                
                return True  # éxito
                
            except redis.WatchError:
                # Alguien modificó las keys mientras calculábamos — reintentar
                if attempt == retries - 1:
                    raise
                time.sleep(0.01 * (2 ** attempt))  # backoff exponencial
    
    return False

# Incrementar un contador SOLO si no supera un límite — no es posible con MULTI/EXEC
increment_if_under = r.register_script("""
local current = redis.call('GET', KEYS[1])
local count = tonumber(current) or 0
local limit = tonumber(ARGV[1])

if count < limit then
    redis.call('INCR', KEYS[1])
    redis.call('EXPIRE', KEYS[1], ARGV[2])
    return count + 1
else
    return -1
end
""")

result = increment_if_under(keys=['api:calls:user:1001'], args=[100, 3600])
if result == -1:
    raise Exception("Rate limit exceeded")

# Otro ejemplo: pop atómico de una lista con condición
conditional_pop = r.register_script("""
local item = redis.call('LINDEX', KEYS[1], 0)
if item and cjson.decode(item)['priority'] == ARGV[1] then
    return redis.call('LPOP', KEYS[1])
end
return nil
""")

# Comandos clave para diagnóstico
redis-cli INFO memory
redis-cli MEMORY USAGE user:1001     # bytes que usa una key específica
redis-cli MEMORY DOCTOR               # análisis automático con recomendaciones
redis-cli --bigkeys                   # encontrar las keys más grandes
redis-cli --hotkeys                   # keys con más accesos (requiere maxmemory-policy LFU)

# Análisis programático
info = r.info('memory')
print(f"Memoria usada: {info['used_memory_human']}")
print(f"Memoria RSS: {info['used_memory_rss_human']}")
print(f"Fragmentación: {info['mem_fragmentation_ratio']:.2f}")
print(f"Peak: {info['used_memory_peak_human']}")

# Fragmentación > 1.5 indica mucha fragmentación — considerar restart o jemalloc
# Fragmentación < 1.0 indica que Redis está usando swap — alerta crítica

import zlib
import json

def set_compressed(key: str, data: dict, ttl: int = 3600):
    serialized = json.dumps(data).encode()
    if len(serialized) > 1024:  # comprimir si > 1KB
        compressed = zlib.compress(serialized, level=6)
        r.setex(f'c:{key}', ttl, compressed)
    else:
        r.setex(key, ttl, serialized)

def get_compressed(key: str) -> dict | None:
    # Intentar key comprimida primero
    data = r.get(f'c:{key}') or r.get(key)
    if not data:
        return None
    try:
        return json.loads(zlib.decompress(data))
    except zlib.error:
        return json.loads(data)

# En redis.conf
maxmemory 2gb
maxmemory-policy allkeys-lru

# Ver el encoding interno de una key
redis-cli OBJECT ENCODING user:1001
# Puede retornar: ziplist, listpack, hashtable, skiplist, intset, embstr, raw, etc.

# Configurar los umbrales en redis.conf
hash-max-listpack-entries 128
hash-max-listpack-value 64
zset-max-listpack-entries 128
zset-max-listpack-value 64
set-max-intset-entries 512

# MAL: KEYS bloquea el servidor mientras escanea TODAS las keys
all_keys = r.keys('user:*')  # puede ser millones de keys

# BIEN: SCAN es incremental y no bloquea
def scan_keys(pattern: str, count: int = 100):
    cursor = 0
    while True:
        cursor, keys = r.scan(cursor, match=pattern, count=count)
        for key in keys:
            yield key
        if cursor == 0:
            break

for key in scan_keys('user:*'):
    process_key(key)

# Del mismo modo: SSCAN, HSCAN, ZSCAN para estructuras grandes
cursor = 0
while True:
    cursor, members = r.sscan('large:set', cursor, count=100)
    for member in members:
        process(member)
    if cursor == 0:
        break

import secrets
import json
from datetime import timedelta

SESSION_TTL = int(timedelta(days=7).total_seconds())

def create_session(user_id: str, metadata: dict = None) -> str:
    session_id = secrets.token_urlsafe(32)
    session_data = {
        'user_id': user_id,
        'created_at': time.time(),
        'last_active': time.time(),
        **(metadata or {})
    }
    r.setex(f'session:{session_id}', SESSION_TTL, json.dumps(session_data))
    # Mantener lista de sesiones por usuario (para invalidación total)
    r.sadd(f'user_sessions:{user_id}', session_id)
    r.expire(f'user_sessions:{user_id}', SESSION_TTL)
    return session_id

def get_session(session_id: str) -> dict | None:
    raw = r.get(f'session:{session_id}')
    if not raw:
        return None
    session = json.loads(raw)
    # Sliding expiration: resetear el TTL en cada acceso
    r.expire(f'session:{session_id}', SESSION_TTL)
    session['last_active'] = time.time()
    r.setex(f'session:{session_id}', SESSION_TTL, json.dumps(session))
    return session

def invalidate_all_sessions(user_id: str):
    """Cerrar todas las sesiones del usuario (ej: cambio de contraseña)"""
    session_ids = r.smembers(f'user_sessions:{user_id}')
    if session_ids:
        pipe = r.pipeline()
        for sid in session_ids:
            pipe.delete(f'session:{sid}')
        pipe.delete(f'user_sessions:{user_id}')
        pipe.execute()

# Leaderboard global con Sorted Set
def update_score(user_id: str, delta: int):
    new_score = r.zincrby('leaderboard:global', delta, user_id)
    return new_score

def get_rank(user_id: str) -> int:
    rank = r.zrevrank('leaderboard:global', user_id)
    return rank + 1 if rank is not None else None

def get_top_n(n: int = 10) -> list:
    return r.zrevrange('leaderboard:global', 0, n - 1, withscores=True)

def get_around_user(user_id: str, range_size: int = 5) -> list:
    """Top 5 por encima y 5 por debajo del usuario"""
    rank = r.zrevrank('leaderboard:global', user_id)
    if rank is None:
        return []
    start = max(0, rank - range_size)
    end = rank + range_size
    return r.zrevrange('leaderboard:global', start, end, withscores=True)

# Para escala masiva: leaderboards particionados por período
def get_current_week_key() -> str:
    week = time.strftime('%Y-W%W')
    return f'leaderboard:weekly:{week}'

def update_weekly_score(user_id: str, delta: int):
    key = get_current_week_key()
    r.zincrby(key, delta, user_id)
    r.expire(key, 60 * 60 * 24 * 14)  # 2 semanas de retención

# Técnica: usar scores 0 para todos y ordenar lexicográficamente
# ZRANGEBYLEX funciona sobre el member cuando todos tienen el mismo score

def add_autocomplete_term(term: str, namespace: str = 'search'):
    # Agregar el término y todos sus prefijos
    r.zadd(f'autocomplete:{namespace}', {term: 0})

def autocomplete(prefix: str, namespace: str = 'search', limit: int = 10) -> list:
    return r.zrangebylex(
        f'autocomplete:{namespace}',
        f'[{prefix}',           # desde el prefijo (inclusive)
        f'[{prefix}\xff',       # hasta el prefijo + \xff (máximo carácter)
        start=0,
        num=limit
    )

# Poblar
for term in ['python', 'pytorch', 'pandas', 'postgresql', 'pubsub']:
    add_autocomplete_term(term)

print(autocomplete('py'))  # ['pandas', 'postgresql', 'pubsub', 'python', 'pytorch']
print(autocomplete('pyt')) # ['python', 'pytorch']

def add_term_with_popularity(term: str, score: float):
    r.zadd('autocomplete:popular', {term: score})

def popular_autocomplete(prefix: str, limit: int = 10) -> list:
    # Necesita un approach diferente: buscar por score descendente
    # filtrado por prefijo en aplicación
    candidates = r.zrevrange('autocomplete:popular', 0, 1000, withscores=True)
    return [
        (term, score) for term, score in candidates
        if term.startswith(prefix)
    ][:limit]

import json
import uuid
import time

class SimpleJobQueue:
    def __init__(self, redis_client, queue_name: str):
        self.r = redis_client
        self.queue = f'jobs:{queue_name}'
        self.processing = f'jobs:{queue_name}:processing'
        self.failed = f'jobs:{queue_name}:failed'
    
    def enqueue(self, payload: dict, delay: float = 0) -> str:
        job_id = str(uuid.uuid4())
        job = {
            'id': job_id,
            'payload': payload,
            'enqueued_at': time.time(),
            'attempts': 0
        }
        
        if delay > 0:
            # Cola retrasada con Sorted Set (score = timestamp de ejecución)
            execute_at = time.time() + delay
            self.r.zadd(f'{self.queue}:delayed', {json.dumps(job): execute_at})
        else:
            self.r.rpush(self.queue, json.dumps(job))
        
        return job_id
    
    def dequeue(self, timeout: int = 30) -> dict | None:
        # Primero mover delayed jobs que ya deben ejecutarse
        self._move_delayed_jobs()
        
        # BRPOPLPUSH: atómico, mueve el job a la lista de processing
        raw = self.r.blmove(self.queue, self.processing, timeout=1, src='LEFT', dest='RIGHT')
        if raw:
            return json.loads(raw)
        return None
    
    def _move_delayed_jobs(self):
        now = time.time()
        jobs = self.r.zrangebyscore(f'{self.queue}:delayed', 0, now)
        if jobs:
            pipe = self.r.pipeline()
            for job in jobs:
                pipe.rpush(self.queue, job)
                pipe.zrem(f'{self.queue}:delayed', job)
            pipe.execute()
    
    def ack(self, job: dict):
        self.r.lrem(self.processing, 1, json.dumps(job))
    
    def nack(self, job: dict, retry: bool = True):
        job['attempts'] += 1
        self.r.lrem(self.processing, 1, json.dumps(job))
        
        if retry and job['attempts'] < 3:
            delay = 60 * (2 ** job['attempts'])  # backoff exponencial
            self.r.zadd(f'{self.queue}:delayed', {json.dumps(job): time.time() + delay})
        else:
            self.r.rpush(self.failed, json.dumps(job))

# Detectar hot keys (requiere Redis 4+ con LFU)
redis-cli --hotkeys -h node-1 -p 7000

# Monitorear comandos en tiempo real
redis-cli MONITOR | head -100

# Ver estadísticas de frecuencia
redis-cli OBJECT FREQ mykey  # sólo con maxmemory-policy lfu

from functools import lru_cache
from datetime import datetime, timedelta

_local_cache = {}
_local_cache_expiry = {}

def get_hot_config(key: str) -> str:
    """Configuración global accedida miles de veces por segundo"""
    now = datetime.now()
    if key in _local_cache and _local_cache_expiry[key] > now:
        return _local_cache[key]
    
    value = r.get(key)
    _local_cache[key] = value
    _local_cache_expiry[key] = now + timedelta(seconds=5)
    return value

import random

def get_sharded(base_key: str, shards: int = 10) -> str:
    shard = random.randint(0, shards - 1)
    return r.get(f'{base_key}:shard:{shard}')

def set_sharded(base_key: str, value: str, shards: int = 10, ttl: int = 3600):
    pipe = r.pipeline()
    for i in range(shards):
        pipe.setex(f'{base_key}:shard:{i}', ttl, value)
    pipe.execute()

from datetime import date

def track_daily_unique(user_id: int, page: str):
    today = date.today().isoformat()
    key = f'unique:{page}:{today}'
    r.setbit(key, user_id, 1)          # O(1), usa 1 bit por user_id
    r.expire(key, 86400 * 7)           # guardar 7 días

def get_daily_unique_count(page: str, day: str = None) -> int:
    day = day or date.today().isoformat()
    return r.bitcount(f'unique:{page}:{day}')

def was_user_active_today(user_id: int, page: str) -> bool:
    today = date.today().isoformat()
    return bool(r.getbit(f'unique:{page}:{today}', user_id))

# Usuarios activos en los últimos 3 días (BITOP AND)
def active_last_3_days(page: str) -> int:
    keys = [f'unique:{page}:{date.today().isoformat()}']  # simplificado
    r.bitop('AND', 'temp:active3days', *keys)
    count = r.bitcount('temp:active3days')
    r.delete('temp:active3days')
    return count

# Para user_ids muy grandes o cuando no controlás el ID, usá HyperLogLog
def track_hll(ip: str, page: str):
    today = date.today().isoformat()
    r.pfadd(f'hll:{page}:{today}', ip)

def estimate_unique_ips(page: str) -> int:
    today = date.today().isoformat()
    return r.pfcount(f'hll:{page}:{today}')
    # Error de ~0.81%, usa máximo 12KB por clave

# Arquitectura: Redis Pub/Sub para delivery en tiempo real + List para offline
# El cliente se conecta vía WebSocket, el servidor escucha los canales del usuario

class NotificationService:
    def __init__(self, redis_client):
        self.r = redis_client
    
    def send_notification(self, user_id: str, notification: dict):
        notification_id = str(uuid.uuid4())
        notification.update({'id': notification_id, 'ts': time.time(), 'read': False})
        
        # 1. Persistir en lista (para usuarios offline)
        key = f'notifications:{user_id}'
        self.r.lpush(key, json.dumps(notification))
        self.r.ltrim(key, 0, 99)  # Mantener sólo las últimas 100
        
        # 2. Publicar para usuarios online
        self.r.publish(f'live:user:{user_id}', json.dumps(notification))
        
        # 3. Incrementar badge counter
        self.r.incr(f'notifications:unread:{user_id}')
        
        return notification_id
    
    def get_notifications(self, user_id: str, limit: int = 20) -> list:
        key = f'notifications:{user_id}'
        items = self.r.lrange(key, 0, limit - 1)
        return [json.loads(item) for item in items]
    
    def mark_all_read(self, user_id: str):
        self.r.set(f'notifications:unread:{user_id}', 0)
        notifications = self.get_notifications(user_id, 100)
        pipe = self.r.pipeline()
        key = f'notifications:{user_id}'
        pipe.delete(key)
        for notif in notifications:
            notif['read'] = True
            pipe.rpush(key, json.dumps(notif))
        pipe.execute()
    
    def get_unread_count(self, user_id: str) -> int:
        count = self.r.get(f'notifications:unread:{user_id}')
        return int(count or 0)

# 1. SLOWLOG: ver los comandos más lentos
redis-cli SLOWLOG GET 10
redis-cli SLOWLOG LEN
redis-cli CONFIG SET slowlog-log-slower-than 1000  # en microsegundos

# 2. INFO stats para ver patrones generales
redis-cli INFO stats | grep -E 'keyspace_hits|keyspace_misses|rejected_connections'

# 3. LATENCY: historial de latencia
redis-cli LATENCY HISTORY event
redis-cli LATENCY LATEST
redis-cli LATENCY RESET

# 4. CLIENT LIST: ver conexiones activas y sus estados
redis-cli CLIENT LIST

# 5. MONITOR: ver todos los comandos en tiempo real (NO usar en producción bajo carga)
redis-cli MONITOR | grep -E 'KEYS|HGETALL' | head -20

# Desde código: instrumentar todas las operaciones
import time
from contextlib import contextmanager

@contextmanager
def redis_timer(operation: str):
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = (time.perf_counter() - start) * 1000
        if elapsed > 10:  # alertar si > 10ms
            print(f"SLOW REDIS OP: {operation} took {elapsed:.2f}ms")

with redis_timer('get_user_1001'):
    user = r.get('user:1001')

# Verificar hit rate del cache
info = r.info('stats')
hits = info['keyspace_hits']
misses = info['keyspace_misses']
hit_rate = hits / (hits + misses) if (hits + misses) > 0 else 0
print(f"Cache hit rate: {hit_rate:.1%}")