17. Transacciones, migraciones, filtros y rendimiento

17.1 Qué vas a poder hacer al terminar

• Explicar qué garantiza ACID y, sobre todo, qué no garantiza, con ejemplos de una aplicación real.
• Elegir con criterio entre la transacción implícita del flush(), em.transactional(), el control manual y una transacción por petición HTTP, sabiendo qué riesgo asume cada opción.
• Reconocer y evitar el error más frecuente del ORM: usar el EntityManager externo dentro del callback de em.transactional().
• Razonar sobre niveles de aislamiento a partir de la anomalía que quieres evitar, no por copiar una línea de configuración, y conocer los valores por defecto de PostgreSQL y MySQL.
• Implementar bloqueo optimista y bloqueo pesimista, traducir mentalmente cada uno al SQL que genera y decidir cuál corresponde a cada caso de uso.
• Construir una cola de trabajos en la propia base de datos con SKIP LOCKED sin que dos trabajadores procesen el mismo mensaje.
• Sustituir updateSchema por migraciones versionadas, escribir un down que funcione y aplicar el patrón expand/contract para desplegar sin tiempo de inactividad.
• Ejecutar migraciones en CI/CD con bloqueo, y saber qué hacer cuando una falla a medias.
• Poblar entornos con Seeder y Factory, y anonimizar datos de producción para entornos inferiores.
• Aplicar filtros globales para borrado lógico y multi-inquilino, y explicar por qué la seguridad nunca debe descansar solo en un filtro.
• Implementar borrado lógico completo, incluida la interacción con restricciones únicas y con el derecho de supresión de datos personales.
• Usar eventos y suscriptores para auditoría sin esconder lógica de negocio en ellos.
• Diagnosticar y corregir los cuatro problemas de rendimiento clásicos del ORM: memoria creciente, pool agotado, arranque lento y escrituras masivas una a una.
• Comparar las tres estrategias de multi-tenancy y elegir la que corresponde al tamaño de tu producto.

17.2 Transacciones: fundamentos

Una transacción es una secuencia de operaciones que la base de datos trata como una unidad indivisible: o se aplican todas o no se aplica ninguna. No es una característica del ORM, es una característica del motor de base de datos; el ORM únicamente decide dónde empieza y dónde acaba.

17.2.1 ACID con ejemplos concretos

17.2.2 Qué ocurre si no usas transacciones

El ejemplo canónico es la transferencia entre dos cuentas. Sin transacción, cada instrucción se confirma por separado (en autocommit) y basta con que el proceso muera entre las dos para que el dinero desaparezca.

async transferir(origenId: string, destinoId: string, importe: number) {
  const origen = await this.em.findOneOrFail(Cuenta, origenId);
  origen.saldo -= importe;
  await this.em.flush();          // COMMIT nº 1: el dinero ya salió
  // Si el proceso muere aquí, o el destino no existe,
  // o la red se corta: el importe se ha evaporado.
  const destino = await this.em.findOneOrFail(Cuenta, destinoId);
  destino.saldo += importe;
  await this.em.flush();          // COMMIT nº 2 (puede no llegar nunca)
}

async transferir(origenId: string, destinoId: string, importe: number) {
  // Un único flush dentro de una única transacción:
  // ambos UPDATE viajan juntos o no viajan.
  await this.em.transactional(async (em) => {
    const origen  = await em.findOneOrFail(Cuenta, origenId);
    const destino = await em.findOneOrFail(Cuenta, destinoId);
    if (origen.saldo < importe) {
      throw new SaldoInsuficienteError(origenId);   // provoca ROLLBACK
    }
    origen.saldo  -= importe;
    destino.saldo += importe;
  });
}

El segundo ejemplo, menos vistoso pero mucho más frecuente, es la creación de un agregado con hijos: un pedido con sus líneas. Sin transacción puedes acabar con cabeceras sin líneas (pedidos fantasma que aparecen en los informes con importe cero) o con líneas huérfanas si la clave foránea es NULL-able. El resultado no es un error visible, es un dato sucio que alguien descubrirá tres meses después cuadrando contabilidad.

BEGIN;
INSERT INTO "pedido" ("cliente_id", "total", "creado_en")
  VALUES (7, 44.80, now()) RETURNING "id";                     -- id = 101
INSERT INTO "pedido_linea" ("pedido_id", "producto_id", "cantidad", "precio")
  VALUES (101, 3, 2, 19.90), (101, 8, 1, 5.00);                -- inserción por lotes
UPDATE "producto" SET "stock" = "stock" - 2 WHERE "id" = 3;
COMMIT;

17.3 Transacciones en MikroORM

17.3.1 La transacción implícita del flush()

MikroORM no escribe nada en la base de datos hasta que llamas a em.flush(). En ese momento el Unit of Work calcula los change sets (qué insertar, qué actualizar, qué borrar), los ordena por dependencias y los ejecuta dentro de una transacción propia si no hay ninguna activa. Es decir: un flush() ya es atómico por sí mismo.

  em.persist(a); em.persist(b); entidadC.nombre = 'x';   ← nada viaja a la BD todavía
                          │
                          ▼  em.flush()
      ┌─────────────────────────────────────────────────────┐
      │ 1. beforeFlush                                      │  (puedes aún modificar entidades)
      │ 2. computeChangeSets()  ← diff contra la instantánea │
      │ 3. onFlush              ← change sets ya calculados  │
      ├─────────────────────────────────────────────────────┤
      │ 4. BEGIN            (solo si no había transacción)   │
      │ 5. INSERT / UPDATE / DELETE ordenados por topología  │
      │ 6. COMMIT  ó  ROLLBACK si algo lanza                 │
      ├─────────────────────────────────────────────────────┤
      │ 7. afterFlush                                       │
      └─────────────────────────────────────────────────────┘
                          ▼
      Identity Map intacto · instantáneas actualizadas al nuevo estado

La consecuencia práctica es importante: lo que determina la unidad atómica no son tus llamadas a persist, es dónde pones el flush. Dos flush() seguidos son dos transacciones separadas, con todo lo que eso implica.

17.3.2 em.transactional(): el EM forkeado

em.transactional(callback) abre una transacción, crea un fork del EntityManager (un contexto nuevo con su propio Identity Map y su propio Unit of Work) y lo pasa como argumento al callback. Al terminar hace flush() y COMMIT; si el callback lanza, hace ROLLBACK y propaga la excepción. Devuelve lo que devuelva el callback.

import { EntityManager } from '@mikro-orm/postgresql';
async crearPedido(dto: CrearPedidoDto): Promise<string> {
  // El valor devuelto por el callback es el valor devuelto por transactional.
  const id = await this.em.transactional(async (em) => {
    const cliente = await em.findOneOrFail(Cliente, dto.clienteId);
    const pedido  = em.create(Pedido, { cliente, estado: 'PENDIENTE' });
    for (const linea of dto.lineas) {
      const producto = await em.findOneOrFail(Producto, linea.productoId);
      // Cualquier excepción provoca el ROLLBACK automático de todo el bloque.
      if (producto.stock < linea.cantidad) throw new StockInsuficienteError(producto.sku);
      producto.stock -= linea.cantidad;
      em.create(PedidoLinea, { pedido, producto, cantidad: linea.cantidad, precio: producto.precio });
    }
    // No hace falta llamar a em.flush(): transactional lo hace al final.
    return pedido.id;
  });
  return id;
}

El error clásico es ignorar el parámetro y seguir usando el em del servicio. Las operaciones lanzadas contra el EM externo no pertenecen a la transacción: viven en otro contexto, con otro Unit of Work y, dependiendo del driver y del pool, con otra conexión. El resultado es un rollback que no deshace la mitad del trabajo o, peor, un bloqueo mutuo entre dos conexiones del mismo proceso.

async emitir(id: string) {
  await this.em.transactional(async (em) => {
    const f = await em.findOneOrFail(Factura, id);
    f.estado = 'EMITIDA';
    // this.em NO es el em de la transacción:
    // este flush se confirma por separado y
    // sobrevive al ROLLBACK de la transacción.
    const contador = await this.em.findOneOrFail(Contador, 'facturas');
    contador.siguiente++;
    await this.em.flush();
    throw new Error('algo falla después');
  });
  // Resultado: la factura NO se emite,
  // pero el contador YA se ha incrementado. Hueco en la numeración.
}

async emitir(id: string) {
  await this.em.transactional(async (em) => {
    const f = await em.findOneOrFail(Factura, id);
    f.estado = 'EMITIDA';
    // Todo dentro del MISMO em forkeado:
    // un único COMMIT o un único ROLLBACK.
    const contador = await em.findOneOrFail(Contador, 'facturas');
    contador.siguiente++;
    throw new Error('algo falla después');
  });
  // Resultado: nada cambia. Ni factura ni contador.
}

17.3.3 Transacciones anidadas y savepoints

Si llamas a em.transactional() con un EntityManager que ya está dentro de una transacción, MikroORM no abre una segunda transacción real (la mayoría de motores SQL no lo permiten): crea un savepoint. Un fallo en el bloque interno deshace solo hasta el savepoint; la transacción externa sigue viva y puede confirmar el resto.

BEGIN;
  UPDATE "pedido" SET "estado" = 'PAGADO' WHERE "id" = 101;
  SAVEPOINT "trx1";                       -- entra el bloque interno
    INSERT INTO "notificacion" ("pedido_id", "canal") VALUES (101, 'EMAIL');
  ROLLBACK TO SAVEPOINT "trx1";           -- el bloque interno falló
  -- la notificación se descarta, el pedido sigue pagado
COMMIT;

Esto habilita un patrón útil: operaciones opcionales dentro de un caso de uso obligatorio. El pago debe confirmarse; el registro de la notificación es deseable pero no crítico, así que se envuelve en su propio bloque con su propio try/catch. Úsalo con moderación: si abusas de los savepoints acabas con transacciones largas y con una semántica difícil de razonar en las revisiones de código.

17.3.4 Control manual: begin, commit, rollback

const em = this.em.fork();          // contexto propio: nunca uses el EM global
await em.begin();
try {
  await em.persistAndFlush(cabecera);
  await this.procesarLotes(em);
  await em.commit();                // commit hace flush implícito de lo pendiente
} catch (e) {
  await em.rollback();              // obligatorio: si no, la conexión queda retenida
  throw e;
}

Cuándo hace falta. Muy pocas veces. El control manual solo se justifica cuando el principio y el final de la transacción no caben en un mismo ámbito léxico: una máquina de estados dirigida por eventos, un adaptador que integra una librería de terceros que exige la conexión abierta, o una transacción por petición implementada en un middleware. En cualquier otro caso, em.transactional() es superior porque el rollback no se puede olvidar.

17.3.5 @Transactional() y @CreateRequestContext()

Son dos decoradores que resuelven dos problemas distintos y que se confunden constantemente.

import { CreateRequestContext, MikroORM } from '@mikro-orm/core';
import { Cron, CronExpression } from '@nestjs/schedule';
@Injectable()
export class InformesCron {
  // El decorador NECESITA encontrar la instancia de MikroORM (o un EntityManager)
  // en una propiedad de la clase; por eso se inyecta aunque no se use directamente.
  constructor(private readonly orm: MikroORM, private readonly em: EntityManager) {}
  @Cron(CronExpression.EVERY_DAY_AT_3AM)
  @CreateRequestContext()
  async generarInformeDiario(): Promise<void> {
    // Aquí this.em resuelve al fork creado por el decorador:
    // Identity Map propio, aislado del resto de la aplicación.
    const ventas = await this.em.find(Venta, { fecha: { $gte: ayer() } });
    this.em.create(InformeDiario, { fecha: ayer(), total: sumar(ventas) });
    await this.em.flush();
  }
}

import { Transactional } from '@mikro-orm/core';
@Injectable()
export class CobrosService {
  constructor(private readonly em: EntityManager) {}
  // Equivale a envolver todo el cuerpo en em.transactional(...).
  // El em inyectado resuelve al contexto transaccional durante la ejecución.
  @Transactional()
  async aplicarCobro(pedidoId: string, importe: number): Promise<void> {
    const pedido = await this.em.findOneOrFail(Pedido, pedidoId);
    pedido.pagado += importe;
    if (pedido.pagado >= pedido.total) pedido.estado = 'PAGADO';
    this.em.create(MovimientoCaja, { pedido, importe });
  }
}

17.3.6 Transacción por petición HTTP

La idea es tentadora: abrir una transacción al principio de cada petición que modifique datos y confirmarla si el controlador responde sin error. Se implementa con un interceptor de Nest que envuelve la ejecución del manejador dentro de em.transactional() y crea el contexto de petición con el EM transaccional, para que los servicios que inyectan EntityManager reciban ese mismo fork.

import { CallHandler, ExecutionContext, Injectable, NestInterceptor } from '@nestjs/common';
import { EntityManager, RequestContext } from '@mikro-orm/core';
import { Observable, from, lastValueFrom } from 'rxjs';
const METODOS_SEGUROS = new Set(['GET', 'HEAD', 'OPTIONS']);
@Injectable()
export class TransactionInterceptor implements NestInterceptor {
  constructor(private readonly em: EntityManager) {}
  intercept(ctx: ExecutionContext, next: CallHandler): Observable<unknown> {
    const req = ctx.switchToHttp().getRequest<{ method: string }>();
    // Las lecturas no necesitan transacción explícita: evita bloqueos innecesarios.
    if (METODOS_SEGUROS.has(req.method)) return next.handle();
    return from(
      // Sin RequestContext.create, los servicios seguirían resolviendo al EM de
      // la petición en lugar de al forkeado de la transacción.
      this.em.transactional((txEm) =>
        RequestContext.create(txEm, () => lastValueFrom(next.handle())),
      ),
    );
  }
}

17.4 Niveles de aislamiento

El aislamiento perfecto (que todo ocurra como si las transacciones se ejecutaran una detrás de otra) es caro. El estándar SQL define cuatro niveles que intercambian corrección por concurrencia, y cada nivel se define por las anomalías que permite.

17.4.1 Las cuatro anomalías, con línea temporal

Lectura sucia (dirty read): leer un dato que otra transacción ha escrito y todavía no ha confirmado.

  T1 (transfiere)                     T2 (informe de saldos)
  ──────────────────────────────────────────────────────────────────
  BEGIN
  UPDATE cuenta SET saldo = 0
    WHERE id = 'A'          (saldo real: 1000)
                                      BEGIN
                                      SELECT saldo FROM cuenta WHERE id='A'
                                        → 0        ¡dato NO confirmado!
                                      -- decide bloquear la cuenta por saldo cero
  ROLLBACK   (el saldo vuelve a 1000)
                                      COMMIT
  ──────────────────────────────────────────────────────────────────
  Resultado: T2 tomó una decisión sobre un dato que nunca existió.

Lectura no repetible (non-repeatable read): la misma fila leída dos veces en la misma transacción devuelve valores distintos, porque otra transacción la modificó y confirmó en medio.

  T1 (valida y cobra)                 T2 (el cliente cambia el descuento)
  ──────────────────────────────────────────────────────────────────
  BEGIN
  SELECT descuento FROM cliente       → 10
                                      BEGIN
                                      UPDATE cliente SET descuento = 50
                                      COMMIT
  SELECT descuento FROM cliente       → 50      ¡cambió dentro de T1!
  -- validó con 10 y cobra con 50
  COMMIT

Lectura fantasma (phantom read): la misma consulta con la misma condición devuelve un conjunto de filas distinto porque otra transacción insertó o borró filas que cumplen el criterio.

  T1 (cierre de caja)                 T2 (nueva venta)
  ──────────────────────────────────────────────────────────────────
  BEGIN
  SELECT count(*) FROM venta
    WHERE dia = hoy()                 → 120
                                      BEGIN
                                      INSERT INTO venta (dia,...) VALUES (hoy(),...)
                                      COMMIT
  SELECT sum(importe) FROM venta
    WHERE dia = hoy()                 → suma de 121 filas    ¡fantasma!
  COMMIT
  ──────────────────────────────────────────────────────────────────
  Resultado: el informe dice 120 ventas y un importe que corresponde a 121.

Anomalía de serialización (incluye el write skew): cada transacción es correcta por separado y ninguna lee datos sucios, pero el resultado conjunto viola una invariante que ninguna de las dos pudo ver.

  Invariante: siempre debe quedar al menos un médico de guardia.
  Estado inicial: guardia = { Ana, Luis }
  T1 (Ana pide el turno libre)        T2 (Luis pide el turno libre)
  ──────────────────────────────────────────────────────────────────
  BEGIN                               BEGIN
  SELECT count(*) FROM guardia → 2    SELECT count(*) FROM guardia → 2
  -- 2 > 1, puedo salir               -- 2 > 1, puedo salir
  DELETE FROM guardia WHERE m='Ana'   DELETE FROM guardia WHERE m='Luis'
  COMMIT                              COMMIT
  ──────────────────────────────────────────────────────────────────
  Estado final: guardia = { }   La invariante se rompió sin ninguna lectura sucia.
  Solo SERIALIZABLE (o un bloqueo explícito) evita esto.

17.4.2 Qué evita cada nivel

17.4.3 Cómo fijar el nivel en MikroORM

import { IsolationLevel } from '@mikro-orm/core';
// Por transacción: lo habitual y lo recomendable.
await this.em.transactional(
  async (em) => {
    const aforo = await em.count(Reserva, { sesion: sesionId });
    if (aforo >= capacidad) throw new AforoCompletoError();
    em.create(Reserva, { sesion: sesionId, usuario: usuarioId });
  },
  { isolationLevel: IsolationLevel.SERIALIZABLE },
);
// Con control manual, el nivel se pasa al begin.
await em.begin({ isolationLevel: IsolationLevel.REPEATABLE_READ });

BEGIN;
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
SELECT count(*) AS "count" FROM "reserva" WHERE "sesion_id" = 42;
INSERT INTO "reserva" ("sesion_id", "usuario_id") VALUES (42, 9);
COMMIT;
-- Si otra transacción concurrente hizo lo mismo:
-- ERROR: could not serialize access due to read/write dependencies among transactions (40001)
-- → la aplicación DEBE reintentar la operación completa.

Cuándo subir el nivel: invariantes que abarcan varias filas y que no puedes expresar con una restricción (aforo, saldo mínimo agregado, «al menos un administrador», asignación de turnos), informes que deben ver una instantánea coherente de varias tablas, y procesos de cierre contable. Cuándo no: como solución genérica al problema de la actualización perdida sobre una sola fila; ahí el bloqueo optimista o pesimista es más barato y más preciso.

17.5 Concurrencia: bloqueo optimista y pesimista

17.5.1 El problema de la actualización perdida

Es el problema de concurrencia más común y el que más dinero cuesta. Ocurre siempre que el código sigue el patrón leer → decidir en memoria → escribir, porque entre la lectura y la escritura hay una ventana en la que otro proceso puede hacer lo mismo.

  Stock inicial del producto 3: 10 unidades.
  Dos peticiones simultáneas compran 8 unidades cada una.
  Petición A (nodo 1)                       Petición B (nodo 2)
  ─────────────────────────────────────────────────────────────────────────
  BEGIN                                     BEGIN
  SELECT stock FROM producto WHERE id=3
     → 10
                                            SELECT stock FROM producto WHERE id=3
                                               → 10          (¡lee lo mismo!)
  -- 10 >= 8, venta permitida
                                            -- 10 >= 8, venta permitida
  UPDATE producto SET stock = 2 WHERE id = 3  (B espera el bloqueo de fila)
  COMMIT
                                            UPDATE producto SET stock = 2 WHERE id = 3
                                            COMMIT
  ─────────────────────────────────────────────────────────────────────────
  Stock final: 2.   Vendidas: 16 unidades de 10 existentes.
  La escritura de A se ha PERDIDO: B la sobrescribió con un valor calculado
  a partir de un dato ya obsoleto.

Hay tres formas de cerrar esa ventana, en orden creciente de coste:

1. Escritura atómica relativa. Que la base de datos haga la aritmética: UPDATE producto SET stock = stock - 8 WHERE id = 3 AND stock >= 8. Si afecta a cero filas, no había stock. Es la opción más eficiente y la que más se olvida.
2. Bloqueo optimista. Permitir la carrera y detectarla al escribir mediante una columna de versión.
3. Bloqueo pesimista. Impedir la carrera bloqueando la fila en el momento de leerla.

// Leer, decidir en memoria y escribir:
// dos peticiones concurrentes venden el mismo stock.
async reservar(id: string, cantidad: number) {
  const p = await this.em.findOneOrFail(Producto, id);
  if (p.stock < cantidad) throw new StockInsuficienteError(id);
  p.stock -= cantidad;      // valor calculado con un dato posiblemente obsoleto
  await this.em.flush();
}

// La condición y la resta las evalúa la base de datos
// en la misma instrucción: no hay ventana de carrera.
async reservar(id: string, cantidad: number) {
  const filas = await this.em.nativeUpdate(
    Producto,
    { id, stock: { $gte: cantidad } },
    { stock: this.em.raw(`stock - ${'?'}`, [cantidad]) },
  );
  if (filas === 0) throw new StockInsuficienteError(id);
}

17.5.2 Bloqueo optimista

La estrategia optimista asume que los conflictos son raros: no bloquea nada, pero añade a la entidad una columna de versión que se incrementa en cada actualización. La cláusula WHERE del UPDATE incluye la versión leída; si otra transacción ya la cambió, el UPDATE afecta a cero filas y el ORM lanza una excepción.

import { Entity, PrimaryKey, Property } from '@mikro-orm/core';
@Entity()
export class Producto {
  @PrimaryKey() id!: string;
  @Property() nombre!: string;
  @Property() stock!: number;
  // Versión numérica: la opción recomendada. Empieza en 1 y la gestiona el ORM.
  @Property({ version: true }) version!: number;
}
@Entity()
export class Documento {
  @PrimaryKey() id!: string;
  @Property({ type: 'text' }) contenido!: string;
  // Versión por marca de tiempo: legible en informes, pero con menos resolución;
  // dos actualizaciones en el mismo milisegundo pueden no detectarse como conflicto.
  @Property({ version: true, length: 6 }) actualizadoEn!: Date;
}

-- Lectura: el ORM guarda version = 7 en la instantánea del Unit of Work
SELECT "id", "nombre", "stock", "version" FROM "producto" WHERE "id" = 'p-3';
-- Escritura: la versión leída viaja en el WHERE y se incrementa en el SET
UPDATE "producto"
   SET "stock" = 2, "version" = 8
 WHERE "id" = 'p-3' AND "version" = 7;
-- Si otra transacción confirmó antes, la fila ya tiene version = 8:
-- filas afectadas = 0  →  MikroORM lanza OptimisticLockError

El mecanismo cubre además un caso que el bloqueo pesimista no puede cubrir: la edición concurrente entre peticiones distintas. Si un formulario web se carga con version = 7, el usuario tarda cinco minutos en enviarlo y otro usuario ha guardado en medio, la versión enviada por el cliente permite detectar el conflicto aunque no exista ninguna transacción abierta durante esos cinco minutos.

import { LockMode, OptimisticLockError } from '@mikro-orm/core';
import { ConflictException } from '@nestjs/common';
@Patch(':id')
// dto.version es la versión que el cliente leyó al cargar el formulario.
async actualizar(@Param('id') id: string, @Body() dto: ActualizarProductoDto): Promise<ProductoDto> {
  try {
    return await this.em.transactional(async (em) => {
      // Comprueba la versión ANTES de aplicar cambios: falla rápido.
      const p = await em.findOneOrFail(Producto, id, {
        lockMode: LockMode.OPTIMISTIC, lockVersion: dto.version,
      });
      em.assign(p, { nombre: dto.nombre, precio: dto.precio });
      return aDto(p);
    });
  } catch (e) {
    if (e instanceof OptimisticLockError) {
      // 409: el recurso cambió desde que el cliente lo leyó.
      throw new ConflictException({ code: 'VERSION_CONFLICT',
        message: 'El producto ha sido modificado por otro usuario. Recarga y vuelve a intentarlo.' });
    }
    throw e;
  }
}

Cuando no puedes o no quieres añadir una columna de versión (por ejemplo en una tabla heredada), MikroORM ofrece concurrencyCheck a nivel de propiedad: las propiedades marcadas se incluyen en el WHERE del UPDATE con el valor que tenían al cargar la entidad. Es un bloqueo optimista basado en el contenido en lugar de en un contador.

@Entity()
export class Tarifa {
  @PrimaryKey()
  id!: string;
  // Sin columna de versión: el propio importe actúa de testigo.
  @Property({ concurrencyCheck: true })
  importe!: number;
  @Property()
  descripcion!: string;
}
// UPDATE "tarifa" SET "importe" = 120 WHERE "id" = 't1' AND "importe" = 100;
// 0 filas afectadas → OptimisticLockError

17.5.3 Bloqueo pesimista

La estrategia pesimista asume que el conflicto es probable y lo evita bloqueando la fila en la base de datos en el momento de leerla. Solo tiene sentido dentro de una transacción: el bloqueo se libera con el COMMIT o el ROLLBACK.

import { LockMode } from '@mikro-orm/core';
async transferir(origenId: string, destinoId: string, importe: number): Promise<void> {
  if (origenId === destinoId) throw new BadRequestException('Cuentas iguales');
  await this.em.transactional(async (em) => {
    // CLAVE: bloquear SIEMPRE en el mismo orden (por id) en todo el sistema.
    // Si un caso de uso bloquea A→B y otro B→A, aparecen deadlocks.
    const [primero, segundo] = [origenId, destinoId].sort();
    const c1 = await em.findOneOrFail(Cuenta, primero,  { lockMode: LockMode.PESSIMISTIC_WRITE });
    const c2 = await em.findOneOrFail(Cuenta, segundo, { lockMode: LockMode.PESSIMISTIC_WRITE });
    const [origen, destino] = c1.id === origenId ? [c1, c2] : [c2, c1];
    if (origen.saldo < importe) throw new SaldoInsuficienteError(origenId);
    origen.saldo  -= importe;
    destino.saldo += importe;
    em.create(Movimiento, { origen, destino, importe });
  });
}
// Si la entidad ya está cargada: await em.lock(cuenta, LockMode.PESSIMISTIC_WRITE);

BEGIN;
SELECT "id", "saldo" FROM "cuenta" WHERE "id" = 'A' FOR UPDATE;   -- bloquea la fila A
SELECT "id", "saldo" FROM "cuenta" WHERE "id" = 'B' FOR UPDATE;   -- bloquea la fila B
UPDATE "cuenta" SET "saldo" = 900 WHERE "id" = 'A';
UPDATE "cuenta" SET "saldo" = 1100 WHERE "id" = 'B';
INSERT INTO "movimiento" ("origen_id", "destino_id", "importe") VALUES ('A', 'B', 100);
COMMIT;   -- aquí se liberan los dos bloqueos

17.5.4 Una cola de trabajos en la base de datos con SKIP LOCKED

SKIP LOCKED resuelve el problema de reparto entre trabajadores sin coordinación externa: cada trabajador toma las primeras filas pendientes que no estén bloqueadas por otro. Es la forma correcta de implementar una cola sencilla cuando no quieres añadir Redis a la arquitectura.

  tabla trabajo:  id | estado    | intentos | payload
                  1..4 | PENDIENTE |   0    | ...
   Trabajador 1                     Trabajador 2
   ────────────────────────────────────────────────────────────────
   BEGIN                            BEGIN
   SELECT ... LIMIT 2
     FOR UPDATE SKIP LOCKED
     → filas 1, 2  (bloqueadas)
                                    SELECT ... LIMIT 2
                                      FOR UPDATE SKIP LOCKED
                                      → filas 3, 4  (omite 1 y 2)
   UPDATE estado='EN_CURSO' (1,2)   UPDATE estado='EN_CURSO' (3,4)
   COMMIT                           COMMIT
   ────────────────────────────────────────────────────────────────
   Ningún trabajo se procesa dos veces y ningún trabajador espera al otro.

import { LockMode } from '@mikro-orm/core';
async reclamarLote(tamano = 10): Promise<Trabajo[]> {
  return this.em.transactional(async (em) => {
    const trabajos = await em.find(
      Trabajo,
      { estado: 'PENDIENTE', disponibleEn: { $lte: new Date() } },
      { limit: tamano, orderBy: { prioridad: 'desc', creadoEn: 'asc' },
        lockMode: LockMode.PESSIMISTIC_PARTIAL_WRITE },   // FOR UPDATE SKIP LOCKED
    );
    for (const t of trabajos) {
      t.estado = 'EN_CURSO';
      t.intentos++;
      t.reclamadoEn = new Date();
    }
    return trabajos;   // flush y COMMIT los hace transactional
  });
}

BEGIN;
SELECT "t0".* FROM "trabajo" AS "t0"
 WHERE "t0"."estado" = 'PENDIENTE' AND "t0"."disponible_en" <= now()
 ORDER BY "t0"."prioridad" DESC, "t0"."creado_en" ASC
 LIMIT 10
   FOR UPDATE SKIP LOCKED;
UPDATE "trabajo" SET "estado" = 'EN_CURSO', "intentos" = 1, "reclamado_en" = now()
 WHERE "id" IN (1, 2, 3);
COMMIT;

17.5.5 Optimista frente a pesimista: criterios de elección

17.5.6 Deadlocks: cómo se producen y cómo convivir con ellos

Un deadlock es un ciclo de esperas: A espera un recurso que tiene B y B espera un recurso que tiene A. Ninguna avanza, así que el motor elige una víctima y la aborta con un error específico (40P01 en PostgreSQL, 1213 en MySQL).

  T1: transferir de A a B            T2: transferir de B a A
  ──────────────────────────────────────────────────────────────────
  BEGIN                              BEGIN
  SELECT A FOR UPDATE   (tiene A)
                                     SELECT B FOR UPDATE   (tiene B)
  SELECT B FOR UPDATE   ── espera a T2 ──┐
                                     SELECT A FOR UPDATE  ── espera a T1 ──┐
                                                                            │
        ┌───────────────────────────────────────────────────────────────────┘
        └── CICLO: el motor aborta una de las dos (deadlock detected)
  Solución: ordenar SIEMPRE los bloqueos por un criterio estable (el id).
  Con orden estable, T2 también pediría A antes que B: esperaría, pero no se
  formaría el ciclo.

• Orden coherente de bloqueo. Es la medida más eficaz: ordena las claves antes de bloquear, como en el ejemplo de la transferencia. Afecta también al orden en que insertas en varias tablas.
• Transacciones cortas. Menos tiempo con bloqueos tomados es menos probabilidad de ciclo. Saca de la transacción todo lo que no necesita estar dentro.
• Menos filas bloqueadas. Un UPDATE con un WHERE poco selectivo puede bloquear miles de filas y, sin índice adecuado, escalar a bloqueos de rango.
• Reintento. Un deadlock no es un error de programación puntual: es un evento esperable. La operación debe poder repetirse.

import { DriverException, OptimisticLockError } from '@mikro-orm/core';
// Códigos SQLSTATE que SIEMPRE merecen reintento: son conflictos, no errores de datos.
const REINTENTABLES = new Set([
  '40001',   // serialization failure (PostgreSQL: SERIALIZABLE / REPEATABLE READ)
  '40P01',   // deadlock detected (PostgreSQL)
]);
function esReintentable(e: unknown): boolean {
  if (e instanceof OptimisticLockError) return true;
  const code = (e as DriverException)?.code ?? (e as { code?: string }).code;
  // MySQL usa códigos numéricos: 1213 deadlock, 1205 lock wait timeout.
  return REINTENTABLES.has(String(code)) || ['1213', '1205'].includes(String(code));
}
export async function conReintentos<T>(op: () => Promise<T>, intentos = 3): Promise<T> {
  let ultimo: unknown;
  for (let i = 0; i < intentos; i++) {
    try {
      return await op();
    } catch (e) {
      if (!esReintentable(e)) throw e;    // un error de negocio no se reintenta
      ultimo = e;
      const espera = 25 * 2 ** i + Math.random() * 50;   // backoff + jitter
      await new Promise((r) => setTimeout(r, espera));
    }
  }
  throw ultimo;
}
// Uso: cada intento vuelve a abrir la transacción y a releer los datos.
await conReintentos(() => this.transferencias.transferir(a, b, 100));

17.5.7 Idempotencia y claves de idempotencia

Reintentar solo es seguro si repetir la operación produce el mismo resultado que ejecutarla una vez. Y hay que reintentar en muchos sitios: el cliente pulsa dos veces, el móvil pierde la cobertura y repite el POST, el balanceador reenvía, la cola entrega el mensaje dos veces. La solución estándar es que el cliente envíe una clave de idempotencia y que el servidor la guarde con una restricción de unicidad.

@Entity()
@Unique({ properties: ['clave'] })
export class OperacionIdempotente {
  @PrimaryKey() id!: string;
  @Property() clave!: string;            // cabecera Idempotency-Key generada por el cliente
  @Property() endpoint!: string;
  @Property() huellaPeticion!: string;   // detecta claves reutilizadas con otro cuerpo
  @Property({ type: 'json', nullable: true }) respuesta?: unknown;
  @Property() creadoEn: Date = new Date();
}

async cobrar(clave: string, dto: CobroDto): Promise<RespuestaCobro> {
  return this.em.transactional(async (em) => {
    const huella = sha256(JSON.stringify(dto));
    // 1. ¿Ya se procesó esta clave? Devuelve la MISMA respuesta.
    const previa = await em.findOne(OperacionIdempotente, { clave });
    if (previa) {
      if (previa.huellaPeticion !== huella) {
        throw new ConflictException('La clave de idempotencia se reutilizó con otro cuerpo');
      }
      return previa.respuesta as RespuestaCobro;
    }
    // 2. Reserva la clave DENTRO de la transacción: si dos peticiones
    //    simultáneas llegan aquí, la restricción única hace fallar a una
    //    (error 23505) y esa devuelve 409 o reintenta la lectura del paso 1.
    const registro = em.create(OperacionIdempotente,
      { clave, endpoint: 'POST /pagos', huellaPeticion: huella });
    // 3. Trabajo real, en la misma transacción.
    const pago = em.create(Pago, { importe: dto.importe, pedido: dto.pedidoId });
    const respuesta: RespuestaCobro = { pagoId: pago.id, estado: 'ACEPTADO' };
    registro.respuesta = respuesta;
    return respuesta;
  });
}

17.6 Esquema y migraciones

17.6.1 SchemaGenerator: útil en desarrollo, prohibido en producción

MikroORM sabe deducir el esquema de la base de datos a partir de los metadatos de tus entidades y compararlo con el esquema real. El objeto que hace ese trabajo es el SchemaGenerator, accesible como orm.schema y expuesto en la CLI como los comandos schema:*.

let orm: MikroORM;
beforeAll(async () => {
  orm = await MikroORM.init(configTest);        // base de datos de test dedicada
  await orm.schema.refreshDatabase();           // esquema limpio desde las entidades
});
beforeEach(async () => {
  await orm.schema.clearDatabase();   // vacía datos y conserva estructura: más rápido
  orm.em.clear();                     // Identity Map limpio entre tests
});
afterAll(() => orm.close(true));

17.6.2 Migraciones: configuración y comandos

Una migración es un fichero versionado, revisable y ordenado que describe un cambio de esquema con SQL explícito. El paquete @mikro-orm/migrations (o @mikro-orm/migrations-mongodb) añade el Migrator y sus comandos de CLI.

import { defineConfig } from '@mikro-orm/postgresql';
import { Migrator } from '@mikro-orm/migrations';
import { SeedManager } from '@mikro-orm/seeder';
export default defineConfig({
  // ...entidades, conexión
  extensions: [Migrator, SeedManager],      // en v6 las extensiones se registran así
  migrations: {
    tableName: 'mikro_orm_migrations',      // tabla de control de versiones
    path: './dist/migrations',              // JS compilado (producción)
    pathTs: './src/migrations',             // TS (desarrollo)
    glob: '!(*.d).{js,ts}',
    emit: 'ts',
    transactional: true,                    // cada migración en su transacción
    allOrNothing: true,                     // todas las pendientes en UNA transacción
    disableForeignKeys: false,              // en PostgreSQL requiere superusuario: déjalo en false
    dropTables: false,                      // sin DROP TABLE generados automáticamente
    safe: true,                             // sin DROP COLUMN generados automáticamente
    snapshot: true,                         // instantánea para calcular diffs incrementales
  },
});

{
  "scripts": {
    "mikro": "mikro-orm --config ./src/mikro-orm.config.ts",
    "db:migration:create": "npm run mikro -- migration:create",
    "db:migration:up": "npm run mikro -- migration:up",
    "db:migration:pending": "npm run mikro -- migration:pending",
    "db:migrate:prod": "mikro-orm --config ./dist/mikro-orm.config.js migration:up"
  }
}

La tabla de migraciones (mikro_orm_migrations) es el estado: una fila por migración aplicada, con su nombre y su fecha. El Migrator compara los ficheros del directorio con esas filas para decidir qué está pendiente. De ahí dos consecuencias operativas: nunca renombres ni edites una migración ya aplicada (deja de reconocerse y se intenta aplicar de nuevo), y nunca borres filas de esa tabla a mano salvo que sepas exactamente qué estás reparando.

17.6.3 Anatomía de una migración

import { Migration } from '@mikro-orm/migrations';
export class Migration20260615103000 extends Migration {
  // up: el cambio hacia adelante.
  async up(): Promise<void> {
    this.addSql(`alter table "pedido" add column "estado" varchar(20) null;`);
    this.addSql(`update "pedido" set "estado" = 'PENDIENTE' where "estado" is null;`);
    this.addSql(`create index "pedido_estado_idx" on "pedido" ("estado");`);
  }
  // down: la vuelta atrás, en orden inverso.
  async down(): Promise<void> {
    this.addSql(`drop index "pedido_estado_idx";`);
    this.addSql(`alter table "pedido" drop column "estado";`);
  }
}

• this.addSql() encola instrucciones que se ejecutan en orden al aplicar la migración; no las ejecuta en el momento. Si necesitas leer datos para decidir, usa this.execute() o el query builder subyacente con this.getKnex().
• Cada migración se ejecuta dentro de una transacción si el motor lo permite (transactional: true), y con allOrNothing todas las pendientes comparten una sola transacción: si la cuarta falla, se deshacen las cuatro.
• El nombre del fichero fija el orden. No lo cambies después de aplicarlo.

Algunas operaciones no pueden ejecutarse dentro de una transacción. La más frecuente es la creación de índices sin bloquear escrituras en PostgreSQL (CREATE INDEX CONCURRENTLY), que el motor rechaza en un bloque transaccional. La migración debe declararse como no transaccional.

export class Migration20260620090000 extends Migration {
  // Desactiva la transacción SOLO para esta migración.
  override isTransactional(): boolean {
    return false;
  }
  async up(): Promise<void> {
    // CONCURRENTLY no bloquea escrituras, pero tarda más y puede quedar
    // el índice en estado INVALID si falla: hay que comprobarlo después.
    this.addSql(`create index concurrently "factura_cliente_fecha_idx"
                   on "factura" ("cliente_id", "fecha");`);
  }
  async down(): Promise<void> {
    this.addSql(`drop index concurrently if exists "factura_cliente_fecha_idx";`);
  }
}

17.6.4 Migraciones sin tiempo de inactividad: expand/contract

Durante un despliegue hay un intervalo en el que convive código antiguo con código nuevo (despliegue progresivo, varias réplicas, rolling update de Kubernetes). Por tanto el esquema debe ser compatible con ambas versiones a la vez. El patrón que lo garantiza es expand/contract: primero se amplía el esquema de forma compatible, luego se migra el código y los datos, y solo al final se contrae eliminando lo viejo.

  Objetivo: renombrar "nombre_completo" en "nombre" + "apellidos".
  PASO 1 · EXPAND (migración)      añade columnas NULL, no toca nada más
  ┌──────────────────────────────────────────────────────────────────────┐
  │ nombre_completo  │ nombre (null)  │ apellidos (null)                 │
  └──────────────────────────────────────────────────────────────────────┘
     código v1 (lee/escribe nombre_completo)  ✔ sigue funcionando
  PASO 2 · DOBLE ESCRITURA (despliegue de código v2)
     v2 escribe en nombre_completo Y en nombre/apellidos; lee nombre_completo
     └── v1 y v2 conviven ✔ · esperar a que TODAS las réplicas sean v2
  PASO 3 · RELLENO (script por lotes, fuera de la transacción de despliegue)
     UPDATE ... SET nombre=..., apellidos=... WHERE nombre IS NULL  (en tandas)
     └── idempotente: se puede repetir y reanudar
  PASO 4 · CAMBIO DE LECTURA (despliegue de código v3)
     v3 lee nombre/apellidos y sigue escribiendo en ambos sitios
     └── ventana de seguridad: si v3 falla, se vuelve a v2 sin pérdida
  PASO 5 · DEJAR DE ESCRIBIR LO VIEJO (despliegue de código v4)
     v4 ignora nombre_completo por completo
  PASO 6 · CONTRACT (migración, días después)
     ALTER TABLE ... DROP COLUMN nombre_completo;  + NOT NULL en las nuevas
  ────────────────────────────────────────────────────────────────────────
  Cada paso es desplegable y reversible por separado: nunca hay un instante en el
  que el esquema y el código en ejecución sean incompatibles.

async up(): Promise<void> {
  // Renombrado "atómico": rompe el código antiguo
  // en el instante en que se aplica.
  this.addSql(`alter table "cliente"
    rename column "nombre_completo" to "nombre";`);
  this.addSql(`alter table "cliente"
    add column "apellidos" varchar(120) not null;`);
}
// Consecuencias durante el rolling update:
//  · las réplicas v1 fallan con "column nombre_completo
//    does not exist" en CADA consulta
//  · NOT NULL sin default hace fallar la propia
//    migración si la tabla ya tiene filas

async up(): Promise<void> {
  // Paso 1: solo añadir, siempre nullable y sin
  // reescribir la tabla. Compatible con v1 y v2.
  this.addSql(`alter table "cliente"
    add column "nombre" varchar(60) null;`);
  this.addSql(`alter table "cliente"
    add column "apellidos" varchar(120) null;`);
}
async down(): Promise<void> {
  this.addSql(`alter table "cliente" drop column "nombre";`);
  this.addSql(`alter table "cliente" drop column "apellidos";`);
}

-- 1. Añadir columna: instantáneo, no reescribe la tabla
ALTER TABLE "cliente" ADD COLUMN "nombre" varchar(60) NULL;
-- 2. Relleno por lotes: nunca en un solo UPDATE de 20 millones de filas
UPDATE "cliente" SET "nombre" = split_part("nombre_completo", ' ', 1)
 WHERE "id" IN (
   SELECT "id" FROM "cliente" WHERE "nombre" IS NULL ORDER BY "id" LIMIT 5000
 );
-- repetir hasta que afecte a 0 filas, con una pausa entre tandas
-- 3. NOT NULL en dos fases, sin bloqueo largo de escritura
ALTER TABLE "cliente" ADD CONSTRAINT "cliente_nombre_no_nulo"
  CHECK ("nombre" IS NOT NULL) NOT VALID;     -- inmediato: no valida lo existente
ALTER TABLE "cliente" VALIDATE CONSTRAINT "cliente_nombre_no_nulo";  -- escanea sin bloquear lecturas
ALTER TABLE "cliente" ALTER COLUMN "nombre" SET NOT NULL;            -- ya es barato
ALTER TABLE "cliente" DROP CONSTRAINT "cliente_nombre_no_nulo";
-- 4. Índice sin bloquear escrituras (fuera de transacción)
CREATE INDEX CONCURRENTLY "cliente_nombre_idx" ON "cliente" ("nombre");
-- 5. Protección universal contra ALTER que se queda esperando un bloqueo
SET lock_timeout = '3s';        -- si no consigue el bloqueo en 3 s, falla en vez de encolar
SET statement_timeout = '30s';

17.6.5 Migraciones de datos frente a migraciones de esquema

Un relleno pequeño (unas miles de filas, tiempo previsible) puede ir dentro de la migración. Sácalo a un script aparte cuando se cumpla cualquiera de estas condiciones: afecta a más de unas decenas de miles de filas, necesita lógica de negocio de la aplicación (no expresable en SQL), debe ejecutarse por tandas con pausas, o puede necesitar reanudarse tras un fallo. Una migración que tarda una hora bloquea el despliegue durante una hora y, con allOrNothing, mantiene una transacción gigante abierta.

// Ejecución: node dist/scripts/rellenar-nombres.js
// Idempotente: procesa solo lo que falta, se puede parar y reanudar.
const orm = await MikroORM.init();
const em = orm.em.fork();
let total = 0;
for (;;) {
  const lote = await em.find(Cliente, { nombre: null }, { limit: 5000, orderBy: { id: 'asc' } });
  if (lote.length === 0) break;
  for (const c of lote) {
    const [nombre, ...resto] = c.nombreCompleto.split(' ');
    c.nombre = nombre;
    c.apellidos = resto.join(' ') || '-';
  }
  await em.flush();     // un COMMIT por lote: transacciones cortas
  em.clear();           // libera el Identity Map: memoria constante
  console.log(`rellenados ${(total += lote.length)}`);
  await new Promise((r) => setTimeout(r, 100));   // deja respirar a la base de datos
}
await orm.close(true);

17.6.6 Migraciones en CI/CD

  git push
     ▼
  ┌─────────────────────────── CI ────────────────────────────┐
  │ build · lint · test                                       │
  │ migration:check   ← ¿alguien tocó una entidad sin migrar? │
  │ up → down → up    ← el down funciona de verdad            │
  └─────────────────────────┬─────────────────────────────────┘
                            ▼
  ┌───────────────── DESPLIEGUE (una sola vez) ───────────────┐
  │ Job / init container / release phase:                     │
  │   1. pg_try_advisory_lock  ← evita ejecuciones simultáneas │
  │   2. migration:up                                         │
  │   3. libera el bloqueo                                    │
  └─────────────────────────┬─────────────────────────────────┘
                            ▼
  ┌──────────── ARRANQUE DE LAS RÉPLICAS DE LA APP ───────────┐
  │ N pods · NINGUNO ejecuta migraciones · solo migration:    │
  │ pending como comprobación de salud al arrancar            │
  └───────────────────────────────────────────────────────────┘

• Cuándo: después de construir la imagen y antes de que arranque el código nuevo. Como el esquema es compatible con la versión anterior (expand/contract), aplicarlas antes es seguro.
• Quién: un paso de despliegue dedicado (un Job de Kubernetes, un init container del Deployment, una release phase), nunca el bootstrap de la aplicación. Con réplicas, el arranque se ejecuta N veces.
• Bloqueo: aunque el paso sea único, un reintento del pipeline o dos despliegues solapados pueden coincidir. Un advisory lock de PostgreSQL es la forma más simple de serializarlo.
• Credenciales: el usuario que migra necesita permisos de DDL; el usuario de la aplicación, no. Dos usuarios distintos reducen el daño de una inyección SQL.

const orm = await MikroORM.init();
const conexion = orm.em.getConnection();
// Identificador arbitrario pero fijo para "migraciones de esta aplicación".
const [{ bloqueado }] = await conexion.execute<{ bloqueado: boolean }[]>(
  'select pg_try_advisory_lock(918273645) as bloqueado');
if (!bloqueado) {
  console.error('Otro proceso está migrando. Abortando sin error.');
  await orm.close(true);
  process.exit(0);          // no es un fallo: alguien más se está encargando
}
try {
  const pendientes = await orm.migrator.getPendingMigrations();
  console.log(`Aplicando ${pendientes.length} migraciones`);
  await orm.migrator.up();
} finally {
  await conexion.execute('select pg_advisory_unlock(918273645)');
  await orm.close(true);
}

17.7 Seeders y datos de prueba

Un entorno sin datos no se puede demostrar ni probar. @mikro-orm/seeder aporta dos piezas: el Seeder (un guion que puebla la base de datos, componible con otros seeders) y la Factory (una plantilla que genera entidades con datos verosímiles y permite sobrescribir lo que importa en cada test).

import { Factory } from '@mikro-orm/seeder';
import { faker } from '@faker-js/faker/locale/es';
import { Usuario } from '../entities/usuario.entity';
export class UsuarioFactory extends Factory<Usuario> {
  model = Usuario;
  // Valores por defecto verosímiles: todo lo que el test no fija.
  definition(): Partial<Usuario> {
    return {
      nombre: faker.person.firstName(), apellidos: faker.person.lastName(),
      email: faker.internet.email().toLowerCase(), activo: true,
      creadoEn: faker.date.past({ years: 2 }),
    };
  }
}

import { EntityManager } from '@mikro-orm/core';
import { Seeder } from '@mikro-orm/seeder';
export class DatabaseSeeder extends Seeder {
  async run(em: EntityManager): Promise<void> {
    // Datos deterministas primero: los que la aplicación necesita para arrancar.
    em.create(Rol, { codigo: 'ADMIN', nombre: 'Administrador' });
    em.create(Rol, { codigo: 'USER',  nombre: 'Usuario' });
    // Un usuario fijo y conocido para poder entrar a mano en desarrollo.
    new UsuarioFactory(em).makeOne({ email: 'admin@ejemplo.local', nombre: 'Admin' });
    // Volumen para ver la paginación y medir consultas.
    new UsuarioFactory(em).each((u) => { u.activo = Math.random() > 0.1; }).make(200);
    // Relaciones: crea el agregado completo desde la factoría del padre.
    new PedidoFactory(em).make(50);
    // El SeedManager hace flush al terminar; llamar a otros seeders:
    // return this.call(em, [CatalogoSeeder, PromocionesSeeder]);
  }
}

npx mikro-orm seeder:create DatabaseSeeder   # andamiaje de un seeder nuevo
npx mikro-orm seeder:run                     # ejecuta el defaultSeeder de la configuración
npx mikro-orm seeder:run --class=CatalogoSeeder
npx mikro-orm migration:fresh --seed         # local desde cero: esquema + migraciones + datos

17.8 Filtros globales

Un filtro global es una condición que MikroORM añade automáticamente al WHERE de todas las consultas sobre una entidad, sin que tengas que escribirla. Resuelve el problema de las condiciones transversales: reglas que deben cumplirse en el 100 % de las consultas y que, escritas a mano, se olvidan exactamente donde más duele.

import { Entity, Filter, PrimaryKey, Property } from '@mikro-orm/core';
@Entity()
// 1. Borrado lógico: activo por defecto en todas las consultas.
@Filter({ name: 'softDelete', cond: { deletedAt: null }, default: true })
// 2. Multi-inquilino: la condición es una función y args: true exige parámetros.
@Filter({ name: 'tenant', args: true, default: true,
          cond: (args: { tenantId: string }) => ({ tenant: args.tenantId }) })
// 3. Visibilidad por permisos: la condición puede depender del tipo de operación
//    ('read', 'update', 'delete'). Con default: false se activa por consulta.
@Filter({ name: 'soloPublicados', default: false,
          cond: (args, type) => (type === 'read' ? { estado: 'PUBLICADO' } : {}) })
export class Documento {
  @PrimaryKey() id!: string;
  @Property() titulo!: string;
  @Property() estado!: 'BORRADOR' | 'PUBLICADO';
  @Property() tenant!: string;
  @Property({ nullable: true }) deletedAt?: Date;
}

// Definidos globalmente en la configuración: se aplican a las entidades indicadas.
export default defineConfig({
  filters: {
    softDelete: { cond: { deletedAt: null }, default: true, entity: ['Documento', 'Usuario'] },
  },
});
// Añadidos en tiempo de ejecución sobre un EM concreto (por ejemplo, en un middleware):
em.addFilter('tenant', (args: { tenantId: string }) => ({ tenant: args.tenantId }), ['Documento']);
// Los parámetros se fijan una vez por contexto y valen para toda la petición:
em.setFilterParams('tenant', { tenantId: peticion.tenantId });
// Y se desactivan o activan por consulta:
await em.find(Documento, {}, { filters: { softDelete: false } });   // incluye borrados
await em.find(Documento, {}, { filters: ['soloPublicados'] });      // activa uno no-default
await em.find(Documento, {}, { filters: false });                   // desactiva TODOS (peligroso)

-- em.find(Documento, { estado: 'PUBLICADO' })   con softDelete y tenant activos
SELECT "d0".* FROM "documento" AS "d0"
 WHERE "d0"."estado" = 'PUBLICADO' AND "d0"."deleted_at" IS NULL AND "d0"."tenant" = 'acme';
-- em.find(Documento, { estado: 'PUBLICADO' }, { filters: false })
SELECT "d0".* FROM "documento" AS "d0" WHERE "d0"."estado" = 'PUBLICADO';
--  ← ninguna protección: ni borrado lógico ni aislamiento por inquilino

17.9 Soft delete (borrado lógico)

Borrar lógicamente es marcar la fila como eliminada en lugar de suprimirla. Se usa cuando el dato tiene valor histórico (facturas, movimientos), cuando hay que poder deshacer, cuando otras filas lo referencian o cuando una auditoría exige rastro. La implementación completa tiene más piezas de las que parece.

// Clase base reutilizable: la columna, el filtro y las operaciones.
@Filter({ name: 'softDelete', cond: { deletedAt: null }, default: true })
export abstract class BaseBorrable {
  @Property({ nullable: true, index: true }) deletedAt?: Date | null;
  @Property({ nullable: true }) deletedBy?: string | null;   // casi siempre se acaba necesitando
  get estaBorrado(): boolean { return this.deletedAt != null; }
}

@Injectable()
export class DocumentosService {
  constructor(private readonly em: EntityManager) {}
  // BORRAR: no es em.remove(), es una actualización.
  async borrar(id: string, usuarioId: string): Promise<void> {
    const doc = await this.em.findOneOrFail(Documento, id);   // el filtro evita reborrar
    doc.deletedAt = new Date();
    doc.deletedBy = usuarioId;
    await this.em.flush();
  }
  // RESTAURAR: hay que desactivar el filtro para poder encontrarlo.
  async restaurar(id: string): Promise<void> {
    const doc = await this.em.findOneOrFail(Documento, id, { filters: { softDelete: false } });
    doc.deletedAt = null;
    doc.deletedBy = null;
    await this.em.flush();
  }
  // CONSULTAR INCLUYENDO BORRADOS: solo para administración o papelera.
  async listarPapelera(): Promise<Documento[]> {
    return this.em.find(Documento, { deletedAt: { $ne: null } },
      { filters: { softDelete: false }, orderBy: { deletedAt: 'desc' } });
  }
  // PURGAR de verdad: borrado físico pasados N días (retención).
  async purgar(dias = 30): Promise<number> {
    const limite = new Date(Date.now() - dias * 86_400_000);
    // Sin desactivar el filtro no borraría nada: exige deletedAt IS NULL.
    return this.em.nativeDelete(Documento, { deletedAt: { $lt: limite } },
      { filters: { softDelete: false } });
  }
}

17.9.1 El problema clásico: restricciones únicas y filas borradas

Si usuario.email es único y borras lógicamente a ana@ejemplo.com, la fila sigue existiendo: la restricción única impide que Ana vuelva a registrarse. Es el error que aparece siempre, semanas después de implementar el borrado lógico, y en forma de duplicate key value violates unique constraint incomprensible para el usuario.

@Entity()
@Filter({ name: 'softDelete', cond: { deletedAt: null },
          default: true })
export class Usuario {
  @PrimaryKey() id!: string;
  // Unicidad global: cuenta también las filas borradas.
  @Property({ unique: true })
  email!: string;
  @Property({ nullable: true }) deletedAt?: Date;
}
// Ana se borra y quiere volver:
// ERROR: duplicate key value violates unique
//        constraint "usuario_email_unique"

@Entity()
@Filter({ name: 'softDelete', cond: { deletedAt: null },
          default: true })
// Índice único PARCIAL: solo aplica a las filas vivas.
// Se declara con una expresión para que la migración lo cree.
@Index({
  name: 'usuario_email_activo_unq',
  expression: `create unique index "usuario_email_activo_unq"
    on "usuario" ("email") where "deleted_at" is null`,
})
export class Usuario {
  @PrimaryKey() id!: string;
  @Property() email!: string;   // sin unique: lo aporta el índice parcial
  @Property({ nullable: true }) deletedAt?: Date;
}

-- OPCIÓN A · Índice único parcial (PostgreSQL, SQLite). La más limpia.
CREATE UNIQUE INDEX "usuario_email_activo_unq" ON "usuario" ("email") WHERE "deleted_at" IS NULL;
-- OPCIÓN B · Columna generada + único compuesto (MySQL 8, que no tiene índices parciales).
--            NULL no rompe la unicidad, pero aquí necesitamos un valor discriminante.
ALTER TABLE "usuario"
  ADD COLUMN "borrado_token" varchar(40)
  GENERATED ALWAYS AS (IF("deleted_at" IS NULL, 'ACTIVO', CONCAT('DEL-', "id"))) STORED;
CREATE UNIQUE INDEX "usuario_email_token_unq" ON "usuario" ("email", "borrado_token");
-- OPCIÓN C · Liberar el valor al borrar (destructiva pero simple):
UPDATE "usuario" SET "deleted_at" = now(), "email" = concat('borrado+', "id", '@invalid')
 WHERE "id" = 'u-1';
-- Pierdes el email original: útil precisamente cuando además hay que anonimizar.

17.9.2 Claves foráneas, integridad y derecho de supresión

• Las claves foráneas no se enteran. ON DELETE CASCADE nunca se dispara porque no hay DELETE. Si borras lógicamente un pedido, sus líneas siguen «vivas» y visibles: el borrado en cascada lógico tienes que programarlo tú, y debe ser transaccional.
• Las consultas por relación pueden mentir. Un JOIN con una tabla cuyo filtro de borrado no se aplica (SQL crudo, vistas, informes) devuelve filas que la aplicación considera inexistentes. Coherencia o dolor: si una entidad es borrable lógicamente, todo lo que la consulte debe saberlo.
• Los agregados y contadores. count(*) en informes, cuotas de plan, límites de uso: cada uno debe decidir explícitamente si cuenta lo borrado.
• El rendimiento. Las filas borradas siguen ocupando la tabla y sus índices. Con mucho borrado lógico, conviene un índice parcial (WHERE deleted_at IS NULL) y una purga periódica.

17.10 Eventos y suscriptores

MikroORM permite enganchar código al ciclo de vida de las entidades de dos formas: hooks (decoradores en métodos de la propia entidad) y suscriptores (clases que implementan EventSubscriber y observan varias entidades). Los hooks son locales y sencillos; los suscriptores son globales, testeables por separado y pueden inyectarse dependencias.

import { BeforeCreate, BeforeUpdate, Entity, EventArgs, OnInit } from '@mikro-orm/core';
@Entity()
export class Articulo {
  @PrimaryKey() id!: string;
  @Property() titulo!: string;
  @Property() slug!: string;
  @Property({ type: 'text' }) cuerpo!: string;
  @Property() palabras: number = 0;
  // Derivar datos de la propia entidad: uso legítimo y sin sorpresas.
  @BeforeCreate()
  @BeforeUpdate()
  recalcular(args: EventArgs<Articulo>): void {
    this.slug = slugify(this.titulo);
    this.palabras = this.cuerpo.trim().split(/\s+/).length;
    // El change set indica qué cambió realmente (útil para evitar trabajo).
    const cambios = args.changeSet?.payload;
    if (cambios && 'cuerpo' in cambios) this.revisionesCuerpo++;
  }
  @OnInit()
  inicializar(): void {
    this.slug ??= '';   // solo memoria: nada de await ni de consultas
  }
}

@Entity()
export class Pedido {
  // Lógica de negocio escondida en un hook:
  @AfterUpdate()
  async alActualizar(args: EventArgs<Pedido>) {
    if (this.estado === 'PAGADO') {
      // Efectos invisibles desde el caso de uso:
      await mailer.enviarFactura(this);       // no se deshace con ROLLBACK
      await almacen.reservarStock(this);      // ¿y si falla?
      await erp.sincronizar(this);            // 800 ms dentro del flush
    }
  }
}
// Problemas: no se dispara con nativeUpdate; sí se dispara al importar datos
// o al ejecutar un seeder; imposible de testear sin base de datos; el orden
// de los efectos depende del orden interno del flush.

@Injectable()
export class PedidosService {
  // El caso de uso es explícito y legible de arriba abajo.
  async marcarPagado(id: string): Promise<void> {
    await this.em.transactional(async (em) => {
      const p = await em.findOneOrFail(Pedido, id);
      p.marcarPagado();                       // invariantes en el dominio
      // Efectos externos: registrados de forma atómica,
      // ejecutados FUERA de la transacción (patrón outbox).
      em.create(EventoSaliente, { tipo: 'PedidoPagado', payload: { pedidoId: p.id } });
    });
  }
}
// Testeable sin base de datos, visible en la revisión,
// idempotente y con reintentos en el consumidor.

17.11 Auditoría: quién cambió qué y cuándo

import { ChangeSetType, EventSubscriber, FlushEventArgs } from '@mikro-orm/core';
// El usuario actual viaja en AsyncLocalStorage (capítulo 11), no como parámetro.
@Injectable()
export class AuditoriaSubscriber implements EventSubscriber {
  constructor(em: EntityManager, private readonly ctx: ContextoPeticion) {
    em.getEventManager().registerSubscriber(this);
  }
  // Limitar el alcance evita auditar la propia tabla de auditoría (bucle infinito).
  getSubscribedEntities(): string[] { return ['Pedido', 'Producto', 'Usuario', 'Tarifa']; }
  // onFlush: los change sets ya están calculados y seguimos DENTRO del flush,
  // así que las filas de auditoría se escriben en la misma transacción.
  async onFlush(args: FlushEventArgs): Promise<void> {
    const autor = this.ctx.usuarioId ?? 'sistema';
    let nuevos = 0;
    for (const cs of args.uow.getChangeSets()) {
      const antes: Record<string, unknown> = {};
      const despues: Record<string, unknown> = {};
      if (cs.type === ChangeSetType.UPDATE) {
        for (const campo of Object.keys(cs.payload)) {
          if (CAMPOS_SENSIBLES.has(campo)) continue;      // nunca audites contraseñas
          antes[campo] = (cs.originalEntity as Record<string, unknown>)?.[campo];
          despues[campo] = (cs.payload as Record<string, unknown>)[campo];
        }
        if (Object.keys(despues).length === 0) continue;   // cambio irrelevante
      }
      args.em.create(RegistroAuditoria, { entidad: cs.name, entidadId: String(cs.entity.id),
        operacion: cs.type, antes, despues, autor, fecha: new Date(),
        peticionId: this.ctx.correlacionId });
      nuevos++;
    }
    // Las entidades creadas aquí no entran en los change sets ya calculados:
    // hay que pedir explícitamente que se recalculen.
    if (nuevos > 0) args.uow.computeChangeSets();
  }
}

BEGIN;
UPDATE "tarifa" SET "importe" = 120 WHERE "id" = 't1' AND "version" = 3;
INSERT INTO "registro_auditoria"
  ("entidad", "entidad_id", "operacion", "antes", "despues", "autor", "peticion_id", "fecha")
VALUES ('Tarifa', 't1', 'update', '{"importe":100}', '{"importe":120}',
        'u-42', 'a1b2c3', now());
COMMIT;   -- el cambio y su rastro se confirman juntos: nunca uno sin el otro

17.12 Rendimiento

17.12.1 Operaciones masivas

El Unit of Work está optimizado para el caso normal: unas decenas de entidades por transacción, con seguimiento de cambios, hooks y relaciones. Para medio millón de filas ese seguimiento es puro coste. MikroORM ofrece atajos que no pasan por el Unit of Work, y por eso son rápidos y por eso hay que usarlos con los ojos abiertos.

// Importar 500.000 filas de un CSV.
async importar(filas: FilaCsv[]) {
  for (const f of filas) this.em.create(Cliente, mapear(f));
  // Un único flush con 500.000 entidades:
  //  · el Identity Map guarda 500.000 objetos
  //  · el UoW guarda otras 500.000 instantáneas y calcula 500.000 diffs
  //  · una transacción gigantesca de varios minutos
  await this.em.flush();
}
// Resultado real: heap de Node por encima de 2 GB, "JavaScript heap out of
// memory" y una transacción que bloquea el autovacuum mientras dura.

// Por lotes, con memoria constante y transacciones cortas.
async importar(filas: AsyncIterable<FilaCsv>) {
  const em = this.em.fork();          // contexto propio y aislado
  let buffer: EntityData<Cliente>[] = [];
  for await (const f of filas) {
    buffer.push(mapear(f));
    if (buffer.length >= 2_000) {
      await em.insertMany(Cliente, buffer);   // sin UoW: rápido
      buffer = []; em.clear();                // libera el Identity Map
    }
  }
  if (buffer.length) await em.insertMany(Cliente, buffer);
}
// Memoria estable (~1 lote), transacciones cortas y progreso reanudable.

17.12.2 El coste del Identity Map en procesos largos

El Identity Map es una tabla hash que garantiza «una fila, un objeto» dentro de un contexto. Su efecto secundario es que nada de lo que cargas se libera mientras el contexto viva: en una petición HTTP de 80 ms es exactamente lo que quieres; en un proceso que recorre un millón de filas es una fuga de memoria por diseño.

  Memoria del proceso durante un recorrido de 1.000.000 de filas
  sin em.clear()                        con em.clear() cada 2.000 filas
  ▲                                     ▲
  │                    ╱ OOM            │
  │                  ╱                  │
  │               ╱                     │  ▁▂▁▂▁▂▁▂▁▂▁▂▁▂▁▂▁▂▁▂▁▂
  │            ╱                        │
  │       ╱╱                            │
  └──────────────────────────► tiempo   └──────────────────────────► tiempo
   Identity Map + instantáneas del UoW    memoria acotada por el lote

• em.clear() vacía el Identity Map del contexto actual. Las entidades que tuvieras en variables locales quedan «desconectadas»: no se les seguirán los cambios.
• em.fork() crea un contexto nuevo. Es lo correcto para cada unidad de trabajo independiente (un mensaje de cola, un fichero, un inquilino).
• Nunca uses el EM global de la aplicación en un bucle largo. Su Identity Map crecerá hasta que el proceso muera.
• Para lecturas de gran volumen que no vas a modificar, evita cargar entidades: consulta con qb.execute() o pide objetos planos. Sin entidades no hay Identity Map ni instantáneas, y el consumo baja de forma drástica.

17.12.3 Pool de conexiones

export default defineConfig({
  pool: {
    min: 2,                        // conexiones siempre listas: evita latencia en frío
    max: 10,                       // TOPE por instancia de la aplicación
    acquireTimeoutMillis: 5_000,   // fallar rápido si el pool está saturado
    idleTimeoutMillis: 30_000,     // devolver conexiones ociosas
  },
  driverOptions: {
    connection: {
      statement_timeout: 15_000,                     // ninguna consulta debe durar más
      idle_in_transaction_session_timeout: 10_000,   // mata transacciones olvidadas
    },
  },
});

  Presupuesto de conexiones (PostgreSQL max_connections = 100)
  ┌────────────────────────────────────────────────────────────────┐
  │ 100  total del servidor                                        │
  │  -3  superusuario reservado (superuser_reserved_connections)    │
  │ -10  migraciones, copias de seguridad, panel, analítica        │
  │ ───────────────────────────────────────────────────────────────│
  │ = 87 disponibles para la aplicación                            │
  │  87 / (réplicas de la API + trabajadores de colas + crons) =    │
  │  87 / (4 pods + 2 workers + 1 cron) ≈ 12  →  pool.max = 10      │
  └────────────────────────────────────────────────────────────────┘
  Más conexiones NO es más rendimiento: por encima de unas 2-4 veces el número
  de núcleos de la base de datos, el rendimiento total BAJA por contención.
  Si necesitas mucha concurrencia, pon un pooler (PgBouncer) delante.

• Fuga de conexiones. Sus dos causas casi únicas son un begin() sin commit/rollback en alguna rama y una transacción que espera algo externo (una llamada HTTP, un bloqueo, la entrada del usuario). Se diagnostica con SELECT state, count(*) FROM pg_stat_activity GROUP BY state: si ves muchas idle in transaction, tienes una fuga.
• Timeout al adquirir. El error «timeout acquiring a connection» casi nunca significa «el pool es pequeño»: significa que alguien retiene conexiones demasiado tiempo. Subir max a ciegas traslada el problema al servidor de base de datos.
• Serverless y funciones. Cada instancia tiene su propio pool y las instancias se multiplican sin control: ahí un pooler externo o un driver HTTP no es opcional.
• Cierra bien. orm.close(true) al terminar un script o al recibir SIGTERM; si no, el proceso no muere y el orquestador acaba matándolo.

17.12.4 Caché de resultados, caché de metadatos y arranque

export default defineConfig({
  // 1. CACHÉ DE RESULTADOS: opcional por consulta, nunca global sin pensarlo.
  //    adapter: RedisCacheAdapter → necesario si hay varias réplicas.
  resultCache: { expiration: 5_000 },   // valor por defecto en milisegundos
  // 2. CACHÉ DE METADATOS: acelera el arranque. En producción, generada en el build.
  metadataCache: { enabled: true },
  // 3. PROVEEDOR DE METADATOS: de dónde salen los tipos.
  // metadataProvider: TsMorphMetadataProvider,   // analiza el código fuente
});
// Uso de la caché de resultados por consulta:
const top = await em.find(Producto, { destacado: true }, { cache: 60_000 });
const menu = await em.find(Categoria, {}, { cache: ['menu-principal', 300_000] });
await em.clearCache('menu-principal');   // invalidación explícita al editar categorías

17.12.5 Lista de comprobación de rendimiento del ORM

17.13 Multi-tenancy

  A · BASE DE DATOS POR INQUILINO      B · ESQUEMA POR INQUILINO       C · COLUMNA DISCRIMINADORA
  ┌──────────┐  ┌──────────┐           ┌───────── una base ────────┐   ┌──────── una base ────────┐
  │ db_acme  │  │ db_globex│           │ schema acme │ schema glob │   │ tabla pedido             │
  │ ┌──────┐ │  │ ┌──────┐ │           │ ┌────────┐  │ ┌────────┐  │   │ id │ tenant_id │ total   │
  │ │pedido│ │  │ │pedido│ │           │ │ pedido │  │ │ pedido │  │   │  1 │ acme      │  120    │
  │ └──────┘ │  │ └──────┘ │           │ └────────┘  │ └────────┘  │   │  2 │ globex    │   80    │
  └──────────┘  └──────────┘           └─────────────┴─────────────┘   └──────────────────────────┘
  Aislamiento máximo                   Aislamiento medio               Aislamiento lógico
  Coste operativo máximo               Coste medio                     Coste mínimo

// 1. La entidad declara el filtro con argumentos.
@Entity()
@Filter({ name: 'tenant', cond: (a: { tenantId: string }) => ({ tenantId: a.tenantId }),
          args: true, default: true })
export class Pedido {
  @PrimaryKey() id!: string;
  @Property({ index: true }) tenantId!: string;   // índice: entra en TODAS las consultas
  @Property() total!: number;
}
// 2. Un middleware fija el inquilino al principio de cada petición.
@Injectable()
export class TenantMiddleware implements NestMiddleware {
  constructor(private readonly em: EntityManager) {}
  use(req: Request, _res: Response, next: NextFunction): void {
    const tenantId = resolverInquilino(req);          // subdominio, JWT o cabecera
    if (!tenantId) throw new ForbiddenException('Inquilino no resuelto');
    this.em.setFilterParams('tenant', { tenantId });  // solo para ESTA petición
    next();
  }
}
// 3. Al crear entidades, el tenantId hay que ponerlo: el filtro solo filtra lecturas
//    y actualizaciones, no rellena valores. Un suscriptor beforeCreate es buen sitio.

// B · ESQUEMA POR INQUILINO: entidades con esquema comodín.
@Entity({ schema: '*' })
export class Pedido { /* ... */ }
// Un fork por petición con el esquema del inquilino resuelto:
const em = orm.em.fork();
em.schema = inquilino.esquema;               // afecta a las consultas de este contexto
const pedidos = await em.find(Pedido, {});   // SELECT ... FROM "acme"."pedido"
await orm.em.find(Pedido, {}, { schema: 'acme' });   // o bien por consulta
// Migraciones: una pasada por esquema, registrando el resultado de cada uno.
for (const t of await listarInquilinos()) await orm.migrator.up({ /* esquema de t */ });
// A · BASE DE DATOS POR INQUILINO: una instancia de MikroORM por inquilino,
// creadas de forma perezosa y con una caché con expiración para no agotar conexiones.
const orms = new Map<string, MikroORM>();
async function ormDe(inquilino: string): Promise<MikroORM> {
  if (!orms.has(inquilino)) {
    orms.set(inquilino, await MikroORM.init({ ...base, dbName: `db_${inquilino}`,
      pool: { min: 0, max: 2 } }));   // pool mínimo: son N instancias
  }
  return orms.get(inquilino)!;
}

17.14 Errores comunes y cómo solucionarlos

17.15 Buenas y malas prácticas

17.16 Preguntas frecuentes

17.17 Ejercicios

// src/cuentas/transferencias.service.ts
@Injectable()
export class TransferenciasService {
  constructor(private readonly em: EntityManager) {}
  async transferir(origenId: string, destinoId: string, importe: number): Promise<void> {
    if (importe <= 0) throw new BadRequestException('Importe no positivo');
    if (origenId === destinoId) throw new BadRequestException('Cuentas iguales');
    await conReintentos(() =>
      this.em.transactional(async (em) => {
        // 1. ORDEN ESTABLE: ambas direcciones bloquean en el mismo orden.
        //    Sin esto, A→B y B→A forman un ciclo y el motor mata una.
        const ids = [origenId, destinoId].sort();
        // 2. Bloqueo secuencial, nunca con Promise.all: el orden importa
        //    y dos consultas en paralelo sobre la misma conexión no lo garantizan.
        const cuentas = new Map<string, Cuenta>();
        for (const id of ids) {
          cuentas.set(id, await em.findOneOrFail(Cuenta, id,
            { lockMode: LockMode.PESSIMISTIC_WRITE }));
        }
        const origen = cuentas.get(origenId)!;
        const destino = cuentas.get(destinoId)!;
        // 3. La invariante se comprueba con el dato ya bloqueado:
        //    nadie puede cambiarlo entre la lectura y el UPDATE.
        if (origen.saldo < importe) throw new SaldoInsuficienteError(origenId);
        origen.saldo -= importe;
        destino.saldo += importe;
        em.create(Movimiento, { origen, destino, importe, fecha: new Date() });
      }),
    );
  }
}

it('conserva la suma de saldos con 50 transferencias cruzadas', async () => {
  await sembrar({ A: 1_000, B: 1_000 });
  // 25 en cada dirección, todas a la vez: sin orden estable, aquí saltan deadlocks.
  await Promise.all(Array.from({ length: 50 }, (_, i) =>
    i % 2 === 0 ? servicio.transferir('A', 'B', 10) : servicio.transferir('B', 'A', 10),
  ));
  const em = orm.em.fork();                 // contexto limpio: sin Identity Map viejo
  const [a, b] = await Promise.all([
    em.findOneOrFail(Cuenta, 'A'), em.findOneOrFail(Cuenta, 'B'),
  ]);
  expect(a.saldo + b.saldo).toBe(2_000);    // ninguna actualización perdida
  expect(await em.count(Movimiento, {})).toBe(50);
});

// src/migrations/Migration20260701090000_expand_nombre.ts
export class Migration20260701090000 extends Migration {
  async up(): Promise<void> {
    this.addSql(`set lock_timeout = '3s';`);
    this.addSql(`alter table "cliente" add column "nombre" varchar(60) null;`);
    this.addSql(`alter table "cliente" add column "apellidos" varchar(120) null;`);
  }
  async down(): Promise<void> {
    this.addSql(`alter table "cliente" drop column "apellidos";`);
    this.addSql(`alter table "cliente" drop column "nombre";`);
  }
}

// src/clientes/cliente.entity.ts (versión de transición)
setNombreCompleto(valor: string): void {
  this.nombreCompleto = valor;                    // formato antiguo (lo que se lee hoy)
  const [nombre, ...resto] = valor.trim().split(/\s+/);
  this.nombre = nombre;                           // formato nuevo (aún no se lee)
  this.apellidos = resto.join(' ') || '-';
}

// Reanudable: filtra por "nombre is null", así que repetirlo no hace daño.
for (;;) {
  const lote = await em.find(Cliente, { nombre: null }, { limit: 5_000, orderBy: { id: 'asc' } });
  if (!lote.length) break;
  for (const c of lote) c.setNombreCompleto(c.nombreCompleto);
  await em.flush();
  em.clear();
  await new Promise((r) => setTimeout(r, 100));
}

// src/migrations/Migration20260715090000_contract_nombre.ts
export class Migration20260715090000 extends Migration {
  async up(): Promise<void> {
    this.addSql(`set lock_timeout = '3s';`);
    // NOT NULL en dos fases: validar sin bloquear lecturas y después marcar.
    this.addSql(`alter table "cliente" add constraint "cliente_nombre_nn"
                   check ("nombre" is not null) not valid;`);
    this.addSql(`alter table "cliente" validate constraint "cliente_nombre_nn";`);
    this.addSql(`alter table "cliente" alter column "nombre" set not null;`);
    this.addSql(`alter table "cliente" drop constraint "cliente_nombre_nn";`);
    this.addSql(`alter table "cliente" drop column "nombre_completo";`);
  }
  async down(): Promise<void> {
    // Reversible en estructura, NO en datos: se documenta explícitamente.
    this.addSql(`alter table "cliente" add column "nombre_completo" varchar(180) null;`);
    this.addSql(`update "cliente" set "nombre_completo" = "nombre" || ' ' || "apellidos";`);
    this.addSql(`alter table "cliente" alter column "nombre" drop not null;`);
  }
}

// src/entities/usuario.entity.ts
@Entity()
@Filter({ name: 'softDelete', cond: { deletedAt: null }, default: true })
export class Usuario {
  @PrimaryKey() id: string = randomUUID();
  @Property() email!: string;                 // unicidad solo entre los NO borrados
  @Property() nombre!: string;
  @Property({ nullable: true, index: true }) deletedAt?: Date | null;
  @Property({ nullable: true }) deletedBy?: string | null;
}

// src/migrations/Migration20260710090000_email_unico_activos.ts
async up(): Promise<void> {
  // 1. Detecta duplicados preexistentes ANTES de crear el índice:
  //    si hay dos usuarios activos con el mismo correo, la creación fallará.
  this.addSql(`create unique index "usuario_email_activo_unq"
                 on "usuario" ("email") where "deleted_at" is null;`);
}
async down(): Promise<void> {
  this.addSql(`drop index "usuario_email_activo_unq";`);
}

// src/usuarios/usuarios.repository.ts
async borrar(id: string, autor: string): Promise<void> {
  await this.em.transactional(async (em) => {
    const u = await em.findOneOrFail(Usuario, id);
    u.deletedAt = new Date();
    u.deletedBy = autor;
    // Cascada LÓGICA: las claves foráneas no se enteran de un borrado lógico.
    await em.nativeUpdate(ApiToken, { usuario: id }, { deletedAt: u.deletedAt });
  });
}
async restaurar(id: string): Promise<void> {
  await this.em.transactional(async (em) => {
    const u = await em.findOneOrFail(Usuario, id, { filters: { softDelete: false } });
    // Antes de restaurar, comprobar que el correo sigue libre: otro usuario
    // pudo registrarse con él mientras este estaba borrado.
    if (await em.count(Usuario, { email: u.email, id: { $ne: id } })) {
      throw new ConflictException('El correo ya está en uso');
    }
    u.deletedAt = null;
    u.deletedBy = null;
  });
}
buscarIncluyendoBorrados(where: FilterQuery<Usuario>): Promise<Usuario[]> {
  return this.em.find(Usuario, where, { filters: { softDelete: false } });
}

it('permite volver a registrar un correo cuyo usuario fue borrado', async () => {
  const ana = await repo.crear({ email: 'ana@ejemplo.com', nombre: 'Ana' });
  await repo.borrar(ana.id, 'admin');
  orm.em.clear();
  // Sin el índice parcial: duplicate key value violates unique constraint
  const nueva = await repo.crear({ email: 'ana@ejemplo.com', nombre: 'Ana 2' });
  expect(nueva.id).not.toBe(ana.id);
  // Y la restauración ahora debe fallar de forma controlada, no con un 500.
  await expect(repo.restaurar(ana.id)).rejects.toThrow(ConflictException);
});

17.18 Resumen del capítulo

17.19 Recursos adicionales

• MikroORM · Transactions and concurrency — referencia de transactional, niveles de aislamiento, LockMode y bloqueo optimista.
• MikroORM · Migrations — configuración del Migrator, comandos y opciones de las migraciones.
• MikroORM · Schema generator — qué hace exactamente cada operación y sus límites.
• MikroORM · Seeding — Seeder, Factory e integración con faker.
• MikroORM · Filters — filtros de entidad y globales, parámetros y activación por consulta.
• MikroORM · Events and hooks — lista completa de eventos y qué es seguro hacer en cada uno.
• MikroORM · Caching — caché de resultados, adaptadores y claves.
• MikroORM · Multiple schemas — base para las estrategias de multi-tenancy por esquema.
• MikroORM · Usage with NestJS — contexto de petición, @CreateRequestContext e inyección del EntityManager.
• PostgreSQL · Transaction isolation — la explicación canónica de las anomalías y de los niveles reales.
• PostgreSQL · Explicit locking — FOR UPDATE, SKIP LOCKED, bloqueos de tabla y advisory locks.
• MySQL · InnoDB transaction isolation levels — diferencias de implementación frente a PostgreSQL.
• Martin Fowler · Parallel Change — la formulación original del patrón expand/contract.