Anti-Corruption Layer: aísla el legacy con adaptadores

Resumen

El anti-corruption layer (ACL) aísla un modelo de dominio nuevo del modelo de un sistema legacy. Cuando un sistema nuevo necesita interactuar con un sistema legacy, los conceptos, estructuras de datos y APIs del legacy pueden “corromper” el modelo de dominio nuevo. La ACL actúa como capa de traducción: acepta peticiones en el lenguaje del dominio nuevo, las traduce al formato del sistema legacy, llama al legacy y traduce la respuesta de vuelta. El dominio nuevo nunca ve las estructuras de datos del legacy. Este patrón es esencial durante migraciones en las que el sistema nuevo debe coexistir con el antiguo, y combina bien con el patrón strangler fig. En el vocabulario de Evans, la ACL es cómo un bounded context mantiene limpio su propio lenguaje ubicuo.

Cuándo Usarlo

• Migrando desde un sistema legacy mientras mantienes la integración
• Un modelo de dominio nuevo que difiere del modelo de datos legacy
• Integrando con sistemas externos que tienen conceptos de dominio distintos
• Evitando que la terminología legacy se filtre al código nuevo
• Microservicios que necesitan llamar a endpoints de un monolito legacy

Cuándo No Usarlo

• Cuando el sistema nuevo y el legacy comparten el mismo modelo de dominio
• Cuando el sistema legacy se reemplaza por completo (no hace falta integración)
• Cuando la traducción es trivial (mapeo de campos 1:1)
• Proyectos greenfield sin integración con legacy

Cómo Funciona la Frontera

La ACL es una capa de traducción, no un wrapper — convierte el modelo legacy al modelo de dominio que tu código nuevo entiende, en ambas direcciones. Ida: tu código pide un “Cliente”, la ACL lo traduce a lo que sea que USR_MAST lo llame. Vuelta: el concepto de “cuenta” del sistema legacy nunca se filtra a tu lenguaje de dominio. Ese es el punto que hacía Eric Evans: la corrupción es el modelo, no la tecnología.

Tres cosas viven dentro: el traductor (objeto de dominio ↔ DTO legacy), la fachada (la interfaz que tu dominio llama realmente — habla tu idioma, no el suyo), y el contrato anti-corrupción (lo que el sistema legacy promete, expresado en términos legacy — léelo, no confíes en él).

Cuándo Te Salva vs. Cuándo Es Overkill

La ACL gana su complejidad cuando el modelo legacy es genuinamente corrupto — nombres distintos para lo mismo, un concepto repartido en cinco tablas, reglas de negocio codificadas en stored procedures que nadie lee. Contra una API externa limpia con schema cuerdo, una ACL es una clase que renombra campos — escribe un adapter normal y salta la ceremonia.

La señal: si sigues escribiendo comentarios “ellos le llaman X a esto” en la lógica de negocio, la frontera no aguanta. La ACL es donde esos comentarios van a vivir — en código, en un solo lugar, donde pueden testearse.

Qué Va Dentro vs. Qué Va Fuera

Dentro de la ACL	Fuera de ella
Traducción de modelos	Reglas de negocio
Llamadas a la API legacy	Lógica de dominio
Mapeo defensivo	Validación de tu modelo
Manejo de errores legacy	Tu taxonomía de errores

La lógica de negocio se queda fuera — en el momento en que la ACL empieza a decidir cosas, se convirtió en parte del dominio y la frontera falló. Su trabajo es representación, no interpretación.

Solución

Facade ACL (Python)

# acl/customer_acl.py — capa anti-corrupción para el dominio customer
from dataclasses import dataclass
from typing import Optional
from legacy.client import LegacyCustomerClient

# Modelo de dominio nuevo — limpio, usa lenguaje del dominio
@dataclass(frozen=True)
class Customer:
    id: str
    email: str
    full_name: str
    status: CustomerStatus
    created_at: str

@dataclass(frozen=True)
class CustomerStatus:
    value: str  # "active", "suspended", "churned"

    @staticmethod
    def active() -> 'CustomerStatus':
        return CustomerStatus("active")

    @staticmethod
    def suspended() -> 'CustomerStatus':
        return CustomerStatus("suspended")

# Traductor ACL — convierte entre el legacy y el dominio nuevo
class CustomerTranslator:
    """Traduce entre las estructuras de datos legacy y el modelo de dominio nuevo."""

    @staticmethod
    def from_legacy(legacy_data: dict) -> Customer:
        """Traduce la respuesta de la API legacy a un Customer de dominio."""
        # El legacy usa 'cust_id', 'email_addr', 'first_nm', 'last_nm', 'stat_cd'
        # El dominio nuevo usa 'id', 'email', 'full_name', 'status'
        full_name = f"{legacy_data.get('first_nm', '')} {legacy_data.get('last_nm', '')}".strip()

        # Códigos de estado legacy: 1=active, 2=suspended, 3=churned
        status_map = {"1": "active", "2": "suspended", "3": "churned"}
        status = status_map.get(legacy_data.get("stat_cd", "1"), "active")

        return Customer(
            id=legacy_data["cust_id"],
            email=legacy_data["email_addr"].lower(),
            full_name=full_name,
            status=CustomerStatus(status),
            created_at=legacy_data.get("create_dt", "")
        )

    @staticmethod
    def to_legacy(customer: Customer) -> dict:
        """Traduce un Customer de dominio al formato de la API legacy."""
        # Separa full_name en first/last para el legacy
        parts = customer.full_name.split(" ", 1)
        first_name = parts[0]
        last_name = parts[1] if len(parts) > 1 else ""

        # Mapeo inverso de estados
        status_map = {"active": "1", "suspended": "2", "churned": "3"}

        return {
            "cust_id": customer.id,
            "email_addr": customer.email,
            "first_nm": first_name,
            "last_nm": last_name,
            "stat_cd": status_map.get(customer.status.value, "1")
        }

# Facade ACL — la interfaz pública que usa el dominio nuevo
class CustomerACL:
    """Fachada de la capa anti-corrupción.
    El dominio nuevo llama a esto — nunca al cliente legacy directamente."""

    def __init__(self, legacy_client: LegacyCustomerClient):
        self._legacy = legacy_client
        self._translator = CustomerTranslator()

    def get_customer(self, customer_id: str) -> Optional[Customer]:
        legacy_data = self._legacy.fetch_customer(customer_id)
        if legacy_data is None:
            return None
        return self._translator.from_legacy(legacy_data)

    def create_customer(self, customer: Customer) -> Customer:
        legacy_data = self._translator.to_legacy(customer)
        result = self._legacy.insert_customer(legacy_data)
        return self._translator.from_legacy(result)

    def update_customer(self, customer: Customer) -> Customer:
        legacy_data = self._translator.to_legacy(customer)
        result = self._legacy.update_customer(customer.id, legacy_data)
        return self._translator.from_legacy(result)

    def suspend_customer(self, customer_id: str) -> Customer:
        legacy_data = self._legacy.update_status(customer_id, "2")
        return self._translator.from_legacy(legacy_data)

Servicio de dominio nuevo usando la ACL

# domain/customer_service.py — el servicio de dominio nuevo usa la ACL, no el legacy
from acl.customer_acl import CustomerACL, Customer, CustomerStatus

class CustomerService:
    """Servicio de dominio nuevo — no sabe nada del sistema legacy."""

    def __init__(self, acl: CustomerACL):
        self._acl = acl

    def onboard_customer(self, email: str, full_name: str) -> Customer:
        customer = Customer(
            id="",  # la ACL lo asigna
            email=email,
            full_name=full_name,
            status=CustomerStatus.active(),
            created_at=""
        )
        return self._acl.create_customer(customer)

    def suspend(self, customer_id: str) -> Customer:
        return self._acl.suspend_customer(customer_id)

    def get_profile(self, customer_id: str) -> dict:
        customer = self._acl.get_customer(customer_id)
        if customer is None:
            raise ValueError(f"Customer {customer_id} not found")
        return {
            "id": customer.id,
            "email": customer.email,
            "name": customer.full_name,
            "status": customer.status.value
        }

ACL en Java con Spring

// CustomerACL.java — capa anti-corrupción en Java
package com.shop.acl;

import com.shop.domain.Customer;
import com.shop.domain.CustomerStatus;
import com.legacy.client.LegacyCustomerClient;
import com.legacy.dto.LegacyCustomerDTO;
import org.springframework.stereotype.Component;

@Component
public class CustomerACL {

    private final LegacyCustomerClient legacyClient;

    public CustomerACL(LegacyCustomerClient legacyClient) {
        this.legacyClient = legacyClient;
    }

    public Customer findById(String customerId) {
        LegacyCustomerDTO legacy = legacyClient.getCustomer(customerId);
        if (legacy == null) return null;
        return translateFromLegacy(legacy);
    }

    public Customer save(Customer customer) {
        LegacyCustomerDTO legacy = translateToLegacy(customer);
        LegacyCustomerDTO saved = legacyClient.insert(legacy);
        return translateFromLegacy(saved);
    }

    public Customer suspend(String customerId) {
        LegacyCustomerDTO legacy = legacyClient.updateStatus(customerId, "2");
        return translateFromLegacy(legacy);
    }

    private Customer translateFromLegacy(LegacyCustomerDTO legacy) {
        String fullName = legacy.getFirstName() + " " + legacy.getLastName();
        CustomerStatus status = switch (legacy.getStatusCode()) {
            case "1" -> CustomerStatus.active();
            case "2" -> CustomerStatus.suspended();
            case "3" -> CustomerStatus.churned();
            default -> CustomerStatus.active();
        };

        return new Customer(
            legacy.getCustomerId(),
            legacy.getEmailAddress().toLowerCase(),
            fullName.trim(),
            status,
            legacy.getCreationDate()
        );
    }

    private LegacyCustomerDTO translateToLegacy(Customer customer) {
        String[] nameParts = customer.getFullName().split(" ", 2);
        String firstName = nameParts[0];
        String lastName = nameParts.length > 1 ? nameParts[1] : "";

        String statusCode = switch (customer.getStatus()) {
            case ACTIVE -> "1";
            case SUSPENDED -> "2";
            case CHURNED -> "3";
        };

        LegacyCustomerDTO dto = new LegacyCustomerDTO();
        dto.setCustomerId(customer.getId());
        dto.setEmailAddress(customer.getEmail());
        dto.setFirstName(firstName);
        dto.setLastName(lastName);
        dto.setStatusCode(statusCode);
        return dto;
    }
}

ACL para integración por eventos

# acl/event_acl.py — traduce eventos legacy a eventos de dominio
from typing import Callable, Dict, List
from dataclasses import dataclass

@dataclass
class LegacyEvent:
    event_type: str  # "CUST_ADD", "CUST_MOD", "CUST_DEL"
    cust_id: str
    email_addr: str
    stat_cd: str
    timestamp: str

@dataclass
class DomainEvent:
    event_type: str  # "CustomerCreated", "CustomerUpdated", "CustomerDeleted"
    customer_id: str
    email: str
    status: str
    occurred_at: str

class EventTranslator:
    """Traduce eventos legacy a eventos de dominio."""

    TYPE_MAP = {
        "CUST_ADD": "CustomerCreated",
        "CUST_MOD": "CustomerUpdated",
        "CUST_DEL": "CustomerDeleted"
    }

    STATUS_MAP = {"1": "active", "2": "suspended", "3": "churned"}

    @staticmethod
    def to_domain(legacy: LegacyEvent) -> DomainEvent:
        return DomainEvent(
            event_type=EventTranslator.TYPE_MAP.get(legacy.event_type, "Unknown"),
            customer_id=legacy.cust_id,
            email=legacy.email_addr.lower() if legacy.email_addr else "",
            status=EventTranslator.STATUS_MAP.get(legacy.stat_cd, "active"),
            occurred_at=legacy.timestamp
        )

class EventACL:
    """Se suscribe a eventos legacy y publica eventos de dominio."""

    def __init__(self):
        self._translator = EventTranslator()
        self._handlers: Dict[str, List[Callable]] = {}

    def subscribe(self, event_type: str, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    def on_legacy_event(self, legacy_event: LegacyEvent):
        """Llamado cuando llega un evento legacy."""
        domain_event = self._translator.to_domain(legacy_event)

        handlers = self._handlers.get(domain_event.event_type, [])
        for handler in handlers:
            handler(domain_event)

ACL para integración por base de datos

# acl/database_acl.py — traduce tablas legacy a repositorios de dominio
from typing import Optional, List
from domain.customer import Customer, CustomerStatus, CustomerRepository

class LegacyCustomerRepository(CustomerRepository):
    """Implementa la interfaz de repositorio del dominio sobre tablas legacy.
    El dominio ve un repositorio limpio; esta clase hace la traducción."""

    def __init__(self, db_connection):
        self._db = db_connection

    def find_by_id(self, customer_id: str) -> Optional[Customer]:
        # Consulta la tabla legacy con nombres de columna legacy
        row = self._db.execute_one(
            "SELECT cust_id, email_addr, first_nm, last_nm, stat_cd, create_dt "
            "FROM LEGACY.CUSTOMERS WHERE cust_id = :id",
            {"id": customer_id}
        )
        if row is None:
            return None

        # Traduce la fila legacy a la entidad de dominio
        status_map = {"1": "active", "2": "suspended", "3": "churned"}
        return Customer(
            id=row["cust_id"],
            email=row["email_addr"].lower(),
            full_name=f"{row['first_nm']} {row['last_nm']}".strip(),
            status=CustomerStatus(status_map.get(row["stat_cd"], "active")),
            created_at=str(row["create_dt"])
        )

    def save(self, customer: Customer) -> Customer:
        # Traduce la entidad de dominio a columnas legacy
        parts = customer.full_name.split(" ", 1)
        status_map = {"active": "1", "suspended": "2", "churned": "3"}

        self._db.execute(
            "MERGE INTO LEGACY.CUSTOMERS "
            "USING (SELECT :id AS cust_id FROM dual) src "
            "ON (CUSTOMERS.cust_id = src.cust_id) "
            "WHEN MATCHED THEN UPDATE SET "
            "  email_addr = :email, first_nm = :first, last_nm = :last, stat_cd = :status "
            "WHEN NOT MATCHED THEN INSERT "
            "  (cust_id, email_addr, first_nm, last_nm, stat_cd, create_dt) "
            "  VALUES (:id, :email, :first, :last, :status, SYSDATE)",
            {
                "id": customer.id,
                "email": customer.email,
                "first": parts[0],
                "last": parts[1] if len(parts) > 1 else "",
                "status": status_map.get(customer.status.value, "1")
            }
        )
        return self.find_by_id(customer.id)

    def find_by_status(self, status: CustomerStatus) -> List[Customer]:
        status_map = {"active": "1", "suspended": "2", "churned": "3"}
        rows = self._db.execute_all(
            "SELECT cust_id, email_addr, first_nm, last_nm, stat_cd, create_dt "
            "FROM LEGACY.CUSTOMERS WHERE stat_cd = :status",
            {"status": status_map.get(status.value, "1")}
        )
        return [self._row_to_customer(row) for row in rows]

Variantes

ACL como API gateway

# acl_gateway.py — la ACL expuesta como API HTTP
from flask import Flask, request, jsonify
from acl.customer_acl import CustomerACL
from legacy.client import LegacyCustomerClient

app = Flask(__name__)
acl = CustomerACL(LegacyCustomerClient())

@app.route("/api/customers/<customer_id>", methods=["GET"])
def get_customer(customer_id):
    customer = acl.get_customer(customer_id)
    if customer is None:
        return jsonify({"error": "Not found"}), 404
    return jsonify({
        "id": customer.id,
        "email": customer.email,
        "fullName": customer.full_name,
        "status": customer.status.value
    })

@app.route("/api/customers", methods=["POST"])
def create_customer():
    data = request.json
    customer = Customer(
        id="",
        email=data["email"],
        full_name=data["fullName"],
        status=CustomerStatus.active(),
        created_at=""
    )
    created = acl.create_customer(customer)
    return jsonify({
        "id": created.id,
        "email": created.email,
        "fullName": created.full_name,
        "status": created.status.value
    }), 201

ACL con caché

# acl_cached.py — la ACL con caché para reducir las llamadas al legacy
from functools import lru_cache
from acl.customer_acl import CustomerACL, Customer

class CachedCustomerACL:
    """Envuelve la ACL con caché para reducir la carga del sistema legacy."""

    def __init__(self, inner_acl: CustomerACL):
        self._inner = inner_acl

    @lru_cache(maxsize=1000)
    def get_customer(self, customer_id: str) -> Customer:
        return self._inner.get_customer(customer_id)

    def create_customer(self, customer: Customer) -> Customer:
        result = self._inner.create_customer(customer)
        # Cachea el customer nuevo
        self.get_customer.cache_clear()
        return result

    def update_customer(self, customer: Customer) -> Customer:
        result = self._inner.update_customer(customer)
        # Invalida la caché de este customer
        self.get_customer.cache_clear()
        return result

    def suspend_customer(self, customer_id: str) -> Customer:
        result = self._inner.suspend_customer(customer_id)
        self.get_customer.cache_clear()
        return result

Buenas Prácticas

• For a deeper guide, see Strangler Fig: Gradually Replace Legacy by Intercepting.

• Mantén la ACL sin estado — solo debe traducir, no guardar estado de negocio

• Haz de la ACL el único camino hacia el sistema legacy — ningún código de dominio debería llamar al legacy directamente

• Prueba las traducciones a fondo — los errores de mapeo causan corrupción silenciosa de datos

• Usa la ACL tanto para lecturas como para escrituras — no dejes que ningún camino se salte la traducción

• Mantén los traductores bidireccionales — from_legacy y to_legacy deberían ser operaciones inversas

• Versiona la ACL — cuando la API legacy cambie, actualiza el traductor, no el dominio

• Considera usar caché — si el sistema legacy es lento, cachea los resultados traducidos

• Planifica la retirada de la ACL — cuando el legacy esté reemplazado del todo, la ACL se puede borrar

Errores Comunes

• Código de dominio llamando al legacy directamente: se salta la ACL. El modelo legacy corrompe el dominio. Asegúrate de que solo la ACL habla con el legacy.
• Traducción incompleta: algunos campos legacy se filtran al dominio. Traduce cada campo que leas — los que parecen poco importantes son los que se cuelan.
• Sin tests para los traductores: los errores de mapeo pasan desapercibidos. Prueba from_legacy y to_legacy con todos los valores posibles.
• La ACL con lógica de negocio dentro: la ACL traduce, no decide. La lógica de negocio va en el servicio de dominio, no en la ACL.
• Sin estrategia de caché: cada operación de dominio golpea al sistema legacy. Añade caché para las operaciones de mucha lectura.

Ejemplos Trabajados

De un registro SOAP a un Customer del dominio

Un pase de ACL real, de punta a punta. El servicio SOAP legacy devuelve esto al pedir un cliente:

<customer>
  <cust_id>8841</cust_id>
  <nm>J. SMITH&nbsp;&nbsp;</nm>
  <stat_cd>2</stat_cd>
  <crncy>1</crncy>
  <last_purch_dt>20231114</last_purch_dt>
</customer>

El dominio no quiere nada de eso: nombres con padding, códigos de estado mágicos, enums de moneda, fechas YYYYMMDD. La ACL se hace cargo de cada una de esas traducciones:

// acl/LegacyCustomerTranslator.ts — el único archivo que sabe que stat_cd existe
const STATUS_MAP: Record<string, CustomerStatus> = {
  '1': CustomerStatus.active,
  '2': CustomerStatus.suspended,
  '3': CustomerStatus.closed,
};
const CURRENCY_MAP: Record<string, Currency> = {
  '1': 'USD', '2': 'EUR', '3': 'GBP',
};

export function toDomain(raw: LegacyCustomerXml): Customer {
  return {
    id: String(raw.cust_id),
    name: raw.nm.trim(),                      // "J. SMITH  " → "J. SMITH"
    status: STATUS_MAP[raw.stat_cd] ?? CustomerStatus.unknown,
    currency: CURRENCY_MAP[raw.crncy] ?? 'USD',
    lastPurchase: parseLegacyDate(raw.last_purch_dt),  // 20231114 → Date
  };
}

Tres propiedades hacen de esto una ACL real y no solo un mapper. Maneja los casos raros — un stat_cd=7 que nadie documentó cae en CustomerStatus.unknown en vez de romper el checkout. Falla en voz alta en la frontera: un payload malformado lanza error dentro del traductor, así que la rotura del legacy aparece como “error de ACL” y no como un misterio en el dominio. Y el lado del dominio (Customer) nunca importa un tipo legacy — borra el archivo cuando el legacy muera y nada más cambia.

La otra cara es honesta: la ACL se paga lento. Cada campo nuevo del legacy es otra línea en el mapa, y el mapa crece con la migración. Ese impuesto de mantenimiento es exactamente lo que la migración strangler fig acepta — financias la frontera para que el lado nuevo se mantenga limpio.

De una fila plana CUST_ACCT a objetos de dominio tipados

Digamos que el sistema legacy llama al cliente CUST_ACCT con CUST_ID, ACCT_NM, CR_LMT_AMT — un registro que junta identidad, crédito y estado de cuenta en una sola fila plana. Tu dominio tiene Client, CreditLimit, AccountStatus como conceptos separados.

El traductor de la ACL hace el mapeo para que el dominio nunca vea CUST_ACCT:

// Shape legacy — lo que USR_MAST devuelve realmente
interface LegacyCustAcct {
  CUST_ID: string;
  ACCT_NM: string;
  CR_LMT_AMT: number;
  ACCT_STAT_CD: 'A' | 'S' | 'C';
}

// Shape de dominio — con lo que tu código trabaja realmente
interface Client {
  id: string;
  name: string;
  creditLimit: Money;
  status: 'active' | 'suspended' | 'closed';
}

class LegacyToClientTranslator {
  toDomain(legacy: LegacyCustAcct): Client {
    return {
      id: legacy.CUST_ID,
      name: legacy.ACCT_NM,
      creditLimit: { amount: legacy.CR_LMT_AMT, currency: 'USD' },
      status: { A: 'active', S: 'suspended', C: 'closed' }[legacy.ACCT_STAT_CD],
    };
  }
}

Ahora el dominio pide un Client y la ACL maneja la llamada a USR_MAST, la traducción, y lo que sea que la API legacy lance. El vocabulario CUST_ACCT nunca pasa de la frontera.

Lectura Adicional

• Eric Evans — Domain-Driven Design — la ACL es el patrón de frontera del capítulo 13
• Microsoft — Patrón Anti-Corruption Layer — la formulación cloud-era
• Patrón Adapter para APIs — cuando la frontera no necesita una ACL completa