11. Nest avanzado: caché, colas, tiempo real, GraphQL y microservicios

11.1 Qué vas a poder hacer al terminar

• Decidir qué se cachea, dónde se cachea y cuándo se invalida, y razonar por qué la invalidación es la parte difícil y no la lectura.
• Montar caché en memoria y con Redis usando @nestjs/cache-manager, y explicar por qué el CacheInterceptor automático es peligroso en endpoints que dependen del usuario.
• Sacar de la petición HTTP todo el trabajo que no necesita estar ahí: colas BullMQ con reintentos, retroceso exponencial, deduplicación y consumidores idempotentes.
• Programar tareas con @nestjs/schedule y resolver el problema clásico del cron que se ejecuta N veces cuando hay N réplicas.
• Distinguir un evento interno de un mensaje en una cola real, y elegir con criterio en lugar de por costumbre.
• Limitar peticiones por IP y por usuario con almacenamiento compartido, funcionando correctamente detrás de un proxy inverso.
• Construir un gateway de WebSockets autenticado en el handshake, con salas, difusión selectiva y adaptador de Redis para escalar horizontalmente; y saber cuándo bastan los Server-Sent Events.
• Exponer una API GraphQL code-first sin caer en el problema N+1, con DataLoader y con límites de profundidad y complejidad.
• Argumentar con datos por qué todavía no necesitas microservicios, y si los necesitas, elegir transporte y aplicar saga, outbox, idempotencia y circuit breaker.
• Implementar subida y descarga de archivos con validación real del contenido y URLs prefirmadas.
• Publicar sondas de liveness y readiness con Terminus y entender qué hace Kubernetes con cada una.

11.2 Caché: el arte de no volver a calcular lo mismo

11.2.1 Por qué se cachea, y qué se puede cachear

Cachear es guardar el resultado de una operación costosa para reutilizarlo en peticiones posteriores. El motivo no es solo la latencia: es que el trabajo que no haces no consume CPU, ni conexiones a la base de datos, ni ancho de banda. Una consulta que tarda 300 ms y se repite 500 veces por minuto no es un problema de 300 ms, es un problema de 2,5 minutos de CPU de base de datos por minuto de reloj, es decir, un sistema que ya no da abasto.

La pregunta previa a cualquier caché no es técnica, es de producto: ¿cuántos segundos de desactualización tolera este dato? Si la respuesta es «cero», no hay caché posible y hay que optimizar la consulta (índices, capítulo 17). Si la respuesta es «treinta segundos», ya tienes tu TTL y todo lo demás es implementación.

11.2.2 Los cinco niveles de caché

Antes de escribir una línea de código conviene situarse: hay caché en muchos sitios y la más barata siempre es la que está más cerca del usuario.

  NAVEGADOR                CDN / EDGE            APLICACIÓN (Node)        DISTRIBUIDA        BASE DE DATOS
 ┌──────────┐            ┌────────────┐        ┌──────────────────┐    ┌─────────────┐    ┌──────────────┐
 │ HTTP     │            │ Cloudflare │        │  Map en memoria  │    │   Redis     │    │ buffer pool  │
 │ Cache-   │──miss────► │ Fastly     │─miss──►│  del proceso     │─►  │ Valkey      │─►  │ plan cache   │
 │ Control  │            │ Varnish    │        │  (LRU + TTL)     │    │ Memcached   │    │ vistas mat.  │
 │ ETag     │            └────────────┘        └──────────────────┘    └─────────────┘    └──────────────┘
 └──────────┘
  0 ms                     ~5-30 ms              ~0,01 ms              ~0,5-2 ms            ~5-300 ms
  gratis                   por petición          por proceso           compartida           última parada

  ── Coste de un fallo de caché ────────────────────────────────────────────────────────────────────►
  ◄─ Coste de una invalidación incorrecta (datos obsoletos servidos a más usuarios) ──────────────────

  REGLA: cachea lo más arriba que la corrección te permita, no lo más arriba posible.

• Navegador. Cabeceras Cache-Control, ETag y Last-Modified. Es la caché más eficiente porque ahorra incluso la conexión, pero no la controlas: una vez enviada una respuesta con max-age=3600 no hay forma de retirarla de los navegadores.
• CDN o proxy inverso. Ideal para recursos estáticos y para respuestas públicas idénticas para todos. Se invalida con purgas explícitas. Cuidado: si la respuesta depende del usuario y no envías Cache-Control: private o Vary: Authorization, la CDN puede servir la respuesta de un usuario a otro. Es un incidente de seguridad clásico.
• Aplicación, en memoria del proceso. Rapidísima y trivial de montar. Su límite es estructural: con N réplicas tienes N cachés incoherentes y cada despliegue las vacía. Sirve para datos de configuración inmutables y para reducir «ráfagas» dentro de una misma petición.
• Distribuida (Redis). Una única fuente de verdad para todas las réplicas, con TTL, invalidación centralizada y estructuras de datos útiles (contadores, conjuntos, pub/sub). Es la opción por defecto en producción. Coste: una dependencia más que puede caerse, y latencia de red.
• Base de datos. Tiene su propia caché de páginas y de planes, y además vistas materializadas. Muchas veces «necesito Redis» significa en realidad «me falta un índice». Mide antes de añadir capas.

11.2.3 Invalidación, TTL y estampida: los tres problemas difíciles

Hay una cita atribuida a Phil Karlton que resume tres décadas de dolor: «solo hay dos cosas difíciles en informática: la invalidación de caché y poner nombres a las cosas». Veamos las estrategias reales.

Recomendación profesional: combina siempre TTL corto como red de seguridad con invalidación explícita. El TTL garantiza que un olvido tuyo dure minutos y no meses; la invalidación explícita da la frescura que el usuario espera. Nunca dependas solo de la invalidación explícita: es código, y el código tiene huecos.

La estampida de caché (cache stampede o thundering herd)

Una clave muy solicitada caduca. En ese instante llegan 400 peticiones concurrentes, las 400 encuentran la caché vacía y las 400 lanzan la misma consulta costosa a la base de datos. La caché, que existía para proteger la base de datos, acaba de convertirse en el detonante de su caída.

  SIN PROTECCIÓN                                CON BLOQUEO (single-flight)

  t=0  clave caduca                             t=0  clave caduca
       ┌── req 1 ──► BD (300 ms)                     ┌── req 1 ──► adquiere lock ──► BD
       ├── req 2 ──► BD (300 ms)                     ├── req 2 ──► lock ocupado ──► espera/valor viejo
       ├── req 3 ──► BD (300 ms)                     ├── req 3 ──► lock ocupado ──► espera/valor viejo
       │   ...  400 consultas idénticas              │   ...
       └── req 400 ► BD (300 ms)                     └── req 400 ► lock ocupado ──► espera/valor viejo
                                                     t=300ms  req 1 escribe caché y libera
       RESULTADO: BD saturada, latencia x20                  RESULTADO: 1 consulta, latencia normal

Mitigaciones, de la más simple a la más robusta:

• TTL con jitter. En lugar de 300 s exactos para todas las entradas, usa 300 s ± 10 % aleatorio. No resuelve la estampida de una clave única, pero evita que miles de claves caduquen en el mismo segundo (típico tras un despliegue que precalienta la caché).
• Single-flight en proceso. Guarda en un Map la promesa en curso por clave: la segunda petición concurrente reutiliza la promesa de la primera en lugar de lanzar otra consulta. Elimina la estampida dentro de cada réplica, que suele ser el 90 % del problema.
• Bloqueo distribuido. Un SET clave:lock valor NX PX 5000 en Redis: solo quien adquiere el lock recalcula; el resto espera brevemente o devuelve el valor antiguo.
• Refresco proactivo (stale-while-revalidate). Guarda dos tiempos: uno «fresco» y otro «tolerable». Si el dato está entre ambos, devuélvelo inmediatamente y lanza el recálculo en segundo plano (una cola, sección 11.3). El usuario nunca espera y la base de datos nunca recibe una avalancha.
• Precalentamiento. Un trabajo programado que rellena las claves críticas antes de que caduquen.

/**
 * Colapsa llamadas concurrentes con la misma clave en una sola ejecución.
 * Ámbito: el proceso actual. Con varias réplicas necesitas además un lock en Redis.
 */
export class SingleFlight {
  private readonly enVuelo = new Map<string, Promise<unknown>>();

  async run<T>(clave: string, fn: () => Promise<T>): Promise<T> {
    const existente = this.enVuelo.get(clave);
    if (existente) return existente as Promise<T>;

    // Importante: guardamos la promesa ANTES del primer await,
    // si no, dos llamadas simultáneas seguirían pasando de largo.
    const promesa = fn().finally(() => this.enVuelo.delete(clave));
    this.enVuelo.set(clave, promesa);
    return promesa;
  }
}

11.2.4 @nestjs/cache-manager: configuración en memoria y con Redis

Nest no implementa su propia caché: envuelve la librería cache-manager, que abstrae el almacén subyacente. Eso permite empezar en memoria y pasar a Redis cambiando la configuración del módulo, sin tocar los servicios.

import { Module } from '@nestjs/common';
import { CacheModule } from '@nestjs/cache-manager';

@Module({
  imports: [
    CacheModule.register({
      isGlobal: true,   // evita importar CacheModule en cada módulo
      ttl: 30_000,      // ms en cache-manager v5+ (¡segundos en v4!)
      max: 500,         // número máximo de elementos en memoria: SIEMPRE ponlo
    }),
  ],
})
export class AppModule {}

import { CacheModule } from '@nestjs/cache-manager';
import { redisStore } from 'cache-manager-redis-yet';
import { ConfigModule, ConfigService } from '@nestjs/config';

CacheModule.registerAsync({
  isGlobal: true,
  imports: [ConfigModule],
  inject: [ConfigService],
  useFactory: async (config: ConfigService) => ({
    store: await redisStore({
      url: config.getOrThrow<string>('REDIS_URL'),
      ttl: 30_000,
    }),
  }),
});

import { CacheModule } from '@nestjs/cache-manager';
import { createKeyv } from '@keyv/redis';
import { ConfigModule, ConfigService } from '@nestjs/config';

CacheModule.registerAsync({
  isGlobal: true,
  imports: [ConfigModule],
  inject: [ConfigService],
  useFactory: (config: ConfigService) => ({
    // 'stores' en plural: admite varios niveles en cascada
    stores: [createKeyv(config.getOrThrow<string>('REDIS_URL'))],
    ttl: 30_000,
  }),
});

11.2.5 CacheInterceptor automático y sus limitaciones

Nest incluye un interceptor que cachea respuestas sin escribir código. Es cómodo y, mal usado, es la vía más rápida a un incidente de privacidad.

import { Controller, Get, UseInterceptors } from '@nestjs/common';
import { CacheInterceptor, CacheKey, CacheTTL } from '@nestjs/cache-manager';

@Controller('paises')
@UseInterceptors(CacheInterceptor)      // aplicado a todo el controlador
export class PaisesController {
  @Get()
  @CacheKey('paises:todos')             // clave fija en lugar de la URL
  @CacheTTL(24 * 60 * 60 * 1000)        // 24 h en ms
  findAll() {
    return this.service.findAll();      // catálogo público e idéntico para todos: caso ideal
  }
}

Lo que hace exactamente el interceptor y por qué importa cada detalle:

• Solo cachea peticiones GET en el transporte HTTP. Un POST nunca se cachea, lo cual es correcto y deliberado.
• La clave por defecto es la URL de la petición (ruta más query string). No incluye cabeceras, ni cookies, ni el usuario autenticado. Esto es el origen del error más grave de esta sección.
• Cachea la respuesta ya serializada por el controlador, es decir, después de los interceptores anteriores pero sin volver a pasar por la lógica de negocio. Si tu autorización vive en un guard, el guard sí se ejecuta (los guards van antes de los interceptores); pero si el filtrado de datos por permisos ocurre en el servicio, ese filtrado se salta.
• No sabe nada de tu dominio, así que no puede invalidar nada. Solo TTL.
• Para adaptarlo, se extiende y se sobrescribe trackBy(context), que devuelve la clave o undefined para no cachear.

// GET /tareas devuelve SOLO las tareas del usuario autenticado.
// La clave por defecto es la URL: '/tareas' para todos.
@Controller('tareas')
@UseInterceptors(CacheInterceptor)
export class TareasController {
  @Get()
  @CacheTTL(60_000)
  findMine(@CurrentUser() user: JwtUser) {
    return this.service.findByOwner(user.id);
  }
}
// Ana entra primero  → se cachea su lista bajo '/tareas'
// Luis entra después → recibe LAS TAREAS DE ANA.
// Fuga de datos entre usuarios: no es un bug de rendimiento,
// es una brecha de seguridad notificable.

// Opción A: no uses el interceptor automático en datos por usuario.
// Opción B: si lo usas, la identidad SIEMPRE forma parte de la clave.
@Injectable()
export class CachePorUsuarioInterceptor extends CacheInterceptor {
  protected trackBy(context: ExecutionContext): string | undefined {
    const req = context.switchToHttp().getRequest();
    if (req.method !== 'GET') return undefined;   // no cachear escrituras

    const userId = req.user?.sub;
    // Sin usuario identificado, mejor no cachear que cachear mal.
    if (!userId) return undefined;

    return `u:${userId}:${req.originalUrl}`;
  }
}

11.2.6 Uso manual: el patrón cache-aside

El interceptor sirve para catálogos. Para todo lo demás se inyecta la caché y se controla a mano. El patrón canónico se llama cache-aside (o lazy loading): leer de caché, y si no está, calcular, guardar y devolver.

            ┌──────────────┐   1. get(clave)    ┌───────────┐
            │   Servicio   │───────────────────►│   CACHÉ   │
            │              │◄───────────────────│  (Redis)  │
            └──────┬───────┘   2. hit → fin     └───────────┘
                   │ 2'. miss                         ▲
                   ▼                                  │ 4. set(clave, valor, ttl)
            ┌──────────────┐   3. consulta            │
            │ Base de datos│──────────────────────────┘
            └──────────────┘

import { Inject, Injectable, Logger } from '@nestjs/common';
import { CACHE_MANAGER } from '@nestjs/cache-manager';
import { Cache } from 'cache-manager';

@Injectable()
export class InformesService {
  private readonly log = new Logger(InformesService.name);

  constructor(
    @Inject(CACHE_MANAGER) private readonly cache: Cache,
    private readonly em: EntityManager,
  ) {}

  async resumenMensual(proyectoId: string, mes: string): Promise<Resumen> {
    const clave = `informe:resumen:${proyectoId}:${mes}`;

    const enCache = await this.cache.get<Resumen>(clave);
    if (enCache) return enCache;                       // acierto

    const resumen = await this.calcularResumen(proyectoId, mes);   // ~800 ms

    // TTL con jitter: evita que todas las claves caduquen a la vez.
    const ttl = 300_000 + Math.floor(Math.random() * 30_000);
    await this.cache.set(clave, resumen, ttl);
    return resumen;
  }

  /** Envoltorio reutilizable: no repitas el try/get/set en veinte servicios. */
  async conCache<T>(clave: string, ttlMs: number, fn: () => Promise<T>): Promise<T> {
    try {
      const hit = await this.cache.get<T>(clave);
      if (hit !== undefined && hit !== null) return hit;
    } catch (e) {
      // Redis caído NO debe tumbar la aplicación: degradar, no fallar.
      this.log.warn(`Caché no disponible en lectura (${clave}): ${e}`);
      return fn();
    }

    const valor = await fn();
    try {
      await this.cache.set(clave, valor, ttlMs);
    } catch (e) {
      this.log.warn(`Caché no disponible en escritura (${clave}): ${e}`);
    }
    return valor;
  }
}

11.2.7 Ejemplo de producción: listado con filtros, usuario e invalidación por evento

Este es el caso realista: un listado paginado, con filtros arbitrarios, específico del usuario, que debe reflejar de inmediato los cambios que ese usuario hace. La solución combina tres ideas: clave canónica (los mismos filtros en distinto orden deben producir la misma clave), versión por usuario (invalidar sin enumerar claves) y eventos de dominio (quien escribe no conoce a quien cachea).

import { Inject, Injectable } from '@nestjs/common';
import { CACHE_MANAGER } from '@nestjs/cache-manager';
import { Cache } from 'cache-manager';
import { OnEvent } from '@nestjs/event-emitter';
import { createHash } from 'node:crypto';

interface FiltroTareas {
  estado?: 'pendiente' | 'hecha';
  etiquetas?: string[];
  page?: number;
  limit?: number;
}

@Injectable()
export class TareasCacheService {
  private static readonly TTL = 120_000;   // 2 min como red de seguridad

  constructor(@Inject(CACHE_MANAGER) private readonly cache: Cache) {}

  /**
   * Clave canónica: ordenamos las entradas y los arrays para que
   * ?estado=hecha&page=1 y ?page=1&estado=hecha compartan entrada.
   * El hash mantiene la clave corta y sin caracteres problemáticos.
   */
  private huella(filtro: FiltroTareas): string {
    const normalizado = {
      estado: filtro.estado ?? null,
      etiquetas: [...(filtro.etiquetas ?? [])].sort(),
      page: filtro.page ?? 1,
      limit: Math.min(filtro.limit ?? 20, 100),
    };
    return createHash('sha1').update(JSON.stringify(normalizado)).digest('hex').slice(0, 16);
  }

  /** Versión del espacio de claves de un usuario: invalidar = incrementarla. */
  private async version(userId: string): Promise<number> {
    const v = await this.cache.get<number>(`tareas:ver:${userId}`);
    if (v === undefined || v === null) {
      await this.cache.set(`tareas:ver:${userId}`, 1, 0);   // 0 = sin caducidad
      return 1;
    }
    return v;
  }

  async listar(
    userId: string,
    filtro: FiltroTareas,
    cargar: () => Promise<Pagina<TareaDto>>,
  ): Promise<Pagina<TareaDto>> {
    const v = await this.version(userId);
    const clave = `tareas:u:${userId}:v:${v}:f:${this.huella(filtro)}`;

    const hit = await this.cache.get<Pagina<TareaDto>>(clave);
    if (hit) return hit;

    const pagina = await cargar();
    await this.cache.set(clave, pagina, TareasCacheService.TTL);
    return pagina;
  }

  /**
   * Invalidación por evento: una sola escritura invalida TODAS las
   * combinaciones de filtros y páginas de ese usuario, sin recorrer claves.
   * (Buscar claves con KEYS * en Redis en producción es un antipatrón:
   *  bloquea el servidor. Con este truco no hace falta.)
   */
  @OnEvent('tarea.creada')
  @OnEvent('tarea.actualizada')
  @OnEvent('tarea.eliminada')
  async invalidar(evento: { propietarioId: string }): Promise<void> {
    const actual = await this.version(evento.propietarioId);
    await this.cache.set(`tareas:ver:${evento.propietarioId}`, actual + 1, 0);
  }
}

11.3 Colas y trabajos en segundo plano

11.3.1 Por qué una petición HTTP no debe hacerlo todo

Un usuario se registra. Hay que crear la fila en la base de datos, enviar un correo de bienvenida, generar su avatar, avisar al CRM y apuntar el evento en la analítica. Si todo eso ocurre dentro del POST /registro, la respuesta tarda tres segundos y, lo que es peor, si el servidor de correo está caído el registro falla aunque la parte importante haya funcionado. Has acoplado la corrección de tu caso de uso a la disponibilidad de un tercero.

Una cola invierte el planteamiento: la petición hace solo lo imprescindible y transaccional, encola el resto y responde. Un proceso aparte —el worker— consume esos trabajos con sus propios reintentos y a su propio ritmo.

   PETICIÓN HTTP (rápida, transaccional)             WORKER (lento, reintentable)
  ┌───────────────────────────┐                     ┌──────────────────────────────┐
  │ 1. valida DTO             │                     │ 5. toma el trabajo (BRPOPLPUSH│
  │ 2. INSERT usuario  ──┐    │                     │    atómico: nadie más lo ve) │
  │ 3. queue.add(...) ───┼────┼── PRODUCTOR ──►     │ 6. envía el correo           │
  │ 4. responde 201      │    │      ┌────────┐     │ 7a. ok  → completed          │
  └──────────────────────┼────┘      │ REDIS  │◄────│ 7b. err → retry con backoff  │
        ~40 ms           │           │  cola  │     │ 7c. agotado → failed (DLQ)   │
                         └──────────►└────────┘     └──────────────────────────────┘
                                      ▲   ▲
                                      │   └── otros workers: escalado horizontal
                                      └────── panel de monitorización

  Estados de un trabajo:  waiting ──► active ──► completed
                            ▲           │
                            └─ delayed ─┴──► failed ──► (reintento) ──► waiting

Las cuatro razones que justifican una cola, por orden de frecuencia real:

• No bloquear la respuesta. Todo lo que el usuario no necesita ver para continuar debe salir del camino crítico: correos, notificaciones push, sincronizaciones, webhooks salientes.
• Reintentos con garantías. Un try/catch en el controlador no sobrevive a un reinicio del proceso. Un trabajo en Redis sí: se reintentará aunque despliegues en medio.
• Absorber picos. Si llegan 10 000 peticiones en un minuto, la cola crece y los workers las procesan a su ritmo sostenible. Sin cola, se saturan las conexiones a la base de datos y caen todas.
• Trabajos largos. Generar un PDF de 400 páginas o procesar un CSV de 2 millones de filas no cabe en el timeout de un balanceador (típicamente 30-60 s) y bloquearía el event loop (capítulo 1).

11.3.2 BullMQ con @nestjs/bullmq

BullMQ es la evolución de Bull: una cola de trabajos sobre Redis, escrita en TypeScript, que usa scripts Lua para garantizar atomicidad en las transiciones de estado. Es el estándar de facto en el ecosistema Node.

import { BullModule } from '@nestjs/bullmq';

@Module({
  imports: [
    BullModule.forRootAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: (c: ConfigService) => ({
        connection: {
          host: c.getOrThrow('REDIS_HOST'),
          port: c.get<number>('REDIS_PORT', 6379),
          // Obligatorio si pasas tu propia instancia de ioredis:
          // los workers usan comandos bloqueantes y no admiten
          // el reintento por comando de ioredis.
          maxRetriesPerRequest: null,
        },
        defaultJobOptions: {
          attempts: 3,
          backoff: { type: 'exponential', delay: 2000 },
          removeOnComplete: { age: 3600, count: 1000 },
          removeOnFail: { age: 24 * 3600 },
        },
      }),
    }),
    BullModule.registerQueue({ name: 'correos' }, { name: 'informes' }),
  ],
})
export class AppModule {}

import { InjectQueue } from '@nestjs/bullmq';
import { Queue } from 'bullmq';

@Injectable()
export class CorreosProducer {
  constructor(@InjectQueue('correos') private readonly cola: Queue) {}

  async bienvenida(usuarioId: string, email: string) {
    await this.cola.add(
      'bienvenida',                      // nombre: obligatorio en BullMQ
      { usuarioId, email },              // datos: deben ser serializables a JSON
      {
        // jobId estable = deduplicación: si ya existe un trabajo
        // con este id, add() no crea otro.
        jobId: `bienvenida:${usuarioId}`,
        attempts: 5,
        backoff: { type: 'exponential', delay: 5000 },
        priority: 1,                     // 1 = máxima prioridad
      },
    );
  }
}

11.3.3 El consumidor: @Processor y WorkerHost

import { Processor, WorkerHost, OnWorkerEvent } from '@nestjs/bullmq';
import { Job, UnrecoverableError } from 'bullmq';
import { Logger } from '@nestjs/common';

@Processor('correos', {
  concurrency: 5,        // 5 trabajos en paralelo POR réplica
  limiter: { max: 100, duration: 60_000 },   // tope de 100/min: respeta la cuota del proveedor
})
export class CorreosProcessor extends WorkerHost {
  private readonly log = new Logger(CorreosProcessor.name);

  constructor(
    private readonly smtp: SmtpService,
    private readonly usuarios: UsuariosService,
  ) {
    super();
  }

  /** Punto de entrada único: un solo process() para todos los nombres. */
  async process(job: Job<DatosCorreo, void, string>): Promise<void> {
    switch (job.name) {
      case 'bienvenida':
        return this.enviarBienvenida(job);
      case 'recuperar-clave':
        return this.enviarRecuperacion(job);
      default:
        // Un nombre desconocido nunca se arreglará reintentando.
        throw new UnrecoverableError(`Trabajo no soportado: ${job.name}`);
    }
  }

  private async enviarBienvenida(job: Job<DatosCorreo>): Promise<void> {
    const usuario = await this.usuarios.findById(job.data.usuarioId);

    // El usuario se borró entre el encolado y el consumo: no es un error del sistema.
    if (!usuario) throw new UnrecoverableError('Usuario inexistente');

    // Idempotencia: si ya se envió, no lo repitas (ver 11.3.5).
    if (usuario.bienvenidaEnviadaEn) {
      this.log.log(`Bienvenida ya enviada a ${usuario.id}, se omite`);
      return;
    }

    await job.updateProgress(30);
    await this.smtp.enviar({ to: usuario.email, plantilla: 'bienvenida' });
    await this.usuarios.marcarBienvenidaEnviada(usuario.id, new Date());
    await job.updateProgress(100);
  }

  @OnWorkerEvent('completed')
  onCompleted(job: Job) {
    this.log.log(`OK ${job.name}#${job.id} en ${Date.now() - job.timestamp} ms`);
  }

  @OnWorkerEvent('failed')
  onFailed(job: Job | undefined, err: Error) {
    // attemptsMade incluye el intento actual: compáralo con opts.attempts
    const agotado = job ? job.attemptsMade >= (job.opts.attempts ?? 1) : true;
    const nivel = agotado ? 'error' : 'warn';
    this.log[nivel](`FALLO ${job?.name}#${job?.id} (${job?.attemptsMade}): ${err.message}`);
    if (agotado) this.alertas.notificar(job, err);   // ahora sí, avisa a una persona
  }

  @OnWorkerEvent('stalled')
  onStalled(jobId: string) {
    // Un trabajo "atascado" es un worker que dejó de renovar su lock:
    // casi siempre significa que bloqueaste el event loop o que te mató el OOM killer.
    this.log.warn(`Trabajo atascado ${jobId}: revisa CPU y memoria del worker`);
  }
}

11.3.4 Opciones de trabajo: el vocabulario completo

11.3.5 Idempotencia del consumidor: lo más importante de esta sección

Toda cola distribuida ofrece entrega «al menos una vez», no «exactamente una vez». Esto no es un defecto de BullMQ: es una consecuencia de que la red y los procesos fallan. Un worker puede enviar el correo y morir antes de marcar el trabajo como completado; el trabajo volverá a la cola y el correo se enviará dos veces. La única defensa es que ejecutar el trabajo dos veces produzca el mismo resultado que ejecutarlo una.

async process(job: Job<{ pedidoId: string }>) {
  const pedido = await this.pedidos.find(job.data.pedidoId);

  // Si el proceso muere DESPUÉS del cargo y ANTES de
  // completar el trabajo, se reintenta y se cobra otra vez.
  await this.pasarela.cobrar(pedido.total, pedido.tarjeta);

  pedido.estado = 'pagado';
  await this.em.flush();

  // Efecto secundario adicional, también duplicable:
  await this.smtp.enviar({ to: pedido.email, plantilla: 'recibo' });
}

async process(job: Job<{ pedidoId: string }>) {
  const pedido = await this.pedidos.find(job.data.pedidoId);

  // 1. Comprobación de estado: barrera barata para el caso normal.
  if (pedido.estado === 'pagado') return;

  // 2. Clave de idempotencia enviada a la pasarela: si repites la
  //    misma clave, la pasarela devuelve el cargo original en lugar
  //    de crear otro. Es el mecanismo que ofrecen Stripe, Adyen, etc.
  const clave = `pedido:${pedido.id}:cobro`;
  const cargo = await this.pasarela.cobrar(pedido.total, pedido.tarjeta, {
    idempotencyKey: clave,
  });

  // 3. Escritura condicional: solo pasa de pendiente a pagado.
  //    Si otro worker ganó la carrera, afectadas = 0 y salimos.
  const afectadas = await this.pedidos.marcarPagadoSiPendiente(pedido.id, cargo.id);
  if (afectadas === 0) return;

  // 4. El correo se encola aparte, con su propio jobId estable.
  await this.correos.add('recibo', { pedidoId: pedido.id }, {
    jobId: `recibo:${pedido.id}`,
  });
}

Las cuatro técnicas de idempotencia, por orden de robustez:

• Operaciones naturalmente idempotentes. UPDATE ... SET estado = 'pagado' es idempotente; UPDATE ... SET intentos = intentos + 1 no lo es. Diseña hacia la primera forma siempre que puedas.
• Escritura condicional. UPDATE ... WHERE id = ? AND estado = 'pendiente' y comprueba las filas afectadas. Es una transición de estado atómica en la propia base de datos, sin locks explícitos.
• Tabla de trabajos procesados. Una tabla con clave primaria (tipo, clave); el consumidor inserta al empezar y una violación de unicidad significa «ya hecho». Funciona con cualquier efecto secundario, incluso los que no controlas.
• Clave de idempotencia en el tercero. Cuando el efecto secundario está fuera de tu sistema (cobros, correos, mensajes), es la única forma de deduplicar de verdad. Casi todas las APIs de pago la ofrecen; úsala.

11.3.6 Eventos, fallos y monitorización

• Eventos locales (@OnWorkerEvent): los emite tu worker sobre sus trabajos. Perfecto para métricas y trazas del consumidor.
• Eventos globales (clase QueueEvents de BullMQ): llegan a cualquier proceso suscrito, incluido el que solo produce. Es lo que necesitas si la API quiere reaccionar cuando termina un trabajo que encoló (por ejemplo, para avisar por WebSocket, sección 11.6).
• Cola de fallidos. En BullMQ los trabajos agotados quedan en el conjunto failed, con su failedReason y su stack trace. Desde ahí se pueden reintentar en masa (job.retry()) una vez arreglada la causa. Es tu dead letter queue de facto.
• Métricas mínimas que debes vigilar: tamaño de waiting (¿crece sin parar?), edad del trabajo más antiguo en espera, ratio de failed y trabajos stalled. Una cola que crece indefinidamente es un incidente aunque nada dé error.
• Panel. bull-board (con su adaptador para BullMQ y su integración para Nest) ofrece una interfaz web para inspeccionar, reintentar y limpiar. Protégelo siempre con autenticación: expone los datos de todos los trabajos y permite borrar la cola.

@Injectable()
export class QueueMetricsService {
  constructor(@InjectQueue('correos') private readonly cola: Queue) {}

  /** Expón esto en /metrics o en el health check de readiness. */
  async estado() {
    const [waiting, active, delayed, failed] = await Promise.all([
      this.cola.getWaitingCount(),
      this.cola.getActiveCount(),
      this.cola.getDelayedCount(),
      this.cola.getFailedCount(),
    ]);
    return { waiting, active, delayed, failed, saludable: waiting < 5_000 };
  }
}

11.3.7 Ejemplo real: informe pesado con seguimiento de progreso

Caso de uso completo: el usuario pide un informe que tarda dos minutos. La API responde de inmediato con un identificador, el worker lo genera actualizando el progreso, lo sube a un almacén de objetos y avisa al usuario. Es el patrón 202 Accepted con consulta de estado.

@Controller('informes')
export class InformesController {
  constructor(
    @InjectQueue('informes') private readonly cola: Queue,
    private readonly usuarios: UsuariosService,
  ) {}

  @Post()
  @HttpCode(202)                       // 202 Accepted: aceptado, aún no terminado
  async solicitar(@Body() dto: CrearInformeDto, @CurrentUser() user: JwtUser) {
    const jobId = `informe:${user.id}:${dto.mes}`;

    // Deduplicación: si ya está generándose el mismo informe, no dupliques trabajo.
    const existente = await this.cola.getJob(jobId);
    if (existente) {
      return { id: jobId, estado: await existente.getState(), duplicado: true };
    }

    await this.cola.add('mensual', { userId: user.id, mes: dto.mes }, {
      jobId,
      attempts: 3,
      backoff: { type: 'exponential', delay: 10_000 },
      removeOnComplete: { age: 7 * 24 * 3600 },   // conserva una semana para consultar estado
    });

    return { id: jobId, estado: 'waiting' };
  }

  @Get(':id/estado')
  async estado(@Param('id') id: string, @CurrentUser() user: JwtUser) {
    const job = await this.cola.getJob(id);
    if (!job) throw new NotFoundException();

    // AUTORIZACIÓN: el id contiene el usuario, pero nunca te fíes del formato.
    if (job.data.userId !== user.id) throw new ForbiddenException();

    return {
      estado: await job.getState(),
      progreso: job.progress,
      url: job.returnvalue?.url ?? null,
      error: job.failedReason ?? null,
    };
  }
}

@Processor('informes', { concurrency: 2 })   // pesados: poca concurrencia
export class InformesProcessor extends WorkerHost {
  constructor(
    private readonly em: EntityManager,
    private readonly almacen: AlmacenService,
    private readonly eventos: EventEmitter2,
  ) {
    super();
  }

  async process(job: Job<{ userId: string; mes: string }>): Promise<{ url: string }> {
    const { userId, mes } = job.data;

    // El worker vive fuera de una petición HTTP: NO hay contexto de petición.
    // Con MikroORM hay que crear un contexto propio (ver capítulo 15):
    // em.fork() da un EntityManager con su propia Identity Map y Unit of Work.
    const em = this.em.fork();

    const filas: FilaInforme[] = [];
    const total = await em.count(Movimiento, { mes });
    const LOTE = 1_000;

    // Paginar en lotes: cargar 2 millones de filas de golpe reventaría la memoria.
    for (let offset = 0; offset < total; offset += LOTE) {
      const lote = await em.find(Movimiento, { mes }, { limit: LOTE, offset, orderBy: { id: 'ASC' } });
      filas.push(...lote.map(transformar));

      // Liberamos la Identity Map: si no, el fork acumula todas las entidades.
      em.clear();

      await job.updateProgress(Math.round((offset / total) * 90));
    }

    const pdf = await generarPdf(filas);
    const url = await this.almacen.subir(`informes/${userId}/${mes}.pdf`, pdf);
    await job.updateProgress(100);

    // Avisamos al resto del sistema (correo, WebSocket) sin acoplarnos a ellos.
    this.eventos.emit('informe.listo', { userId, mes, url });

    // Lo devuelto queda en job.returnvalue y lo lee el endpoint de estado.
    return { url };
  }
}

11.3.8 Tareas programadas con @nestjs/schedule

Cuando el trabajo no lo dispara un usuario sino el reloj, no hace falta una cola: basta un planificador en el proceso. @nestjs/schedule envuelve la librería cron y añade tres decoradores.

import { Cron, CronExpression, Interval, Timeout, SchedulerRegistry } from '@nestjs/schedule';

@Injectable()
export class MantenimientoService {
  private readonly log = new Logger(MantenimientoService.name);

  // Expresión explícita con zona horaria: imprescindible en producción,
  // porque el contenedor casi siempre corre en UTC.
  @Cron('0 30 3 * * *', { name: 'purga-nocturna', timeZone: 'Europe/Madrid' })
  async purgarSesiones() {
    const borradas = await this.sesiones.borrarExpiradas();
    this.log.log(`Sesiones purgadas: ${borradas}`);
  }

  // Constantes legibles para los casos habituales.
  @Cron(CronExpression.EVERY_10_MINUTES)
  async sincronizarTarifas() { /* ... */ }

  // Cada 30 s desde el arranque. No es cron: es un setInterval gestionado.
  @Interval('latido', 30_000)
  latido() { this.metricas.gauge('app.vivo', 1); }

  // Una sola vez, 10 s después del arranque: ideal para precalentar caché.
  @Timeout(10_000)
  async precalentar() { await this.catalogo.precargar(); }

  /** Programación dinámica: añadir o quitar tareas en tiempo de ejecución. */
  constructor(private readonly registry: SchedulerRegistry) {}

  pausarPurga() {
    this.registry.getCronJob('purga-nocturna').stop();
  }
}

Ejemplos leídos en voz alta, que es la única forma fiable de revisar un cron:

• 0 30 3 * * * → «segundo 0, minuto 30, hora 3, cualquier día»: todos los días a las 3:30.
• 0 */15 * * * * → cada 15 minutos, en el segundo 0 (*/n significa «cada n»).
• 0 0 9 * * 1-5 → de lunes a viernes a las 9:00.
• 0 0 0 1 * * → el día 1 de cada mes a medianoche.
• 0 0 2 * * 0 → los domingos a las 2:00. Cuidado: si además pones día del mes, la mayoría de implementaciones interpretan los dos campos como un O lógico, no como un Y.

@Cron('0 0 3 1 * *')
async facturarMes() {
  // Con 3 réplicas: 3 ejecuciones simultáneas.
  // Además, todo el trabajo pesado ocurre DENTRO del cron:
  // si falla a mitad, no hay reintento ni rastro de por dónde iba.
  const clientes = await this.clientes.activos();
  for (const c of clientes) {
    const factura = await this.crearFactura(c);
    await this.smtp.enviar(factura);
  }
}

@Cron('0 0 3 1 * *', { timeZone: 'Europe/Madrid' })
async planificarFacturacion() {
  // 1. Cerrojo distribuido: SET NX PX es atómico en Redis.
  //    Solo una réplica sigue; el TTL evita cerrojos huérfanos.
  const periodo = new Date().toISOString().slice(0, 7);   // '2026-07'
  const conseguido = await this.redis.set(
    `lock:facturacion:${periodo}`, process.env.HOSTNAME ?? '1',
    'PX', 300_000, 'NX',
  );
  if (!conseguido) return;

  // 2. El cron solo ENCOLA; el trabajo pesado, con reintentos, va a la cola.
  const clientes = await this.clientes.activos({ fields: ['id'] });
  await this.cola.addBulk(clientes.map((c) => ({
    name: 'facturar-cliente',
    data: { clienteId: c.id, periodo },
    // jobId por cliente y periodo: la garantía real de "una factura por mes".
    opts: { jobId: `factura:${c.id}:${periodo}`, attempts: 5 },
  })));
}

11.4 Eventos internos: desacoplar módulos sin salir del proceso

Cuando un caso de uso termina, a menudo hay que avisar a otros módulos: se ha creado una tarea, hay que invalidar la caché, notificar por WebSocket, apuntar la métrica y quizá encolar un correo. Si el servicio de tareas llama a esos cuatro servicios, el módulo de tareas acaba importando media aplicación y cada nueva reacción obliga a modificar el código que la origina, violando el principio de abierto/cerrado.

@nestjs/event-emitter invierte esa dependencia: el emisor publica un hecho y no sabe quién escucha. Por debajo es eventemitter2, es decir, memoria del proceso actual.

EventEmitterModule.forRoot({
  wildcard: true,          // permite escuchar 'tarea.*'
  delimiter: '.',
  maxListeners: 20,
  verboseMemoryLeak: true, // avisa si un evento acumula listeners
});

// Evento de dominio: una clase, no un objeto suelto.
// Nombre en pasado: describe algo que YA ocurrió.
export class TareaCreadaEvent {
  constructor(
    readonly tareaId: string,
    readonly propietarioId: string,
    readonly proyectoId: string,
  ) {}
}

@Injectable()
export class TareasService {
  constructor(private readonly eventos: EventEmitter2) {}

  async crear(dto: CrearTareaDto, userId: string) {
    const tarea = this.em.create(Tarea, { ...dto, propietario: userId });
    await this.em.flush();      // primero persistimos...

    // ...y solo después publicamos: nunca anuncies algo que aún puede fallar.
    this.eventos.emit('tarea.creada', new TareaCreadaEvent(tarea.id, userId, dto.proyectoId));
    return tarea;
  }
}

@Injectable()
export class TareasListener {
  private readonly log = new Logger(TareasListener.name);

  // async: true ejecuta el listener sin bloquear al emisor.
  @OnEvent('tarea.creada', { async: true })
  async alCrear(e: TareaCreadaEvent) {
    // Un listener NUNCA debe romper al emisor: captura tus errores.
    try {
      await this.push.notificar(e.propietarioId, 'Tarea creada');
    } catch (err) {
      this.log.error(`Fallo notificando ${e.tareaId}`, err as Error);
    }
  }

  // Comodín: útil para auditoría y métricas transversales.
  @OnEvent('tarea.*')
  auditar(payload: unknown, ...args: unknown[]) {
    this.auditoria.registrar(payload);
  }
}

Regla de decisión. Pregúntate: «si esta reacción no ocurre, ¿alguien se queja?». Si la respuesta es sí, necesitas una cola. Si es «se pierde una métrica», un evento interno es perfecto. Y si la reacción es parte de la corrección del caso de uso —descontar el stock al confirmar el pedido— entonces no es un evento: es una llamada directa dentro de la misma transacción.

11.5 Rate limiting y protección de la API

Sin límite de peticiones, cualquiera puede probar diez mil contraseñas por minuto en tu /login, agotar tu cuota de un servicio de pago o tumbar la base de datos con un bucle. @nestjs/throttler implementa una ventana temporal por identificador: N peticiones cada T milisegundos.

import { ThrottlerModule, ThrottlerGuard } from '@nestjs/throttler';
import { ThrottlerStorageRedisService } from 'nestjs-throttler-storage-redis';

@Module({
  imports: [
    ThrottlerModule.forRootAsync({
      inject: [ConfigService],
      useFactory: (c: ConfigService) => ({
        // Varias ventanas simultáneas: todas deben cumplirse.
        // Corta las ráfagas sin castigar el uso sostenido normal.
        throttlers: [
          { name: 'corta', ttl: 1_000, limit: 10 },      // 10 por segundo
          { name: 'media', ttl: 60_000, limit: 120 },    // 120 por minuto
          { name: 'larga', ttl: 3_600_000, limit: 2_000 },
        ],
        // Sin almacenamiento compartido, cada réplica cuenta por su cuenta:
        // con 4 réplicas el límite real es 4 veces el configurado.
        storage: new ThrottlerStorageRedisService(c.getOrThrow('REDIS_URL')),
        // No limitar las sondas de salud ni el scrapeo de métricas.
        skipIf: (ctx) => ['/health', '/metrics'].includes(
          ctx.switchToHttp().getRequest().path,
        ),
      }),
    }),
  ],
  providers: [{ provide: APP_GUARD, useClass: ThrottlerGuard }],   // global
})
export class AppModule {}

// El login hereda el límite global de 120/min: suficiente
// para 172 800 intentos de contraseña al día por IP.
@Post('login')
login(@Body() dto: LoginDto) {
  return this.auth.login(dto);
}

// Y en el arranque, detrás de Nginx o de un balanceador,
// req.ip es la IP DEL PROXY: todos los usuarios del mundo
// comparten un único contador. Un solo usuario activo
// bloquea la API para todos los demás.

// Límite específico y agresivo en el punto sensible.
@Throttle({ corta: { limit: 3, ttl: 60_000 } })
@Post('login')
login(@Body() dto: LoginDto) {
  return this.auth.login(dto);
}

@SkipThrottle()                 // webhook interno de confianza
@Post('webhooks/stripe')
webhook(@Body() body: unknown) { /* ... */ }

// main.ts: decirle a Express de cuántos proxies fiarse.
// El número es la cantidad de saltos de confianza; NUNCA
// uses 'true' en internet abierta, porque entonces cualquiera
// puede falsificar X-Forwarded-For y saltarse el límite.
const app = await NestFactory.create(AppModule);
app.set('trust proxy', 1);      // Fastify: { trustProxy: 1 }

Para limitar por usuario en lugar de por IP —lo correcto en endpoints autenticados, porque una oficina entera comparte IP— se extiende el guard y se sobrescribe getTracker:

@Injectable()
export class ThrottlerUsuarioGuard extends ThrottlerGuard {
  protected async getTracker(req: Record<string, any>): Promise<string> {
    // Usuario autenticado → contador propio; anónimo → por IP.
    // Prefijos distintos para que un id numérico no colisione con una IP.
    return req.user?.sub ? `u:${req.user.sub}` : `ip:${req.ips?.[0] ?? req.ip}`;
  }
}

11.6 WebSockets y tiempo real

11.6.1 El gateway: decoradores y ciclo de vida

HTTP es petición-respuesta: el servidor no puede hablar primero. Para notificar cambios en el momento en que ocurren hace falta una conexión persistente y bidireccional. Un gateway de Nest es el equivalente a un controlador, pero para mensajes de WebSocket; por defecto usa Socket.IO, y con WsAdapter puede usar ws nativo.

import {
  WebSocketGateway, WebSocketServer, SubscribeMessage, MessageBody,
  ConnectedSocket, OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect,
  WsException,
} from '@nestjs/websockets';
import { Server, Socket } from 'socket.io';

@WebSocketGateway({
  namespace: '/tareas',                       // aísla este canal de otros
  cors: { origin: ['https://app.ejemplo.com'], credentials: true },
  transports: ['websocket'],                  // evita el fallback a long-polling
})
export class TareasGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
  private readonly log = new Logger(TareasGateway.name);

  @WebSocketServer() private readonly server!: Server;   // instancia de Socket.IO

  constructor(private readonly jwt: JwtService) {}

  /** 1. Una vez, al inicializar. Sitio idóneo para middleware de conexión. */
  afterInit(server: Server): void {
    // El middleware corre ANTES de handleConnection: rechaza aquí lo inválido
    // para no gastar recursos en sockets que no deberían existir.
    server.use(async (socket: Socket, next) => {
      try {
        const token = socket.handshake.auth?.token
          ?? (socket.handshake.headers.authorization ?? '').replace('Bearer ', '');
        if (!token) throw new Error('sin token');

        const payload = await this.jwt.verifyAsync(token);
        socket.data.user = { id: payload.sub, roles: payload.roles };
        next();
      } catch {
        // El cliente recibe un connect_error, no una conexión a medias.
        next(new Error('No autorizado'));
      }
    });
  }

  /** 2. Por cada cliente que se conecta (ya autenticado por el middleware). */
  async handleConnection(client: Socket): Promise<void> {
    const userId = client.data.user.id as string;

    // Sala personal: permite enviar a "este usuario" en todos sus dispositivos.
    await client.join(`user:${userId}`);
    this.log.log(`Conectado ${client.id} (usuario ${userId})`);
  }

  /** 3. Al desconectar. Socket.IO abandona las salas solo; libera TU estado. */
  handleDisconnect(client: Socket): void {
    this.presencia.marcarOffline(client.data.user?.id);
  }

  /** Mensaje entrante: el cliente pide seguir un proyecto. */
  @SubscribeMessage('proyecto:seguir')
  async seguir(
    @ConnectedSocket() client: Socket,
    @MessageBody() body: { proyectoId: string },
  ): Promise<{ ok: true }> {
    // AUTORIZACIÓN: quien se conecta no puede unirse a la sala que quiera.
    // Olvidar esta comprobación es el equivalente en WebSocket a un IDOR.
    const permitido = await this.acl.puedeVerProyecto(client.data.user.id, body.proyectoId);
    if (!permitido) throw new WsException('Sin acceso a ese proyecto');

    await client.join(`proyecto:${body.proyectoId}`);
    return { ok: true };            // se entrega como acknowledgement al cliente
  }

  /** API interna: la usan los servicios y los listeners de eventos. */
  notificarCambio(proyectoId: string, tarea: TareaDto): void {
    this.server.to(`proyecto:${proyectoId}`).emit('tarea:actualizada', tarea);
  }

  notificarUsuario(userId: string, payload: unknown): void {
    this.server.to(`user:${userId}`).emit('notificacion', payload);
  }
}

11.6.2 Escalado horizontal: el adaptador de Redis no es opcional

Un socket está atado al proceso que lo aceptó. Con una sola instancia todo funciona; en cuanto hay dos, la mitad de tus notificaciones desaparecen y el error es desconcertante porque «en local funciona».

  SIN ADAPTADOR (roto)                          CON ADAPTADOR DE REDIS (correcto)

  Ana ──ws──► [ Réplica A ]                     Ana ──ws──► [ Réplica A ]──┐
  Luis ─ws──► [ Réplica B ]                     Luis ─ws──► [ Réplica B ]──┤
                                                                          │ pub/sub
  POST /tareas/7  ──► Réplica A                 POST /tareas/7 ──► Réplica A│
     server.to('proyecto:1').emit(...)             emit(...) ──► publica ──►├──► REDIS
     └─► llega SOLO a los sockets de A                                      │
         Luis NO recibe nada                    Réplicas A y B reciben ◄────┘
                                                y entregan a SUS sockets
  Síntoma: "a veces llega y a veces no",        Resultado: entrega a todos los
  proporcional al número de réplicas            sockets de la sala, esté donde esté

import { IoAdapter } from '@nestjs/platform-socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';
import { ServerOptions } from 'socket.io';

export class RedisIoAdapter extends IoAdapter {
  private adapterConstructor!: ReturnType<typeof createAdapter>;

  async connectToRedis(url: string): Promise<void> {
    // Dos clientes: uno publica y otro se suscribe. Un cliente en modo
    // suscripción no puede ejecutar otros comandos, de ahí el duplicado.
    const pubClient = createClient({ url });
    const subClient = pubClient.duplicate();
    await Promise.all([pubClient.connect(), subClient.connect()]);
    this.adapterConstructor = createAdapter(pubClient, subClient);
  }

  createIOServer(port: number, options?: ServerOptions): unknown {
    const server = super.createIOServer(port, options);
    server.adapter(this.adapterConstructor);
    return server;
  }
}

// main.ts
const app = await NestFactory.create(AppModule);
const redisAdapter = new RedisIoAdapter(app);
await redisAdapter.connectToRedis(process.env.REDIS_URL!);
app.useWebSocketAdapter(redisAdapter);   // ANTES de app.listen()

11.6.3 Server-Sent Events: cuando media conexión basta

Muchas veces solo necesitas que el servidor empuje datos y el cliente no tiene nada que decir: progreso de un informe, contador de notificaciones, precios en directo. Para eso hay una solución mucho más simple que WebSocket, integrada en HTTP: Server-Sent Events. Es texto sobre una respuesta que no se cierra, con reconexión automática por parte del navegador.

import { Sse, MessageEvent } from '@nestjs/common';
import { interval, map, switchMap, takeWhile, from } from 'rxjs';

@Controller('informes')
export class InformesSseController {
  /** El método devuelve un Observable<MessageEvent>; Nest mantiene la conexión abierta. */
  @Sse(':id/progreso')
  @UseGuards(JwtAuthGuard)          // los guards funcionan igual que en HTTP normal
  progreso(@Param('id') id: string): Observable<MessageEvent> {
    return interval(1000).pipe(
      switchMap(() => from(this.cola.getJob(id))),
      map((job) => ({
        // 'data' es obligatorio; 'type' se corresponde con el nombre del evento
        // y 'id' permite al navegador reanudar con Last-Event-ID.
        type: 'progreso',
        data: { progreso: job?.progress ?? 0, estado: job ? 'activo' : 'desconocido' },
      })),
      takeWhile((e) => (e.data as { progreso: number }).progreso < 100, true),
    );
  }
}

11.6.4 Ejemplo completo: cambios de tarea en tiempo real

Unimos las piezas: el servicio de tareas emite un evento de dominio, un listener lo traduce a difusión por WebSocket y el cliente Angular lo consume como una señal. Nótese que el servicio de tareas no conoce el gateway: si mañana se sustituye Socket.IO, el dominio no cambia.

@Injectable()
export class RealtimeListener {
  constructor(private readonly gateway: TareasGateway) {}

  @OnEvent('tarea.actualizada', { async: true })
  async difundir(e: TareaActualizadaEvent) {
    // Enviamos un DTO, nunca la entidad: evita filtrar campos internos
    // y las referencias circulares del ORM al serializar.
    this.gateway.notificarCambio(e.proyectoId, {
      id: e.tareaId,
      estado: e.estado,
      actualizadaEn: e.fecha.toISOString(),
      // Marca de versión: el cliente descarta mensajes desordenados.
      version: e.version,
    });
  }

  @OnEvent('informe.listo', { async: true })
  async informe(e: { userId: string; url: string }) {
    this.gateway.notificarUsuario(e.userId, { tipo: 'informe', url: e.url });
  }
}

import { Injectable, signal, inject, DestroyRef } from '@angular/core';
import { io, Socket } from 'socket.io-client';

@Injectable({ providedIn: 'root' })
export class RealtimeService {
  private socket?: Socket;
  private readonly destroyRef = inject(DestroyRef);

  readonly conectado = signal(false);
  readonly ultimoCambio = signal<TareaDto | null>(null);

  conectar(token: string): void {
    this.socket = io('https://api.ejemplo.com/tareas', {
      transports: ['websocket'],
      auth: { token },              // llega a handshake.auth en el servidor
      reconnectionDelay: 1000,
      reconnectionDelayMax: 10_000, // retroceso: no martillees un servidor caído
    });

    this.socket.on('connect', () => this.conectado.set(true));
    this.socket.on('disconnect', () => this.conectado.set(false));
    this.socket.on('tarea:actualizada', (t: TareaDto) => this.ultimoCambio.set(t));

    // Token caducado o revocado: el servidor rechazó el handshake.
    this.socket.on('connect_error', (err) => {
      if (err.message === 'No autorizado') this.auth.renovarYReconectar();
    });

    // Liberar SIEMPRE: un socket huérfano sigue recibiendo y filtra memoria.
    this.destroyRef.onDestroy(() => this.socket?.disconnect());
  }

  seguirProyecto(proyectoId: string): Promise<void> {
    // El tercer argumento es el ack: la promesa se resuelve con lo que
    // devuelva el @SubscribeMessage del servidor.
    return new Promise((resolve, reject) => {
      this.socket?.emit('proyecto:seguir', { proyectoId }, (res: { ok?: boolean }) =>
        res?.ok ? resolve() : reject(new Error('No autorizado')),
      );
    });
  }
}

11.7 GraphQL

11.7.1 Qué problema resuelve y qué problemas trae

REST expone recursos con una forma fija decidida por el servidor. Eso genera tres fricciones cuando el cliente es una aplicación rica: over-fetching (la pantalla necesita el nombre y llegan cuarenta campos), under-fetching (necesita el nombre del autor, que no viene incluido) y múltiples viajes (para pintar una lista con autores y etiquetas hacen falta tres peticiones en cascada, con la latencia sumada).

GraphQL invierte el control: el cliente declara exactamente qué campos quiere, en una sola petición, y el servidor responde con esa forma. A cambio introduce cuatro problemas nuevos que hay que aceptar con los ojos abiertos: la caché HTTP deja de funcionar (todo es un POST a /graphql), el N+1 aparece de forma natural en los resolvers, la seguridad requiere limitar consultas maliciosas y la complejidad operativa sube (esquema, resolvers, herramientas, monitorización por operación).

11.7.2 Code-first con @nestjs/graphql

Hay dos enfoques: schema-first (escribes SDL y se generan los tipos) y code-first (escribes clases con decoradores y se genera el SDL). En un proyecto Nest, code-first es casi siempre mejor: una única fuente de verdad en TypeScript, sin desincronización entre esquema y código.

import { ObjectType, Field, ID, Int, registerEnumType } from '@nestjs/graphql';

export enum EstadoTarea { PENDIENTE = 'pendiente', HECHA = 'hecha' }
registerEnumType(EstadoTarea, { name: 'EstadoTarea' });

@ObjectType({ description: 'Tarea de un proyecto' })
export class Tarea {
  @Field(() => ID) id!: string;
  @Field() titulo!: string;
  // nullable explícito: en GraphQL todo es opcional salvo que digas lo contrario
  @Field({ nullable: true }) descripcion?: string;
  @Field(() => EstadoTarea) estado!: EstadoTarea;
  @Field(() => Int) comentariosTotal!: number;

  // El tipo se resuelve en un @ResolveField: no viaja en la consulta base.
  @Field(() => Usuario) propietario!: Usuario;
  @Field(() => [Etiqueta]) etiquetas!: Etiqueta[];
}

@InputType()
export class CrearTareaInput {
  @Field() @IsString() @Length(3, 120) titulo!: string;
  @Field(() => ID) @IsUUID() proyectoId!: string;
}

@Resolver(() => Tarea)
@UseGuards(GqlAuthGuard)
export class TareasResolver {
  constructor(private readonly servicio: TareasService) {}

  @Query(() => [Tarea], { name: 'tareas' })
  listar(@Args('estado', { nullable: true }) estado?: EstadoTarea) {
    return this.servicio.listar({ estado });
  }

  @Mutation(() => Tarea)
  crearTarea(
    @Args('input') input: CrearTareaInput,
    @CurrentUser() user: JwtUser,
  ) {
    return this.servicio.crear(input, user.id);
  }

  /** Campo calculado: solo se ejecuta si el cliente lo pide. */
  @ResolveField(() => Usuario)
  propietario(@Parent() tarea: Tarea, @Context() ctx: GqlContext) {
    // OJO: aquí nace el N+1 (ver 11.7.3). Con DataLoader, no.
    return ctx.loaders.usuarios.load(tarea.propietarioId);
  }

  @Subscription(() => Tarea, {
    // Filtro en el servidor: no envíes a quien no le corresponde.
    filter: (payload, variables) =>
      payload.tareaActualizada.proyectoId === variables.proyectoId,
  })
  tareaActualizada(@Args('proyectoId') proyectoId: string) {
    return this.pubSub.asyncIterableIterator('tareaActualizada');
  }
}

11.7.3 El problema N+1 y DataLoader

Un resolver de campo se ejecuta una vez por elemento del resultado padre. Si la consulta devuelve 50 tareas y cada una resuelve su propietario con una consulta, se ejecutan 51 consultas: una para la lista y 50 para los autores. Con dos niveles de anidamiento, cientos. La API parece correcta en desarrollo con tres filas y se derrumba en producción.

  SIN DATALOADER (N+1)                          CON DATALOADER (2 consultas)
  query { tareas { titulo propietario { nombre } } }

  1  SELECT * FROM tarea LIMIT 50                1  SELECT * FROM tarea LIMIT 50
  2  SELECT * FROM usuario WHERE id = 'a'        │  ┌─ load('a') ─┐
  3  SELECT * FROM usuario WHERE id = 'b'        │  ├─ load('b') ─┤ mismo tick del
  4  SELECT * FROM usuario WHERE id = 'a'  ◄──   │  ├─ load('a') ─┤ event loop: se
  …  … 50 consultas, muchas repetidas    dup     │  └─ load('c') ─┘ agrupan y deduplican
  51 SELECT * FROM usuario WHERE id = 'z'        2  SELECT * FROM usuario WHERE id IN ('a','b','c')

  Latencia: 51 × RTT                            Latencia: 2 × RTT

DataLoader resuelve dos cosas a la vez: agrupa (batching) las claves solicitadas en el mismo turno del event loop en una única consulta con IN, y deduplica con una caché por petición. Es imprescindible que los loaders se creen por petición: si son singletons, la caché mezcla datos entre usuarios y entre momentos, exactamente el problema de la sección 11.2.

import DataLoader from 'dataloader';

export function crearLoaders(em: EntityManager) {
  return {
    usuarios: new DataLoader<string, Usuario>(async (ids) => {
      const filas = await em.find(Usuario, { id: { $in: [...ids] } });
      const porId = new Map(filas.map((u) => [u.id, u]));
      // CONTRATO: el array devuelto debe tener el MISMO tamaño y orden
      // que las claves recibidas. Si no, DataLoader devuelve valores cruzados.
      return ids.map((id) => porId.get(id) ?? new Error(`Usuario ${id} no existe`));
    }),

    // Relación uno-a-muchos: agrupa hijos por clave del padre.
    etiquetasPorTarea: new DataLoader<string, Etiqueta[]>(async (tareaIds) => {
      const filas = await em.find(TareaEtiqueta, { tarea: { $in: [...tareaIds] } },
        { populate: ['etiqueta'] });
      const grupos = new Map<string, Etiqueta[]>();
      for (const f of filas) {
        const lista = grupos.get(f.tarea.id) ?? [];
        lista.push(f.etiqueta);
        grupos.set(f.tarea.id, lista);
      }
      // Sin resultados debe devolver [], no undefined.
      return tareaIds.map((id) => grupos.get(id) ?? []);
    }),
  };
}
export type GqlContext = { req: Request; loaders: ReturnType<typeof crearLoaders> };

GraphQLModule.forRoot<ApolloDriverConfig>({
  driver: ApolloDriver,
  autoSchemaFile: 'schema.gql',
  sortSchema: true,
  // Una fábrica de contexto POR PETICIÓN: aquí nacen y mueren los loaders.
  context: ({ req }: { req: Request }) => ({ req, loaders: crearLoaders(orm.em.fork()) }),
  introspection: process.env.NODE_ENV !== 'production',
  validationRules: [depthLimit(7)],
});

11.7.4 Seguridad: profundidad, complejidad e introspección

• Profundidad. Un esquema con relaciones cíclicas permite tarea { proyecto { tareas { proyecto { … } } } } hasta agotar el servidor. Limita con depthLimit(n) en validationRules; entre 7 y 10 suele bastar.
• Complejidad. La profundidad no captura el coste de primeros: 10000. Con graphql-query-complexity se asigna un coste a cada campo y se rechaza la consulta si el total excede un máximo, antes de ejecutar nada.
• Paginación obligatoria. Ningún campo de lista debe poder devolver todo. Impón un limit máximo en el propio resolver, no confíes en el argumento del cliente.
• Introspección desactivada en producción. Es la que permite descargar el esquema completo y facilita la búsqueda de campos olvidados. Desactiva también el explorador gráfico.
• Consultas persistidas (allowlist). El nivel máximo de protección: solo se aceptan consultas previamente registradas, identificadas por su hash. Elimina de golpe todo el vector de consultas arbitrarias.
• Autorización por campo. Un guard a nivel de resolver no protege un @ResolveField al que se llega por otro camino del grafo. Autoriza en la capa de servicio o con directivas, no solo en la entrada.

11.8 Microservicios

11.8.1 Cuándo NO usarlos, que es casi siempre

Los microservicios resuelven un problema organizativo: permitir que muchos equipos despliegen sin coordinarse. No resuelven problemas de rendimiento —añadir latencia de red nunca hace nada más rápido— ni de calidad de código: un monolito mal estructurado se convierte en varios servicios mal estructurados, ahora con red en medio.

11.8.2 Transportes de @nestjs/microservices

Nest abstrae el transporte: el mismo @MessagePattern funciona sobre TCP o sobre Kafka. La elección, sin embargo, no es un detalle: determina las garantías de entrega.

11.8.3 @MessagePattern, @EventPattern y ClientProxy

@Controller()
export class FacturasController {
  /** Petición-respuesta: el emisor ESPERA un valor. */
  @MessagePattern({ cmd: 'facturas.obtener' })
  async obtener(@Payload() data: { id: string }): Promise<FacturaDto> {
    const f = await this.servicio.buscar(data.id);
    // Los errores deben viajar como RpcException para que el
    // cliente reciba un error estructurado y no un objeto vacío.
    if (!f) throw new RpcException({ code: 'NOT_FOUND', message: 'No existe' });
    return f;
  }

  /** Evento: notificación sin respuesta. Puede llegar más de una vez. */
  @EventPattern('pedido.confirmado')
  async alConfirmar(@Payload() e: PedidoConfirmado, @Ctx() ctx: RmqContext) {
    // Con noAck: false, TÚ decides cuándo se confirma el mensaje.
    const canal = ctx.getChannelRef();
    const mensaje = ctx.getMessage();
    try {
      await this.servicio.emitirFactura(e.pedidoId);   // idempotente
      canal.ack(mensaje);
    } catch (err) {
      // requeue en false: va a la dead letter queue en lugar de
      // volver a la cola para siempre (bucle infinito de veneno).
      canal.nack(mensaje, false, false);
    }
  }
}

@Module({
  imports: [ClientsModule.register([{
    name: 'FACTURAS',
    transport: Transport.RMQ,
    options: {
      urls: [process.env.RMQ_URL!],
      queue: 'facturas',
      queueOptions: { durable: true },
      noAck: false,
    },
  }])],
})
export class PedidosModule {}

@Injectable()
export class PedidosService {
  constructor(@Inject('FACTURAS') private readonly facturas: ClientProxy) {}

  async detalle(id: string) {
    // send() devuelve un Observable FRÍO: sin subscribe no se envía nada.
    // Un timeout SIEMPRE: sin él, un servicio lento cuelga esta petición
    // hasta agotar el pool de conexiones del gateway.
    return firstValueFrom(
      this.facturas.send({ cmd: 'facturas.obtener' }, { id }).pipe(
        timeout(3_000),
        retry({ count: 2, delay: 300 }),
        catchError((err) => {
          if (err instanceof TimeoutError) {
            // Degradar es mejor que fallar: devuelve lo que tengas.
            return of({ id, estado: 'desconocido', degradado: true });
          }
          return throwError(() => new ServiceUnavailableException('Facturas no disponible'));
        }),
      ),
    );
  }

  notificar(e: PedidoConfirmado) {
    // emit() es fuego y olvido: no esperes confirmación de proceso.
    this.facturas.emit('pedido.confirmado', e);
  }
}

11.8.4 Aplicaciones híbridas

Un mismo proceso puede atender HTTP y escuchar mensajes. Es la forma habitual de migrar sin big bang: el servicio conserva su API REST y empieza a consumir eventos.

const app = await NestFactory.create(AppModule);
app.useGlobalPipes(new ValidationPipe({ whitelist: true, transform: true }));

// inheritAppConfig hace que pipes, filtros e interceptores globales
// se apliquen TAMBIÉN a los mensajes; sin él, los mensajes entran sin validar.
app.connectMicroservice<MicroserviceOptions>(
  { transport: Transport.RMQ, options: { urls: [process.env.RMQ_URL!], queue: 'pedidos' } },
  { inheritAppConfig: true },
);

await app.startAllMicroservices();
await app.listen(3000);   // el mismo proceso: mismo pool de BD, mismo event loop

11.8.5 Patrones distribuidos imprescindibles

• API Gateway. Un único punto de entrada que autentica, limita, enruta y compone respuestas. Evita que el cliente conozca la topología interna. Riesgo: convertirse en un monolito con lógica de negocio; debe quedarse en orquestación y traducción.
• Outbox transaccional. El problema: no puedes escribir en la base de datos y publicar en el broker atómicamente; si publicas antes del commit, anuncias algo que puede no ocurrir, y si publicas después, puedes perder el mensaje. Solución: en la misma transacción inserta la fila y un registro en la tabla outbox; un proceso aparte lee esa tabla y publica, marcando lo enviado. Así el mensaje se publica «al menos una vez» y siempre coherente con el estado. Es el patrón que hace innecesario el commit en dos fases.
• Idempotencia. Igual que en las colas (11.3.5): identificador de mensaje, tabla de procesados y operaciones idempotentes. En un sistema distribuido es obligatoria, no recomendable.
• Reintentos con retroceso exponencial y jitter. Reintentar de inmediato y a la vez es la receta de la manada atronadora: el servicio que se está recuperando vuelve a caer.
• Circuit breaker. Si un servicio falla el 50 % de las últimas 20 llamadas, el interruptor «abre» y las siguientes fallan de inmediato sin llamar. Pasado un tiempo entra en «semiabierto» y prueba con una llamada. Evita gastar hilos y timeouts en algo que se sabe roto y da margen a la recuperación. Suele implementarse con opossum envolviendo el cliente.
• Dead letter queue. Los mensajes que fallan repetidamente salen del flujo a una cola aparte. Sin DLQ, un solo mensaje «venenoso» bloquea el consumo indefinidamente. Con DLQ hay que vigilarla: una DLQ que nadie mira es una papelera.
• Saga. La sustituta de la transacción distribuida: una secuencia de transacciones locales, cada una con su compensación. No hay rollback global; hay operaciones inversas.

  SAGA COREOGRAFIADA (cada servicio reacciona a eventos)

  PEDIDOS            PAGOS               ALMACÉN            ENVÍOS
     │ pedido.creado    │                   │                  │
     ├─────────────────►│ cobra             │                  │
     │                  ├──pago.ok─────────►│ reserva stock    │
     │                  │                   ├──stock.ok───────►│ crea envío
     │                  │                   │                  ├──envio.ok──┐
     │◄─────────────────┴───────────────────┴──────────────────┴────────────┘
     │ pedido.completado                              (sin coordinador central)

  COMPENSACIÓN (el almacén no tiene stock)
     │                  │                   │ stock.fallido    │
     │                  │◄──────────────────┤                  │
     │                  ├─ reembolsa (compensación de "cobra")  │
     │◄─────────────────┤ pago.reembolsado                     │
     ├─ marca pedido como cancelado                            │

  SAGA ORQUESTADA (un coordinador dirige)          ┌─────────────────┐
                                                   │  ORQUESTADOR    │
   1. cobrar ──► 2. reservar ──► 3. enviar         │  (máquina de    │
        ▲              │                           │   estados)      │
        └──── compensar si falla ◄─────────────────┤  guarda el      │
                                                   │  estado de cada │
   Coreografía: menos acoplamiento, flujo          │  saga en curso  │
   difícil de ver. Orquestación: flujo             └─────────────────┘
   explícito y depurable, un componente más.

La elección entre coreografía y orquestación no es estética: decide dónde vive la complejidad. En la coreografía cada servicio conoce solo los eventos que consume y los que publica, el acoplamiento es mínimo y, en cambio, el flujo completo no está escrito en ningún sitio: para saber qué ocurre al confirmar un pedido hay que leer cuatro repositorios distintos y confiar en que la documentación esté al día. En la orquestación hay una máquina de estados explícita, consultable y depurable, al precio de un componente más que desplegar, escalar y hacer tolerante a fallos. Regla práctica: hasta tres pasos, coreografía; a partir de cuatro pasos con compensaciones, plazos y estados intermedios visibles para el usuario, orquestación.

11.9 Errores comunes y cómo solucionarlos

Los fallos de este capítulo comparten una firma característica: casi ninguno se manifiesta en desarrollo. Con una réplica, un usuario y tres filas en la base de datos, la caché es coherente, el cron se ejecuta una vez, todos los sockets están en el mismo proceso y no hay N+1 apreciable. Aparecen el día del despliegue a producción, y por eso conviene conocerlos antes de sufrirlos.

11.10 Buenas y malas prácticas

11.11 Preguntas frecuentes

11.12 Ejercicios

Los ejercicios están pensados para hacerse sobre el proyecto de tareas que se viene construyendo desde el capítulo 9. Necesitarás un Redis local; la forma más rápida es docker run -p 6379:6379 redis:7-alpine. Cuando un ejercicio pida «medir», mide de verdad: casi todas las decisiones de este capítulo se justifican con números y se estropean con intuiciones.

// src/pagos/pagos.processor.ts
@Processor('pagos', { concurrency: 3 })
export class PagosProcessor extends WorkerHost {
  constructor(
    private readonly em: EntityManager,
    private readonly pasarela: PasarelaService,
    @InjectQueue('correos') private readonly correos: Queue,
  ) { super(); }

  async process(job: Job<{ pedidoId: string }>): Promise<void> {
    // El worker vive fuera de una petición: sin fork(), dos trabajos
    // concurrentes compartirían Identity Map y Unit of Work.
    const em = this.em.fork();

    // 1 · Barrera barata. Cubre el caso normal (el reintento llega segundos
    //     después) sin molestar a la pasarela. NO es suficiente por sí sola:
    //     dos workers concurrentes pueden leer 'pendiente' los dos.
    const pedido = await em.findOneOrFail(Pedido, { id: job.data.pedidoId });
    if (pedido.estado === 'pagado') return;

    // 2 · Clave de idempotencia ESTABLE: la misma en todos los reintentos del
    //     mismo pedido. Nunca uses job.attemptsMade ni Date.now() aquí: si la
    //     clave cambia, la pasarela crea un cargo nuevo y has cobrado dos veces.
    const cargo = await this.pasarela.cobrar(pedido.total, pedido.tarjeta, {
      idempotencyKey: `pedido:${pedido.id}:cobro`,
    });

    // 3 · Transición atómica en la base de datos: la carrera la resuelve el
    //     motor, no nuestro código. Solo una de las ejecuciones ve 1 fila.
    const afectadas = await em.nativeUpdate(
      Pedido,
      { id: pedido.id, estado: 'pendiente' },
      { estado: 'pagado', cargoId: cargo.id, pagadoEn: new Date() },
    );
    if (afectadas === 0) return;   // otro ganó: no dupliques los efectos siguientes

    // 4 · El recibo es OTRO trabajo, con su propio jobId estable: si el envío
    //     falla, se reintenta el correo y no el cobro.
    await this.correos.add('recibo', { pedidoId: pedido.id }, {
      jobId: `recibo:${pedido.id}`,
      attempts: 5,
      backoff: { type: 'exponential', delay: 5_000 },
    });
  }
}

// test/pagos.processor.spec.ts
describe('PagosProcessor · idempotencia', () => {
  it('ejecutar el MISMO trabajo dos veces cobra una sola vez', async () => {
    const pedido = await crearPedido({ total: 4990, estado: 'pendiente' });
    const job = { name: 'cobrar', data: { pedidoId: pedido.id } } as Job<DatosCobro>;

    await processor.process(job);
    await processor.process(job);   // reentrega: es EXACTAMENTE lo que hará la cola

    expect(pasarela.cobrar).toHaveBeenCalledTimes(1);          // barrera de estado
    expect(await contarTrabajos('correos', `recibo:${pedido.id}`)).toBe(1);
    expect((await recargar(pedido)).estado).toBe('pagado');
  });

  it('dos ejecuciones CONCURRENTES no duplican efectos', async () => {
    const pedido = await crearPedido({ total: 4990, estado: 'pendiente' });
    const job = { name: 'cobrar', data: { pedidoId: pedido.id } } as Job<DatosCobro>;

    // Aquí la comprobación de estado NO sirve: las dos leen 'pendiente'.
    // Quien salva la situación es la escritura condicional del paso 3, y la
    // clave de idempotencia, que hace que la pasarela devuelva el MISMO cargo.
    await Promise.all([processor.process(job), processor.process(job)]);

    expect(await contarTrabajos('correos', `recibo:${pedido.id}`)).toBe(1);
    expect(cargosCreadosEn(pasarela)).toBe(1);
  });
});

// src/common/cache/cache-robusta.service.ts
interface Envoltorio<T> { valor: T; frescoHasta: number; }

@Injectable()
export class CacheRobustaService {
  private readonly log = new Logger(CacheRobustaService.name);
  /** Promesas en curso por clave: elimina la estampida DENTRO de este proceso. */
  private readonly enVuelo = new Map<string, Promise<unknown>>();

  constructor(
    @Inject(CACHE_MANAGER) private readonly cache: Cache,
    @Inject('REDIS') private readonly redis: Redis,
  ) {}

  /**
   * frescoMs   → durante este tiempo se sirve sin más.
   * toleradoMs → pasado el frescor, se sigue sirviendo el valor viejo
   *              mientras se recalcula por detrás (stale-while-revalidate).
   */
  async obtener<T>(
    clave: string, frescoMs: number, toleradoMs: number, calcular: () => Promise<T>,
  ): Promise<T> {
    const guardado = await this.leer<Envoltorio<T>>(clave);

    // 1 · Fresco: camino rápido, sin sorpresas.
    if (guardado && Date.now() < guardado.frescoHasta) return guardado.valor;

    // 2 · Rancio pero tolerable: respondemos YA y refrescamos en segundo plano.
    //     El usuario nunca espera y la base de datos nunca recibe la avalancha.
    if (guardado) {
      void this.refrescar(clave, frescoMs, toleradoMs, calcular)
        .catch((e) => this.log.warn(`Refresco en segundo plano falló: ${e}`));
      return guardado.valor;
    }

    // 3 · Caché vacía de verdad (arranque en frío): colapsamos las llamadas
    //     concurrentes de este proceso en una sola ejecución.
    return this.singleFlight(clave, () =>
      this.refrescar(clave, frescoMs, toleradoMs, calcular));
  }

  private singleFlight<T>(clave: string, fn: () => Promise<T>): Promise<T> {
    const existente = this.enVuelo.get(clave) as Promise<T> | undefined;
    if (existente) return existente;
    // Se registra ANTES del primer await: si no, dos llamadas simultáneas
    // pasarían de largo y volveríamos a tener dos consultas.
    const promesa = fn().finally(() => this.enVuelo.delete(clave));
    this.enVuelo.set(clave, promesa);
    return promesa;
  }

  private async refrescar<T>(
    clave: string, frescoMs: number, toleradoMs: number, calcular: () => Promise<T>,
  ): Promise<T> {
    // Cerrojo distribuido: SET con NX y PX es una única operación atómica.
    // El PX evita cerrojos huérfanos si el proceso muere sin liberarlo.
    const conseguido = await this.redis.set(`lock:${clave}`, '1', 'PX', 10_000, 'NX');
    if (!conseguido) {
      // Otra réplica está recalculando: devolvemos lo que haya antes de insistir.
      const actual = await this.leer<Envoltorio<T>>(clave);
      if (actual) return actual.valor;
    }

    try {
      const valor = await calcular();
      // Jitter sobre el frescor: miles de claves no deben caducar en el
      // mismo segundo (típico tras un despliegue que precalienta la caché).
      const jitter = Math.floor(Math.random() * frescoMs * 0.1);
      const envoltorio: Envoltorio<T> = { valor, frescoHasta: Date.now() + frescoMs + jitter };
      // TTL real = frescor + tolerancia: la entrada sobrevive rancia a propósito.
      await this.cache.set(clave, envoltorio, frescoMs + jitter + toleradoMs);
      return valor;
    } finally {
      if (conseguido) await this.redis.del(`lock:${clave}`);
    }
  }

  /** La caché es opcional; la aplicación no. Un Redis caído degrada, no rompe. */
  private async leer<T>(clave: string): Promise<T | null> {
    try {
      return (await this.cache.get<T>(clave)) ?? null;
    } catch (e) {
      this.log.warn(`Caché no disponible en lectura (${clave}): ${e}`);
      return null;
    }
  }
}

// Uso: 60 s de frescor y 5 min de tolerancia. En la práctica, la clave
// caliente no vuelve a provocar una consulta síncrona nunca más.
const resumen = await this.cacheRobusta.obtener(
  `informe:${proyectoId}:${mes}`, 60_000, 300_000,
  () => this.calcularResumen(proyectoId, mes),
);

// src/outbox/outbox-message.entity.ts
@Entity({ tableName: 'outbox' })
@Index({ properties: ['publicadoEn', 'creadoEn'] })   // el índice del publicador
export class OutboxMessage {
  @PrimaryKey() id: string = randomUUID();
  @Property() tipo!: string;                       // 'pedido.confirmado'
  @Property({ type: 'json' }) payload!: Record<string, unknown>;
  @Property() creadoEn: Date = new Date();
  @Property({ nullable: true }) publicadoEn?: Date; // null = pendiente
  @Property({ default: 0 }) intentos!: number;
}

// src/pedidos/pedidos.service.ts
async confirmar(pedidoId: string): Promise<void> {
  await this.em.transactional(async (em) => {
    const pedido = await em.findOneOrFail(Pedido, { id: pedidoId },
      { lockMode: LockMode.PESSIMISTIC_WRITE });
    if (pedido.estado !== 'pendiente') return;      // idempotencia del caso de uso
    pedido.estado = 'confirmado';

    // LA CLAVE DEL PATRÓN: el mensaje se guarda en la MISMA transacción que
    // el cambio de estado. O se confirman los dos, o ninguno. Desaparece la
    // ventana en la que anunciamos algo que aún puede no ocurrir (publicar
    // antes del commit) o cambiamos el estado sin avisar a nadie (publicar
    // después y morir en medio).
    em.persist(em.create(OutboxMessage, {
      tipo: 'pedido.confirmado',
      payload: { pedidoId: pedido.id, total: pedido.total, version: pedido.version },
    }));
  });
}

// src/outbox/outbox.publisher.ts
@Injectable()
export class OutboxPublisher {
  private readonly log = new Logger(OutboxPublisher.name);

  constructor(
    private readonly em: EntityManager,
    @Inject('REDIS') private readonly redis: Redis,
    @Inject('BROKER') private readonly broker: ClientProxy,
  ) {}

  @Cron(CronExpression.EVERY_5_SECONDS)
  async publicar(): Promise<void> {
    // Sin cerrojo, las N réplicas leerían las mismas filas y publicarían N veces.
    // No es catastrófico (el consumidor deduplica), pero es ruido evitable.
    const turno = await this.redis.set('lock:outbox', '1', 'PX', 10_000, 'NX');
    if (!turno) return;

    const em = this.em.fork();
    const pendientes = await em.find(OutboxMessage, { publicadoEn: null },
      { orderBy: { creadoEn: 'ASC' }, limit: 100 });

    for (const msg of pendientes) {
      try {
        // El id del mensaje viaja al consumidor: es su clave de deduplicación.
        this.broker.emit(msg.tipo, { ...msg.payload, mensajeId: msg.id });
        msg.publicadoEn = new Date();
      } catch (e) {
        // No marcamos como publicado: la próxima pasada lo reintenta.
        msg.intentos += 1;
        this.log.warn(`Outbox ${msg.id} falló (intento ${msg.intentos}): ${e}`);
        if (msg.intentos > 20) this.alertas.avisar(msg);   // algo va mal de verdad
      }
      await em.flush();
    }
  }

  /** La tabla outbox crece: purga lo publicado hace días o se convierte en un lastre. */
  @Cron(CronExpression.EVERY_DAY_AT_4AM)
  async purgar(): Promise<void> {
    const limite = new Date(Date.now() - 7 * 24 * 3_600_000);
    await this.em.nativeDelete(OutboxMessage, { publicadoEn: { $lt: limite } });
  }
}

// src/facturas/facturas.consumer.ts · el otro extremo del contrato
@EventPattern('pedido.confirmado')
async alConfirmar(@Payload() e: PedidoConfirmado & { mensajeId: string }) {
  try {
    // Insertar primero, procesar después: la clave primaria (tipo, mensajeId)
    // convierte el duplicado en una violación de unicidad, no en un efecto doble.
    await this.em.insert(MensajeProcesado, {
      tipo: 'pedido.confirmado', mensajeId: e.mensajeId, procesadoEn: new Date(),
    });
  } catch (err) {
    if (esViolacionDeUnicidad(err)) return;   // ya lo hicimos: nada que hacer
    throw err;
  }
  await this.facturas.emitir(e.pedidoId);
}

11.13 Resumen del capítulo

11.14 Recursos adicionales

• NestJS · Caching — la referencia para @nestjs/cache-manager: registro del módulo, almacenes, interceptor automático y personalización de la clave. Consúltala siempre en la versión que tengas instalada, porque las opciones de almacén y las unidades del TTL han cambiado entre versiones mayores.
• NestJS · Queues — integración con BullMQ: @Processor, WorkerHost, eventos del worker, colas con flujos y separación de productores y consumidores.
• BullMQ · documentación oficial — el detalle que la documentación de Nest no cubre: estados de un trabajo, trabajos atascados, rate limiting, job schedulers, deduplicación, flujos padre-hijo y guía de buenas prácticas de producción.
• NestJS · Task Scheduling — @Cron, @Interval, @Timeout y el SchedulerRegistry para programación dinámica, con la sintaxis exacta de las expresiones y las zonas horarias.
• Redis · documentación — comandos, expiración de claves, políticas de maxmemory, contadores atómicos y patrones de cerrojo distribuido. Es la pieza que sostiene la caché, las colas, el limitador de peticiones y la difusión entre gateways de este capítulo.
• NestJS · WebSockets Gateways — decoradores, ciclo de vida del gateway, uso de pipes, guards y filtros en mensajes, y adaptadores para sustituir Socket.IO o escalar horizontalmente.
• Socket.IO · Redis adapter — cómo funciona la propagación por pub/sub entre instancias, qué operaciones pasan a ser asíncronas y la comparación con el resto de adaptadores disponibles.
• MDN · Using server-sent events — el formato del flujo, EventSource, reconexión automática y Last-Event-ID: todo lo que necesitas para decidir si te basta SSE en lugar de WebSocket.
• NestJS · GraphQL — enfoque code-first, resolvers, @ResolveField, suscripciones, complejidad de consultas y la integración con el driver de Apollo.
• Apollo Server · documentación — el servidor que hay debajo del driver por defecto: plugins, formato de errores, consultas persistidas, caché de respuestas y métricas por operación.
• DataLoader — la librería que resuelve el N+1, con el contrato exacto de la función de lote (mismo tamaño y orden que las claves) y las advertencias sobre el ámbito de su caché.
• NestJS · Microservices — @MessagePattern, @EventPattern, ClientProxy, aplicaciones híbridas y la documentación específica de cada transporte (Redis, NATS, MQTT, RabbitMQ, Kafka y gRPC).
• Chris Richardson · patrón Saga — la descripción canónica de la alternativa a las transacciones distribuidas, con las variantes coreografiada y orquestada y sus contrapartidas. En el mismo catálogo está el outbox transaccional, imprescindible para publicar sin perder ni inventar mensajes.
• Opossum — implementación de circuit breaker para Node con métricas y respuestas de reserva, la que se usa habitualmente para envolver clientes de servicios remotos.
• RFC 9111 · HTTP Caching — la especificación vigente de Cache-Control, ETag y validación condicional: la capa de caché que suele estar sin explotar antes de montar Redis.