diff --git a/.env.example b/.env.example index 74481c2..6b0adea 100644 --- a/.env.example +++ b/.env.example @@ -112,3 +112,23 @@ SYNC_SECRET_TOKEN=change-this-sync-token-in-production # Lista de spokes (Solo si es HUB y desea retransmitir a otros - Opcional) SPOKE_URLS="" + +# ── EFC (expediente electronico) ───────────────────────────────────────────── +# Carril CRM -> EFC: los documentos del CRM se resguardan en el expediente de EFC. +# Los nombres son los MISMOS que usa el gateway de Anexo22 contra el mismo EFC. +# +# EFC_API_URL VACIA = integracion APAGADA. Todo el enganche es best-effort y hace no-op: +# el CRM sigue funcionando igual, guardando los archivos solo en su MinIO. +# EFC_API_KEY debe coincidir con CRM_INTEGRATION_API_KEY del lado de EFC. +EFC_API_URL= +EFC_API_KEY= +EFC_API_VERIFY_SSL=true +# Metadatos: resolver organizacion, crear expediente, completar. +EFC_API_TIMEOUT_MS=8000 +# Subidas. Debe quedar POR DEBAJO del proxy_read_timeout del nginx de EFC: si el CRM +# esperara mas, veria un 504 opaco y no sabria si el documento entro. +EFC_UPLOAD_TIMEOUT_MS=55000 +# Scaffolding de mTLS (pre-produccion). Vacio = TLS normal. +EFC_MTLS_CA_PATH= +EFC_MTLS_CERT_PATH= +EFC_MTLS_KEY_PATH= diff --git a/backend/alembic/versions/f7a8b9c0d1e2_crm_efc_outbox.py b/backend/alembic/versions/f7a8b9c0d1e2_crm_efc_outbox.py new file mode 100644 index 0000000..b76d63b --- /dev/null +++ b/backend/alembic/versions/f7a8b9c0d1e2_crm_efc_outbox.py @@ -0,0 +1,113 @@ +"""Outbox del carril CRM -> EFC: crm.efc_sync_outbox (expedientes) y crm.efc_file_outbox (archivos). + +Revision ID: f7a8b9c0d1e2 +Revises: e6f7a8b9c0d1 +Create Date: 2026-08-07 00:00:00.000000 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "f7a8b9c0d1e2" +down_revision: Union[str, None] = "e6f7a8b9c0d1" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ---------- crm.efc_sync_outbox: metadatos (alta del provisional y completado) ---------- + op.create_table( + "efc_sync_outbox", + sa.Column("id", sa.Integer(), nullable=False, autoincrement=True), + sa.Column("kind", sa.String(length=20), nullable=False), + sa.Column("payload", sa.JSON(), nullable=False), + sa.Column("expediente_ref", sa.Integer(), nullable=True), + sa.Column("status", sa.String(length=10), nullable=False, server_default=sa.text("'pending'")), + sa.Column("attempts", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("sent_at", sa.DateTime(), nullable=True), + sa.Column("efc_pedimento_id", sa.String(length=36), nullable=True), + sa.Column("tenant_id", sa.Integer(), nullable=False), + sa.Column("company_id", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")), + sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")), + sa.Column("deleted_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + schema="crm", + ) + op.create_index("ix_crm_efc_sync_outbox_status", "efc_sync_outbox", ["status"], schema="crm") + op.create_index("ix_crm_efc_sync_outbox_kind_status", "efc_sync_outbox", ["kind", "status"], schema="crm") + op.create_index("ix_crm_efc_sync_outbox_expediente_ref", "efc_sync_outbox", ["expediente_ref"], schema="crm") + op.create_index("ix_crm_efc_sync_outbox_tenant_id", "efc_sync_outbox", ["tenant_id"], schema="crm") + op.create_index("ix_crm_efc_sync_outbox_company_id", "efc_sync_outbox", ["company_id"], schema="crm") + op.create_foreign_key( + "fk_crm_efc_sync_outbox_tenant_id", "efc_sync_outbox", "tenants", + ["tenant_id"], ["id"], source_schema="crm", referent_schema="core", + ) + + # ---------- crm.efc_file_outbox: archivos ---------- + op.create_table( + "efc_file_outbox", + sa.Column("id", sa.Integer(), nullable=False, autoincrement=True), + sa.Column("kind", sa.String(length=30), nullable=False), + sa.Column("s3_key", sa.String(length=1024), nullable=False), + sa.Column("file_name", sa.String(length=255), nullable=False), + sa.Column("content_type", sa.String(length=100), nullable=True), + sa.Column("efc_tipo", sa.String(length=40), nullable=False), + # La pareja (tabla, id) desambigua entre las DOS secuencias de documentos del CRM: + # crm.documents.id = 5 y ops.shipment_documents.id = 5 coexisten. + sa.Column("source_table", sa.String(length=30), nullable=False), + sa.Column("source_id", sa.Integer(), nullable=True), + sa.Column("crm_document_ref", sa.String(length=64), nullable=True), + sa.Column("expediente_ref", sa.Integer(), nullable=False), + sa.Column("delete_local", sa.Boolean(), nullable=False, server_default=sa.text("true")), + sa.Column("status", sa.String(length=10), nullable=False, server_default=sa.text("'pending'")), + sa.Column("attempts", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("sent_at", sa.DateTime(), nullable=True), + sa.Column("efc_document_id", sa.String(length=36), nullable=True), + sa.Column("tenant_id", sa.Integer(), nullable=False), + sa.Column("company_id", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")), + sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")), + sa.Column("deleted_at", sa.DateTime(), nullable=True), + sa.PrimaryKeyConstraint("id"), + schema="crm", + ) + op.create_index("ix_crm_efc_file_outbox_status", "efc_file_outbox", ["status"], schema="crm") + op.create_index("ix_crm_efc_file_outbox_kind_status", "efc_file_outbox", ["kind", "status"], schema="crm") + # Índice de la guarda _ya_entregado, que es lo que se consulta en cada encolado. + op.create_index("ix_crm_efc_file_outbox_source", "efc_file_outbox", ["source_table", "source_id"], schema="crm") + op.create_index("ix_crm_efc_file_outbox_expediente_ref", "efc_file_outbox", ["expediente_ref"], schema="crm") + op.create_index("ix_crm_efc_file_outbox_tenant_id", "efc_file_outbox", ["tenant_id"], schema="crm") + op.create_index("ix_crm_efc_file_outbox_company_id", "efc_file_outbox", ["company_id"], schema="crm") + op.create_foreign_key( + "fk_crm_efc_file_outbox_tenant_id", "efc_file_outbox", "tenants", + ["tenant_id"], ["id"], source_schema="crm", referent_schema="core", + ) + op.create_foreign_key( + "fk_crm_efc_file_outbox_expediente_ref", "efc_file_outbox", "expedientes", + ["expediente_ref"], ["id"], source_schema="crm", referent_schema="crm", + ) + + +def downgrade() -> None: + op.drop_constraint("fk_crm_efc_file_outbox_expediente_ref", "efc_file_outbox", schema="crm", type_="foreignkey") + op.drop_constraint("fk_crm_efc_file_outbox_tenant_id", "efc_file_outbox", schema="crm", type_="foreignkey") + op.drop_index("ix_crm_efc_file_outbox_company_id", table_name="efc_file_outbox", schema="crm") + op.drop_index("ix_crm_efc_file_outbox_tenant_id", table_name="efc_file_outbox", schema="crm") + op.drop_index("ix_crm_efc_file_outbox_expediente_ref", table_name="efc_file_outbox", schema="crm") + op.drop_index("ix_crm_efc_file_outbox_source", table_name="efc_file_outbox", schema="crm") + op.drop_index("ix_crm_efc_file_outbox_kind_status", table_name="efc_file_outbox", schema="crm") + op.drop_index("ix_crm_efc_file_outbox_status", table_name="efc_file_outbox", schema="crm") + op.drop_table("efc_file_outbox", schema="crm") + + op.drop_constraint("fk_crm_efc_sync_outbox_tenant_id", "efc_sync_outbox", schema="crm", type_="foreignkey") + op.drop_index("ix_crm_efc_sync_outbox_company_id", table_name="efc_sync_outbox", schema="crm") + op.drop_index("ix_crm_efc_sync_outbox_tenant_id", table_name="efc_sync_outbox", schema="crm") + op.drop_index("ix_crm_efc_sync_outbox_expediente_ref", table_name="efc_sync_outbox", schema="crm") + op.drop_index("ix_crm_efc_sync_outbox_kind_status", table_name="efc_sync_outbox", schema="crm") + op.drop_index("ix_crm_efc_sync_outbox_status", table_name="efc_sync_outbox", schema="crm") + op.drop_table("efc_sync_outbox", schema="crm") diff --git a/backend/api/v1/modules/crm/expediente_gateway/__init__.py b/backend/api/v1/modules/crm/expediente_gateway/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/api/v1/modules/crm/expediente_gateway/models.py b/backend/api/v1/modules/crm/expediente_gateway/models.py new file mode 100644 index 0000000..a924f7a --- /dev/null +++ b/backend/api/v1/modules/crm/expediente_gateway/models.py @@ -0,0 +1,131 @@ +"""Outbox transaccional del carril CRM Agentes de Carga -> EFC. + +DOS tablas separadas POR PROPÓSITO, igual que en el carril de referencia de Anexo22: una para los +expedientes (metadatos, JSON) y otra para los archivos (binarios que viven en MinIO y se referencian +por su ``s3_key``). Un worker de Celery las drena hacia EFC con reintentos. + +**Diferencia con el original, y es necesaria:** aquí las filas se insertan en la MISMA transacción +que el expediente o el documento, porque el CRM es mono-base. En Anexo22 el outbox vivía en otra +base que el pedimento, y ese doble-commit es justamente lo que obligó a inventar el barrido de +huecos. Aquí el barrido se conserva —cubre lo creado antes de activar la integración y cualquier +crash— pero deja de ser el parche de una ventana estructural. +""" +from datetime import datetime +from typing import Optional + +from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Index, Integer, String, Text, text +from sqlalchemy.orm import Mapped, mapped_column + +from api.v1.common.base_models import TenantScopedMixin, TimestampMixin +from core.database import Base + +# Tipo de trabajo (columna kind) del outbox de EXPEDIENTES. +KIND_EXPEDIENTE = "expediente" +KIND_COMPLETAR = "completar" + +# Tipos del outbox de ARCHIVOS (efc_file_outbox). +FILE_KIND_DOCUMENTO = "documento" + +# Tablas de origen posibles de un archivo. El CRM tiene DOS tablas de documentos con secuencias +# independientes, así que `source_id` por sí solo es ambiguo: crm.documents.id = 5 y +# ops.shipment_documents.id = 5 coexisten. +SOURCE_CRM_DOCUMENTS = "crm.documents" +SOURCE_OPS_SHIPMENT_DOCUMENTS = "ops.shipment_documents" +SOURCE_FIN_INVOICES = "fin.invoices" + +# Estados (columna status). +STATUS_PENDING = "pending" +STATUS_SENT = "sent" +STATUS_FAILED = "failed" + +# Tope de reintentos antes de marcar 'failed' (reconciliación / reintento manual). +# Heredado del carril de Anexo22. Con barridos de 120 s son ~17 minutos de insistencia antes de +# rendirse y dejar la fila visible para que una persona la reintente a mano. +MAX_ATTEMPTS = 8 + + +class EfcSyncOutbox(Base, TenantScopedMixin, TimestampMixin): + """Cola de metadatos hacia EFC: crear el expediente provisional y completarlo.""" + + __tablename__ = "efc_sync_outbox" + __table_args__ = ( + Index("ix_crm_efc_sync_outbox_status", "status"), + Index("ix_crm_efc_sync_outbox_kind_status", "kind", "status"), + {"schema": "crm"}, + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + kind: Mapped[str] = mapped_column(String(20), nullable=False) + + # Datos para construir el request a EFC (folio, storage_token, tenant slug, company, y la data + # aduanera si el kind es 'completar'). + payload: Mapped[dict] = mapped_column(JSON, nullable=False) + + # id local del expediente (crm.expedientes.id) que originó la fila. + expediente_ref: Mapped[Optional[int]] = mapped_column(Integer, nullable=True, index=True) + + # Ciclo de vida. + status: Mapped[str] = mapped_column(String(10), nullable=False, server_default=text(f"'{STATUS_PENDING}'")) + attempts: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) + last_error: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + sent_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) + + # Acuse de EFC al confirmar (trazabilidad). + efc_pedimento_id: Mapped[Optional[str]] = mapped_column(String(36), nullable=True) + + +class EfcFileOutbox(Base, TenantScopedMixin, TimestampMixin): + """Cola de ARCHIVOS hacia EFC. + + El binario vive en el MinIO del CRM (durable); esta fila referencia su ``s3_key`` y el expediente + destino. El worker lo sube a EFC y, con ``delete_local`` (corte directo), BORRA la copia local al + confirmar la entrega. + + ``delete_local`` **es el mecanismo de «EFC es la fuente única»**: "solo EFC" es el estado FINAL + (eventual), no el inmediato. Entre que el usuario sube el archivo y que EFC lo confirma, la copia + local es lo único que hay, y borrarla antes perdería el archivo si la entrega fallara. + + ``source_table`` es un añadido necesario sobre el original de Anexo22, que solo llevaba + ``source_id``. El CRM tiene dos tablas de documentos con secuencias independientes, así que un + entero solo es ambiguo entre ellas. Es el mismo problema que Anexo22 resolvió con su mapa por + ``kind``, y su comentario dice qué pasa si se ignora: un UPDATE con el id de otra tabla **vacía la + columna de un documento ajeno** que tuviera ese mismo entero — daño en el dato de otro, sin un + solo error visible. Un ``(kind, source_table)`` que no esté en el mapa **no toca nada**, en vez + de caer por omisión. + """ + + __tablename__ = "efc_file_outbox" + __table_args__ = ( + Index("ix_crm_efc_file_outbox_status", "status"), + Index("ix_crm_efc_file_outbox_kind_status", "kind", "status"), + Index("ix_crm_efc_file_outbox_source", "source_table", "source_id"), + {"schema": "crm"}, + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + kind: Mapped[str] = mapped_column(String(30), nullable=False) + + # Objeto en MinIO a subir + metadata para el upload a EFC. + s3_key: Mapped[str] = mapped_column(String(1024), nullable=False) + file_name: Mapped[str] = mapped_column(String(255), nullable=False) + content_type: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) + efc_tipo: Mapped[str] = mapped_column(String(40), nullable=False) # tipo de documento en EFC + + # Origen: la pareja (tabla, id) desambigua entre las dos secuencias de documentos del CRM. + source_table: Mapped[str] = mapped_column(String(30), nullable=False) + source_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + # El handle autoritativo que viaja a EFC y garantiza la idempotencia del lado de allá. + crm_document_ref: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + + expediente_ref: Mapped[int] = mapped_column( + Integer, ForeignKey("crm.expedientes.id"), nullable=False, index=True + ) + delete_local: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true")) + + # Ciclo de vida. + status: Mapped[str] = mapped_column(String(10), nullable=False, server_default=text(f"'{STATUS_PENDING}'")) + attempts: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("0")) + last_error: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + sent_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) + efc_document_id: Mapped[Optional[str]] = mapped_column(String(36), nullable=True) diff --git a/backend/api/v1/modules/crm/expediente_gateway/routes.py b/backend/api/v1/modules/crm/expediente_gateway/routes.py new file mode 100644 index 0000000..7ac4bd5 --- /dev/null +++ b/backend/api/v1/modules/crm/expediente_gateway/routes.py @@ -0,0 +1,58 @@ +"""Endpoints de operación y observabilidad del carril CRM -> EFC. + +Tablero mínimo para ver y reintentar la entrega de expedientes y documentos a EFC. Autenticado con +el auth normal del CRM y acotado por tenant/company, como el resto del módulo. +Montado bajo ``/v1/crm`` → ``/v1/crm/expediente-gateway/...`` +""" +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.orm import Session + +from core.database import get_core_db +from core.security import get_current_user + +from . import service + +router = APIRouter(prefix="/expediente-gateway", tags=["EFC Gateway (ops)"]) + + +@router.get("/outbox") +def list_outbox( + company_id: int = Query(..., description="Company ID"), + tipo: str | None = Query(None, description="Filtrar por tabla: sync|file"), + status: str | None = Query(None, description="Filtrar por status: pending|sent|failed"), + limit: int = Query(100, ge=1, le=500), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """Filas de los dos outbox, para ver los fallos y su ``last_error``.""" + return service.list_outbox(db, current_user["tenant_id"], company_id, tipo, status, limit) + + +@router.post("/outbox/{outbox_id}/retry") +def retry_outbox( + outbox_id: int, + company_id: int = Query(..., description="Company ID"), + tipo: str = Query("file", description="Tabla de la fila: sync|file"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """Reintento manual de una fila: la resetea a ``pending`` y la re-despacha. + + Una fila inexistente devuelve **404 con mensaje específico**, no un 200 silencioso: el frontend + pinta el botón de reintento según lo que reciba, y un 200 le haría creer que la entrega volvió a + la cola cuando no hay nada que entregar. + """ + ok = service.retry_outbox_row(db, outbox_id, current_user["tenant_id"], company_id, tipo) + if not ok: + raise HTTPException(status_code=404, detail="Fila de outbox no encontrada") + return {"status": "requeued", "id": outbox_id} + + +@router.get("/metrics") +def metrics( + company_id: int = Query(..., description="Company ID"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """Conteo de los dos outbox por status (pending/sent/failed) para monitoreo.""" + return service.outbox_metrics(db, current_user["tenant_id"], company_id) diff --git a/backend/api/v1/modules/crm/expediente_gateway/service.py b/backend/api/v1/modules/crm/expediente_gateway/service.py new file mode 100644 index 0000000..43a2029 --- /dev/null +++ b/backend/api/v1/modules/crm/expediente_gateway/service.py @@ -0,0 +1,731 @@ +"""Carril CRM Agentes de Carga -> EFC: encolado, entrega y reconciliación. + +Clon del gateway de Anexo22 (``anexo22/.../pedimentos/pedimento_gateway/service.py``), que es el +carril de referencia ya en producción. Quien conozca uno debe poder leer el otro, así que la tabla +de equivalencias va aquí: + +====================================== ====================================== +Anexo22 CRM +====================================== ====================================== +``replicate_pedimento_best_effort`` ``replicate_expediente_best_effort`` +``_enqueue_pedimento_outbox`` ``_enqueue_expediente_outbox`` +``_dispatch_delivery`` igual +``deliver_row`` / ``_deliver_pedimento`` ``deliver_row`` / ``_deliver_expediente`` +``_register_failure`` **idéntico** +``_ya_entregado(source_id, kind)`` ``_ya_entregado(source_table, source_id, kind)`` +``deliver_file_row`` **idéntico**, con ensure-then-upload y ``delete_local`` +``_register_file_failure`` **idéntico** +``_resolve_org_id`` + ``_org_id_cache`` igual — dict módulo-global, por worker, sin invalidación +``list_outbox`` / ``retry_outbox_row`` / ``outbox_metrics`` igual, para las dos tablas +``find_pedimento_gaps`` ``find_expediente_gaps`` +====================================== ====================================== + +**La máquina de reintentos tiene tres capas y las tres se conservan:** + +1. En el cliente HTTP: 3 intentos, backoff lineal ``0.15 * (attempt + 1)``, corte seco en 4xx. +2. En el worker: ``deliver_row`` **nunca lanza**; registra el fallo en la propia fila. +3. En el beat: barridos cada 120 s que re-despachan lo ``pending``. + +No hay ``autoretry_for``, ``retry_backoff`` ni ``max_retries`` en las tareas: duplicarían el +mecanismo que ya está en el cliente y en el barrido. + +**Cuatro guardas de idempotencia**, en este orden: +1. ``_ya_entregado(source_table, source_id, kind)`` antes de encolar. +2. ``if row.status == STATUS_SENT: return`` al entrar a entregar. +3. El ``crm_document_ref`` que viaja con la subida: EFC devuelve 200 con el que ya existía. +4. El UNIQUE parcial del lado de EFC — la única que garantiza la base. + +**Por qué ``find_expediente_gaps`` sigue aquí aunque el CRM sea mono-base.** En Anexo22 el outbox se +commitea aparte del pedimento (dos bases distintas) y ese doble-commit es lo que obligó a inventar el +barrido de huecos. Aquí la fila del outbox va en la MISMA transacción que el expediente, así que esa +ventana no existe. El barrido se conserva porque cubre otras dos cosas: los expedientes creados +**antes** de activar la integración, y cualquier crash. Queda escrito para que el siguiente que lo +lea no lo borre creyendo que es redundante. +""" +import logging +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Optional + +from sqlalchemy.orm import Session + +from core.config import settings +from core.database import scoped_core_db +from core.efc_client import EfcClient, EfcClientError, efc_client + +from ..expedientes.models import Expediente +from .models import ( + KIND_COMPLETAR, + KIND_EXPEDIENTE, + MAX_ATTEMPTS, + STATUS_FAILED, + STATUS_PENDING, + STATUS_SENT, + EfcFileOutbox, + EfcSyncOutbox, +) + +logger = logging.getLogger(__name__) + +# Cache de organización EFC por slug de tenant. Dict módulo-global: vive por worker y NO se +# invalida, igual que el del carril de Anexo22. Es correcto porque la organización de un tenant no +# cambia de id: el resolver de EFC es idempotente y devuelve siempre la misma. Si algún día pudiera +# cambiar, reiniciar el worker la vuelve a resolver. +_org_id_cache: dict[str, str] = {} + + +@contextmanager +def _savepoint(db: Session): + """Aísla un encolado dentro de la transacción del usuario con un SAVEPOINT. + + **Esto es lo único del encolado que NO se clona del carril de Anexo22, y la razón es de fondo.** + Allá el outbox vive en otra base que el pedimento, así que su ``except`` podía hacer + ``db.rollback()`` sin consecuencias: revertía la sesión del outbox y la del pedimento ni se + enteraba. + + Aquí el CRM es mono-base y el encolado corre DENTRO de la transacción del usuario. Un + ``db.rollback()`` en el ``except`` se llevaría por delante la solicitud y el expediente que el + usuario acaba de crear — exactamente lo contrario de best-effort, y sin un solo error visible + para él. Con el SAVEPOINT, un fallo del encolado deshace **solo** la fila del outbox y la + operación local sigue en pie para que el llamador la commitee. + """ + nested = db.begin_nested() + try: + yield nested + except Exception: + nested.rollback() + raise + + +# ══ Expediente: encolado y entrega ══════════════════════════════════════════ + +def replicate_expediente_best_effort(db: Session, expediente: Expediente) -> None: + """Encola la réplica del expediente a EFC y dispara la entrega inmediata. + + Best-effort en todo: si EFC no está configurado, o si el encolado o el despacho fallan, **no se + propaga el error**. El expediente local ya existe y la operación del usuario no se puede romper + porque un sistema de terceros no conteste. El barrido periódico recoge lo que quede pendiente. + """ + if not settings.EFC_API_URL: + return + row = _enqueue_expediente_outbox(db, expediente) + if row is None: + return + _dispatch_delivery(row.id, row.tenant_id, row.company_id) + + +def _enqueue_expediente_outbox(db: Session, expediente: Expediente) -> Optional[EfcSyncOutbox]: + """Inserta la fila de outbox del expediente. Devuelve ``None`` si falla, sin romper nada. + + A diferencia del original, **no commitea**: el CRM es mono-base, así que la fila viaja en la + misma transacción que el expediente. Eso cierra de raíz la ventana del doble-commit que en + Anexo22 obligó a inventar el barrido de huecos. + """ + try: + if _expediente_ya_encolado(db, expediente.id): + return None + # El slug del tenant NO se resuelve aquí: se rellena al ENTREGAR. Resolverlo ahora abriría + # una segunda sesión de base (``scoped_core_db``) dentro de la transacción del usuario, que + # es justo lo que el encolado debe evitar. Es además lo que hace el carril de referencia. + payload = { + "source": "crm", + "crm_company_id": expediente.company_id, + "crm_expediente_id": expediente.id, + "folio": expediente.folio, + "storage_token": expediente.efc_storage_token, + } + row = EfcSyncOutbox( + kind=KIND_EXPEDIENTE, + payload=payload, + expediente_ref=expediente.id, + status=STATUS_PENDING, + tenant_id=expediente.tenant_id, + company_id=expediente.company_id, + ) + with _savepoint(db): + db.add(row) + db.flush() + return row + except Exception: + logger.warning( + "expediente_gateway: no se pudo encolar el expediente id=%s en el outbox", + getattr(expediente, "id", None), exc_info=True, + ) + return None + + +def _expediente_ya_encolado(db: Session, expediente_id: int) -> bool: + """¿Ya hay una fila viva de alta para este expediente? Evita encolar la misma réplica dos veces.""" + return ( + db.query(EfcSyncOutbox.id) + .filter( + EfcSyncOutbox.expediente_ref == expediente_id, + EfcSyncOutbox.kind == KIND_EXPEDIENTE, + EfcSyncOutbox.status.in_((STATUS_PENDING, STATUS_SENT)), + ) + .first() + is not None + ) + + +def enqueue_completar_best_effort(db: Session, expediente: Expediente, campos: dict) -> None: + """Encola el completado del provisional en EFC con la data aduanera real.""" + if not settings.EFC_API_URL: + return + try: + row = EfcSyncOutbox( + kind=KIND_COMPLETAR, + payload={ + "source": "crm", + "crm_company_id": expediente.company_id, + "crm_expediente_id": expediente.id, + "folio": expediente.folio, + "pedimento": campos, + }, + expediente_ref=expediente.id, + status=STATUS_PENDING, + tenant_id=expediente.tenant_id, + company_id=expediente.company_id, + ) + with _savepoint(db): + db.add(row) + db.flush() + except Exception: + logger.warning( + "expediente_gateway: no se pudo encolar el completado del expediente id=%s", + getattr(expediente, "id", None), exc_info=True, + ) + return + _dispatch_delivery(row.id, row.tenant_id, row.company_id) + + +def _dispatch_delivery(outbox_id: int, tenant_id: int, company_id: int) -> None: + """Dispara la tarea de entrega propagando el contexto RLS por headers de Celery. + + Best-effort: si el broker no responde, el barrido la recoge. Los headers son obligatorios — + ``core/celery_app.py`` materializa el contexto de RLS a partir de ellos, y sin ellos la tarea + corre sin tenant y no ve nada. + """ + try: + from .tasks import deliver_outbox_row # import diferido: evita ciclo con celery_app + deliver_outbox_row.apply_async( + args=[outbox_id, tenant_id, company_id], + headers={"rls_tenant_id": str(tenant_id), "rls_company_id": str(company_id)}, + ) + except Exception: + logger.warning( + "expediente_gateway: no se pudo despachar la entrega outbox_id=%s (lo tomará el sweep)", + outbox_id, exc_info=True, + ) + + +def deliver_row(db: Session, row: EfcSyncOutbox, client: Optional[EfcClient] = None) -> None: + """Entrega una fila del outbox de expedientes a EFC. Actualiza estado y ``attempts``. + + **No lanza nunca**: los fallos se registran en la propia fila para reconciliación. Un fallo no + puede matar al worker ni perder la intención de entregar. + """ + client = client or efc_client + if not client.is_configured: + logger.info("expediente_gateway: EFC no configurado; se deja pendiente row=%s", row.id) + return + if row.status == STATUS_SENT: + return + try: + if row.kind == KIND_EXPEDIENTE: + _deliver_expediente(db, row, client) + elif row.kind == KIND_COMPLETAR: + _deliver_completar(db, row, client) + else: + row.status = STATUS_FAILED + row.last_error = f"kind desconocido: {row.kind}" + db.commit() + except EfcClientError as exc: + _register_failure(db, row, exc, retryable=exc.retryable) + except Exception as exc: # noqa: BLE001 — cualquier fallo se registra, no rompe el worker + _register_failure(db, row, exc, retryable=True) + + +def _register_failure(db: Session, row: EfcSyncOutbox, exc: Exception, retryable: bool) -> None: + row.attempts = (row.attempts or 0) + 1 + row.last_error = str(exc)[:2000] + if (not retryable) or row.attempts >= MAX_ATTEMPTS: + row.status = STATUS_FAILED + db.commit() + logger.warning( + "expediente_gateway: entrega falló row=%s attempts=%s retryable=%s status=%s: %s", + row.id, row.attempts, retryable, row.status, exc, + ) + + +def _deliver_expediente(db: Session, row: EfcSyncOutbox, client: EfcClient) -> None: + payload = dict(row.payload or {}) + org_id = _resolve_org_id(client, row.tenant_id) + payload["organizacion"] = {"efc_organizacion_id": org_id} + payload["crm_tenant_slug"] = _tenant_slug(row.tenant_id)[0] or "" + resp = client.ingest_expediente(payload) + efc = (resp or {}).get("efc") or {} + + row.status = STATUS_SENT + row.sent_at = datetime.now(timezone.utc) + row.efc_pedimento_id = efc.get("pedimento_id") + _stamp_expediente_link(db, row.expediente_ref, org_id, efc.get("pedimento_id")) + db.commit() + logger.info( + "expediente_gateway: expediente replicado row=%s efc_pedimento_id=%s", + row.id, row.efc_pedimento_id, + ) + + +def _deliver_completar(db: Session, row: EfcSyncOutbox, client: EfcClient) -> None: + payload = dict(row.payload or {}) + org_id = _resolve_org_id(client, row.tenant_id) + payload["organizacion"] = {"efc_organizacion_id": org_id} + payload["crm_tenant_slug"] = _tenant_slug(row.tenant_id)[0] or "" + folio = payload.get("folio") + client.completar_expediente(folio, payload) + row.status = STATUS_SENT + row.sent_at = datetime.now(timezone.utc) + db.commit() + logger.info("expediente_gateway: expediente completado en EFC row=%s folio=%s", row.id, folio) + + +def _stamp_expediente_link(db: Session, expediente_id: Optional[int], org_id: str, + pedimento_id: Optional[str]) -> None: + """Refleja en la fila del expediente que EFC ya lo tiene, para que la UI lo pinte. + + Es un espejo, no un handle: el CRM sigue hablando de este expediente por su ``folio``. Se guarda + porque el proxy de descarga necesita el ``organizacion_id`` para preguntarle a EFC. + """ + if expediente_id is None: + return + expediente = db.query(Expediente).filter(Expediente.id == expediente_id).first() + if expediente is None: + return + expediente.efc_organizacion_id = org_id + if pedimento_id: + expediente.efc_pedimento_id = pedimento_id + expediente.efc_link_state = "LINKED" + expediente.efc_error_code = None + expediente.efc_error_detail = None + + +# ══ Archivos: encolado y entrega ════════════════════════════════════════════ + +def _ya_entregado(db: Session, source_table: str, source_id: int, kind: str) -> bool: + """¿Este archivo ya se entregó al expediente? Evita re-encolar lo que ya está allá. + + Sin esta guarda, un reintento encolaba otra entrega del mismo archivo — que además **falla al + leer el objeto local, porque la primera entrega ya lo borró** con ``delete_local``. Ruido en el + log y una fila del outbox condenada a ``failed``. + + Lleva ``source_table`` además de ``source_id``, a diferencia del original: el CRM tiene dos + tablas de documentos con secuencias independientes, así que el id solo es ambiguo y esta guarda + se dispararía de más, saltándose la entrega de un documento distinto que casualmente comparte + entero. + """ + return ( + db.query(EfcFileOutbox.id) + .filter( + EfcFileOutbox.source_table == source_table, + EfcFileOutbox.source_id == source_id, + EfcFileOutbox.kind == kind, + EfcFileOutbox.status == STATUS_SENT, + ) + .first() + is not None + ) + + +def enqueue_file_best_effort( + db: Session, + *, + kind: str, + s3_key: str, + file_name: str, + content_type: Optional[str], + efc_tipo: str, + source_table: str, + source_id: int, + crm_document_ref: str, + expediente_ref: int, + tenant_id: int, + company_id: int, + delete_local: bool = True, +) -> Optional[EfcFileOutbox]: + """Encola un archivo hacia el expediente de EFC. Devuelve la fila, o ``None`` si no se encoló. + + **No commitea**: la fila va en la misma transacción que el documento que la origina, de modo que + no puede existir un documento sin su intención de entrega ni al revés. + """ + if not settings.EFC_API_URL: + return None + if _ya_entregado(db, source_table, source_id, kind): + return None + try: + row = EfcFileOutbox( + kind=kind, + s3_key=s3_key, + file_name=file_name, + content_type=content_type, + efc_tipo=efc_tipo, + source_table=source_table, + source_id=source_id, + crm_document_ref=crm_document_ref, + expediente_ref=expediente_ref, + delete_local=delete_local, + status=STATUS_PENDING, + tenant_id=tenant_id, + company_id=company_id, + ) + with _savepoint(db): + db.add(row) + db.flush() + return row + except Exception: + logger.warning( + "expediente_gateway: no se pudo encolar el archivo %s (%s:%s)", + s3_key, source_table, source_id, exc_info=True, + ) + return None + + +def _dispatch_file_delivery(outbox_id: int, tenant_id: int, company_id: int) -> None: + try: + from .tasks import deliver_file_outbox_row # import diferido + deliver_file_outbox_row.apply_async( + args=[outbox_id, tenant_id, company_id], + headers={"rls_tenant_id": str(tenant_id), "rls_company_id": str(company_id)}, + ) + except Exception: + logger.warning( + "expediente_gateway: no se pudo despachar entrega de archivo outbox_id=%s (lo tomará el sweep)", + outbox_id, exc_info=True, + ) + + +def deliver_file_row(db: Session, row: EfcFileOutbox, client: Optional[EfcClient] = None) -> None: + """Sube el archivo de ``row.s3_key`` al expediente de EFC y, si ``delete_local``, borra la copia. + + **Ensure-then-upload**: si EFC contesta 404 ``expediente_no_encontrado``, la creación del + provisional puede venir en camino (el outbox de expedientes y el de archivos son colas + distintas), así que se asegura el expediente y se reintenta el upload **una** vez. + + **No lanza nunca**: como ``deliver_row``, registra el fallo en la propia fila. + """ + client = client or efc_client + if not client.is_configured or row.status == STATUS_SENT: + return + try: + org_id = _resolve_org_id(client, row.tenant_id) + expediente = db.query(Expediente).filter(Expediente.id == row.expediente_ref).first() + if expediente is None: + raise EfcClientError( + f"expediente {row.expediente_ref} no encontrado para el archivo '{row.kind}'", + retryable=True, + ) + + from core.storage_s3 import get_object_bytes + content = get_object_bytes(row.s3_key) + ct = row.content_type or "application/octet-stream" + + try: + resp = client.upload_documento( + org_id, row.company_id, expediente.id, row.efc_tipo, + row.file_name, content, ct, crm_document_ref=row.crm_document_ref, + ) + except EfcClientError as exc: + if exc.status_code == 404 and exc.code == "expediente_no_encontrado": + # La creación del provisional puede venir en camino: se asegura y se reintenta UNA vez. + client.ingest_expediente({ + "source": "crm", + "crm_tenant_slug": (_tenant_slug(row.tenant_id)[0] or ""), + "crm_company_id": row.company_id, + "crm_expediente_id": expediente.id, + "folio": expediente.folio, + "storage_token": expediente.efc_storage_token, + "organizacion": {"efc_organizacion_id": org_id}, + }) + resp = client.upload_documento( + org_id, row.company_id, expediente.id, row.efc_tipo, + row.file_name, content, ct, crm_document_ref=row.crm_document_ref, + ) + else: + raise + + doc_id = resp.get("id") if isinstance(resp, dict) else None + + if row.delete_local: + try: + from core.storage_s3 import delete_object_if_exists + delete_object_if_exists(row.s3_key) + except Exception: + # Ya está en EFC: no poder borrar la copia local no invalida la entrega. + logger.warning( + "expediente_gateway: no se pudo borrar el archivo local %s (ya en EFC)", + row.s3_key, exc_info=True, + ) + + row.status = STATUS_SENT + row.sent_at = datetime.now(timezone.utc) + row.efc_document_id = doc_id + db.commit() + _marcar_documento_entregado(db, row, doc_id) + logger.info( + "expediente_gateway: archivo entregado row=%s kind=%s efc_document_id=%s", + row.id, row.kind, doc_id, + ) + except EfcClientError as exc: + _register_file_failure(db, row, exc, exc.retryable) + except Exception as exc: # noqa: BLE001 + _register_file_failure(db, row, exc, True) + + +def _register_file_failure(db: Session, row: EfcFileOutbox, exc: Exception, retryable: bool) -> None: + row.attempts = (row.attempts or 0) + 1 + row.last_error = str(exc)[:2000] + if (not retryable) or row.attempts >= MAX_ATTEMPTS: + row.status = STATUS_FAILED + db.commit() + _marcar_documento_fallido(db, row, exc) + logger.warning( + "expediente_gateway: entrega de archivo falló row=%s attempts=%s status=%s: %s", + row.id, row.attempts, row.status, exc, + ) + + +# El mapa (kind, source_table) -> modelo del documento de origen. Un par que NO esté aquí **no toca +# nada**, en vez de caer por omisión sobre una tabla cualquiera: escribir con el id de otra tabla +# vaciaría las columnas de un documento ajeno que tuviera ese mismo entero — daño en el dato de otro, +# sin un solo error visible. +def _modelo_de_origen(source_table: str): + if source_table == "crm.documents": + from ..documents.models import Document + return Document + if source_table == "ops.shipment_documents": + from api.v1.modules.ops.shipments.models import ShipmentDocument + return ShipmentDocument + return None + + +def _fila_de_origen(db: Session, row: EfcFileOutbox): + modelo = _modelo_de_origen(row.source_table) + if modelo is None or row.source_id is None: + return None + return ( + db.query(modelo) + .filter( + modelo.id == row.source_id, + modelo.tenant_id == row.tenant_id, + modelo.company_id == row.company_id, + ) + .first() + ) + + +def _marcar_documento_entregado(db: Session, row: EfcFileOutbox, doc_id) -> None: + """Cierra la entrega en la fila del documento: el badge de la UI pasa a «En expediente».""" + documento = _fila_de_origen(db, row) + if documento is None: + return + documento.efc_document_id = str(doc_id) if doc_id else None + documento.efc_sync_state = "SYNCED" + documento.efc_synced_at = datetime.now(timezone.utc) + documento.efc_error_code = None + documento.efc_error_detail = None + if row.delete_local: + # El objeto local ya no está: dejar la key apuntaría a algo inexistente y la descarga se + # ramificaría por el camino equivocado. + documento.file_key = None + db.commit() + + +def _marcar_documento_fallido(db: Session, row: EfcFileOutbox, exc: Exception) -> None: + """Refleja el fallo en la fila del documento para que la ficha lo muestre sin ir a los logs.""" + documento = _fila_de_origen(db, row) + if documento is None: + return + documento.efc_attempts = row.attempts + documento.efc_error_detail = str(exc)[:2000] + documento.efc_error_code = getattr(exc, "code", None) + if row.status == STATUS_FAILED: + documento.efc_sync_state = "FAILED" + db.commit() + + +# ══ Organización ════════════════════════════════════════════════════════════ + +def _resolve_org_id(client: EfcClient, tenant_id: int) -> str: + slug, name = _tenant_slug(tenant_id) + if not slug: + raise EfcClientError( + f"tenant {tenant_id} sin slug; no se puede resolver la organización EFC.", + retryable=False, + ) + if slug in _org_id_cache: + return _org_id_cache[slug] + resp = client.resolve_organizacion(slug, name) + org_id = resp.get("id") if isinstance(resp, dict) else None + if not org_id: + raise EfcClientError("El resolver de organización de EFC no devolvió id.", retryable=True) + _org_id_cache[slug] = org_id + return org_id + + +def _tenant_slug(tenant_id: int) -> tuple[Optional[str], Optional[str]]: + from api.v1.modules.core.tenants.models import Tenant + + with scoped_core_db(tenant_id=tenant_id) as db: + t = db.query(Tenant).filter(Tenant.id == tenant_id).first() + if t is None: + return None, None + return t.slug, t.name + + +# ══ Tablero de ops ══════════════════════════════════════════════════════════ + +def _outbox_to_dict(r: EfcSyncOutbox) -> dict: + return { + "id": r.id, + "tabla": "sync", + "kind": r.kind, + "status": r.status, + "attempts": r.attempts, + "last_error": r.last_error, + "expediente_ref": r.expediente_ref, + "efc_pedimento_id": r.efc_pedimento_id, + "created_at": r.created_at.isoformat() if r.created_at else None, + "sent_at": r.sent_at.isoformat() if r.sent_at else None, + } + + +def _file_outbox_to_dict(r: EfcFileOutbox) -> dict: + return { + "id": r.id, + "tabla": "file", + "kind": r.kind, + "status": r.status, + "attempts": r.attempts, + "last_error": r.last_error, + "expediente_ref": r.expediente_ref, + "file_name": r.file_name, + "efc_tipo": r.efc_tipo, + "source_table": r.source_table, + "source_id": r.source_id, + "crm_document_ref": r.crm_document_ref, + "efc_document_id": r.efc_document_id, + "created_at": r.created_at.isoformat() if r.created_at else None, + "sent_at": r.sent_at.isoformat() if r.sent_at else None, + } + + +def list_outbox(db: Session, tenant_id: int, company_id: int, tipo: Optional[str] = None, + status: Optional[str] = None, limit: int = 100) -> list[dict]: + """Lista filas de los DOS outbox para el tablero de ops. ``tipo`` ∈ ``sync`` | ``file``.""" + salida: list[dict] = [] + + if tipo in (None, "", "sync"): + q = db.query(EfcSyncOutbox).filter( + EfcSyncOutbox.tenant_id == tenant_id, EfcSyncOutbox.company_id == company_id + ) + if status: + q = q.filter(EfcSyncOutbox.status == status) + salida += [ + _outbox_to_dict(r) + for r in q.order_by(EfcSyncOutbox.created_at.desc()).limit(limit).all() + ] + + if tipo in (None, "", "file"): + q = db.query(EfcFileOutbox).filter( + EfcFileOutbox.tenant_id == tenant_id, EfcFileOutbox.company_id == company_id + ) + if status: + q = q.filter(EfcFileOutbox.status == status) + salida += [ + _file_outbox_to_dict(r) + for r in q.order_by(EfcFileOutbox.created_at.desc()).limit(limit).all() + ] + + salida.sort(key=lambda d: (d.get("created_at") or ""), reverse=True) + return salida[:limit] + + +def retry_outbox_row(db: Session, outbox_id: int, tenant_id: int, company_id: int, + tipo: str = "file") -> bool: + """Reintento manual: resetea la fila a ``pending`` (``attempts=0``) y la re-despacha. + + Devuelve ``False`` si no existe para ese tenant/company — el llamador lo traduce a **404 con + mensaje específico**, no a un 200 silencioso: es contrato con el frontend, que pinta el botón + según lo que reciba. + """ + modelo = EfcSyncOutbox if tipo == "sync" else EfcFileOutbox + r = ( + db.query(modelo) + .filter(modelo.id == outbox_id, modelo.tenant_id == tenant_id, modelo.company_id == company_id) + .first() + ) + if r is None: + return False + r.status = STATUS_PENDING + r.attempts = 0 + r.last_error = None + db.commit() + if tipo == "sync": + _dispatch_delivery(r.id, r.tenant_id, r.company_id) + else: + _reset_documento_pendiente(db, r) + _dispatch_file_delivery(r.id, r.tenant_id, r.company_id) + return True + + +def _reset_documento_pendiente(db: Session, row: EfcFileOutbox) -> None: + documento = _fila_de_origen(db, row) + if documento is None: + return + documento.efc_sync_state = "PENDING" + documento.efc_error_code = None + documento.efc_error_detail = None + db.commit() + + +def outbox_metrics(db: Session, tenant_id: int, company_id: int) -> dict: + """Conteo de los dos outbox por status (monitoreo). Los conteos suman las dos tablas.""" + from sqlalchemy import func + + counts = {STATUS_PENDING: 0, STATUS_SENT: 0, STATUS_FAILED: 0} + for modelo in (EfcSyncOutbox, EfcFileOutbox): + rows = ( + db.query(modelo.status, func.count()) + .filter(modelo.tenant_id == tenant_id, modelo.company_id == company_id) + .group_by(modelo.status) + .all() + ) + for estado, n in rows: + counts[estado] = counts.get(estado, 0) + n + return { + "pending": counts.get(STATUS_PENDING, 0), + "sent": counts.get(STATUS_SENT, 0), + "failed": counts.get(STATUS_FAILED, 0), + } + + +def find_expediente_gaps(db: Session, limit: int = 200) -> list: + """Expedientes (no borrados) SIN ninguna fila de outbox que los referencie. + + Nunca se encolaron: expedientes creados **antes** de activar la integración, o un crash. Se + re-encolan para no perder la réplica. + + Los ``failed`` **no son huecos** —existen como fila, son visibles y reintentables desde el + tablero—, así que la fila los excluye por estar presente, no por su estado. Corre sin contexto + de tenant (beat); cada expediente lleva el suyo. + """ + from sqlalchemy import exists + + ya_encolado = exists().where(EfcSyncOutbox.expediente_ref == Expediente.id) + return ( + db.query(Expediente) + .filter(Expediente.deleted_at.is_(None), ~ya_encolado) + .order_by(Expediente.id.desc()) + .limit(limit) + .all() + ) diff --git a/backend/api/v1/modules/crm/expediente_gateway/tasks.py b/backend/api/v1/modules/crm/expediente_gateway/tasks.py new file mode 100644 index 0000000..171370d --- /dev/null +++ b/backend/api/v1/modules/crm/expediente_gateway/tasks.py @@ -0,0 +1,122 @@ +"""Tareas Celery del carril CRM Agentes de Carga -> EFC. + +- ``deliver_outbox_row`` / ``sweep_outbox``: expedientes (alta del provisional y completado). +- ``deliver_file_outbox_row`` / ``sweep_file_outbox``: archivos. +- ``sweep_expediente_gaps``: reconciliación de expedientes que nunca se encolaron. + +**La trampa de RLS, que es lo que más fácil se pasa por alto.** ``core/celery_app.py`` materializa el +contexto desde los headers ``rls_tenant_id`` / ``rls_company_id``. Por tanto: + +- Las tareas **por fila** se despachan siempre con esos headers. +- Los **barridos corren sin contexto de tenant**: leen los ids pendientes con una sesión sin scope y + despachan una tarea hija por fila con sus propios headers. Si un barrido abriera una sesión con + scope e iterara, o no vería nada o se saltaría el aislamiento. + +**Sin ``autoretry_for``, ``retry_backoff`` ni ``max_retries``**: duplicarían el mecanismo de +reintento que ya está en el cliente (3 intentos con backoff lineal) y en el barrido (cada 120 s +hasta ``MAX_ATTEMPTS``). +""" +import logging + +from core.celery_app import celery_app +from core.config import settings +from core.database import scoped_core_db + +from . import service +from .models import STATUS_PENDING, EfcFileOutbox, EfcSyncOutbox + +logger = logging.getLogger(__name__) + + +# ── Expedientes ───────────────────────────────────────────────────────────── + +@celery_app.task(name="expediente_gateway.deliver_outbox_row") +def deliver_outbox_row(outbox_id: int, tenant_id: int, company_id: int) -> None: + with scoped_core_db(tenant_id, company_id) as db: + row = db.query(EfcSyncOutbox).filter(EfcSyncOutbox.id == outbox_id).first() + if row is None: + logger.warning( + "expediente_gateway: outbox_id=%s no encontrado (tenant=%s)", outbox_id, tenant_id + ) + return + service.deliver_row(db, row) + + +@celery_app.task(name="expediente_gateway.sweep_outbox") +def sweep_outbox(limit: int = 100) -> int: + """Re-despacha filas pendientes de expediente. Sin contexto de tenant: cada fila lleva el suyo.""" + with scoped_core_db() as db: + rows = ( + db.query(EfcSyncOutbox.id, EfcSyncOutbox.tenant_id, EfcSyncOutbox.company_id) + .filter(EfcSyncOutbox.status == STATUS_PENDING) + .order_by(EfcSyncOutbox.created_at.asc()) + .limit(limit) + .all() + ) + + for rid, tid, cid in rows: + deliver_outbox_row.apply_async( + args=[rid, tid, cid], + headers={"rls_tenant_id": str(tid), "rls_company_id": str(cid) if cid is not None else ""}, + ) + if rows: + logger.info("expediente_gateway: sweep (expedientes) re-despachó %s filas pendientes", len(rows)) + return len(rows) + + +# ── Archivos ──────────────────────────────────────────────────────────────── + +@celery_app.task(name="expediente_gateway.deliver_file_outbox_row") +def deliver_file_outbox_row(outbox_id: int, tenant_id: int, company_id: int) -> None: + with scoped_core_db(tenant_id, company_id) as db: + row = db.query(EfcFileOutbox).filter(EfcFileOutbox.id == outbox_id).first() + if row is None: + logger.warning( + "expediente_gateway: file outbox_id=%s no encontrado (tenant=%s)", outbox_id, tenant_id + ) + return + service.deliver_file_row(db, row) + + +@celery_app.task(name="expediente_gateway.sweep_file_outbox") +def sweep_file_outbox(limit: int = 100) -> int: + """Re-despacha archivos pendientes (EFC o el broker caídos cuando el usuario subió el archivo).""" + with scoped_core_db() as db: + rows = ( + db.query(EfcFileOutbox.id, EfcFileOutbox.tenant_id, EfcFileOutbox.company_id) + .filter(EfcFileOutbox.status == STATUS_PENDING) + .order_by(EfcFileOutbox.created_at.asc()) + .limit(limit) + .all() + ) + for rid, tid, cid in rows: + deliver_file_outbox_row.apply_async( + args=[rid, tid, cid], + headers={"rls_tenant_id": str(tid), "rls_company_id": str(cid) if cid is not None else ""}, + ) + if rows: + logger.info("expediente_gateway: sweep (archivos) re-despachó %s archivos pendientes", len(rows)) + return len(rows) + + +# ── Reconciliación de huecos ──────────────────────────────────────────────── + +@celery_app.task(name="expediente_gateway.sweep_expediente_gaps") +def sweep_expediente_gaps(limit: int = 200) -> int: + """Detecta expedientes que nunca se encolaron a EFC y los re-encola. + + No-op si la integración está apagada. + """ + if not settings.EFC_API_URL: + return 0 + n = 0 + with scoped_core_db() as db: + gaps = service.find_expediente_gaps(db, limit=limit) + for expediente in gaps: + service.replicate_expediente_best_effort(db, expediente) + n += 1 + if n: + db.commit() + if n: + logger.info("expediente_gateway: sweep de huecos re-encoló %s expedientes", n) + return n diff --git a/backend/api/v1/modules/crm/expedientes/service.py b/backend/api/v1/modules/crm/expedientes/service.py index d337e70..2b74aba 100644 --- a/backend/api/v1/modules/crm/expedientes/service.py +++ b/backend/api/v1/modules/crm/expedientes/service.py @@ -8,6 +8,7 @@ from datetime import datetime, timezone from fastapi import HTTPException, status from sqlalchemy.orm import Session +from ..expediente_gateway import service as gateway from ..service_requests.models import ServiceRequest from .dto import ExpedienteCompleteInput from .folio import next_folio, storage_token @@ -121,6 +122,9 @@ def ensure_expediente_for_service_request( ) db.add(expediente) db.flush() + # Réplica hacia EFC: best-effort y en la MISMA transacción. Si EFC está apagado + # (``EFC_API_URL`` vacía) esto es un no-op y el expediente vive igual, solo en el CRM. + gateway.replicate_expediente_best_effort(db, expediente) return expediente @@ -144,6 +148,19 @@ def ensure_expediente( return expediente +def _campos_para_efc(campos: dict) -> dict: + """Traduce los campos del expediente al vocabulario del contrato de EFC. + + Las fechas van en ISO porque el payload del outbox se serializa a JSON, y un ``date`` de Python + no es serializable: sin esto la fila se encolaría bien y **fallaría al entregar**, que es el peor + momento para descubrirlo. + """ + salida = {} + for clave, valor in campos.items(): + salida[clave] = valor.isoformat() if hasattr(valor, "isoformat") else valor + return salida + + def complete_expediente( db: Session, expediente_id: int, @@ -164,10 +181,16 @@ def complete_expediente( status_code=status.HTTP_409_CONFLICT, detail="El expediente ya está completado" ) - for field, value in payload.model_dump(exclude_unset=True).items(): + campos = payload.model_dump(exclude_unset=True) + for field, value in campos.items(): setattr(expediente, field, value) expediente.status = "completado" expediente.updated_by = user_id + + # El completado del provisional en EFC va por el outbox, no inline: puede devolver 409 si allá + # ya existe ese pedimento real, y eso no debe impedir que el CRM guarde lo que se capturó. + gateway.enqueue_completar_best_effort(db, expediente, _campos_para_efc(campos)) + db.commit() db.refresh(expediente) return expediente diff --git a/backend/api/v1/modules/crm/router.py b/backend/api/v1/modules/crm/router.py index bab1eb6..c3cb361 100644 --- a/backend/api/v1/modules/crm/router.py +++ b/backend/api/v1/modules/crm/router.py @@ -16,6 +16,7 @@ from .addresses.routes import router as addresses_router from .catalogs.routes import router as catalogs_router from .contacts.routes import router as contacts_router from .documents.routes import router as documents_router +from .expediente_gateway.routes import router as expediente_gateway_router from .expedientes.routes import router as expedientes_router from .leads.routes import router as leads_router from .metrics.routes import router as metrics_router @@ -37,6 +38,7 @@ router.include_router(contacts_router) router.include_router(addresses_router) router.include_router(documents_router) router.include_router(expedientes_router) +router.include_router(expediente_gateway_router) router.include_router(service_requests_router) router.include_router(quotes_router) router.include_router(leads_router) diff --git a/backend/core/celery_app.py b/backend/core/celery_app.py index 08b1c58..6a10b7b 100644 --- a/backend/core/celery_app.py +++ b/backend/core/celery_app.py @@ -96,6 +96,7 @@ def _reset_rls_context_from_task(task_id=None, task=None, **_): celery_app.conf.update( include=[ "api.v1.modules.core.help_center.tasks", + "api.v1.modules.crm.expediente_gateway.tasks", # Agrega aquí las tareas de tu proyecto: # "api.v1.modules.example.tasks", ] @@ -120,6 +121,21 @@ celery_app.conf.beat_schedule = { "task": "cleanup_orphan_layout_imports", "schedule": 3600.0, }, + # Carril CRM -> EFC. Los tres intervalos vienen del carril de referencia de Anexo22: 120 s para + # las dos colas y 300 s para la reconciliacion. El reintento NO es exponencial a proposito —el + # backoff corto vive en el cliente HTTP y el largo es este barrido de intervalo fijo. + "efc-sweep-outbox-every-2-min": { + "task": "expediente_gateway.sweep_outbox", + "schedule": 120.0, + }, + "efc-sweep-file-outbox-every-2-min": { + "task": "expediente_gateway.sweep_file_outbox", + "schedule": 120.0, + }, + "efc-sweep-expediente-gaps-every-5-min": { + "task": "expediente_gateway.sweep_expediente_gaps", + "schedule": 300.0, + }, } if __name__ == "__main__": diff --git a/backend/core/config.py b/backend/core/config.py index ff4be27..2cf6f12 100644 --- a/backend/core/config.py +++ b/backend/core/config.py @@ -62,7 +62,7 @@ class Settings(BaseSettings): # URL pública del frontend — usada en links de email (invitaciones, etc.) APP_PUBLIC_URL: str = "http://localhost:3000" - @field_validator("CENTRAL_SERVER_URL", "SPOKE_URLS", "HUB_URL", "HUB_API_BASE_URL", mode="before") + @field_validator("CENTRAL_SERVER_URL", "SPOKE_URLS", "HUB_URL", "HUB_API_BASE_URL", "EFC_API_URL", mode="before") @classmethod def strip_quotes(cls, v: str) -> str: if v and isinstance(v, str): @@ -103,6 +103,25 @@ class Settings(BaseSettings): S3_FILE_STORAGE: bool = True S3_PRESIGNED_EXPIRES_SECONDS: int = 3600 + # ── EFC (expediente electrónico) ──────────────────────────────────────────────────────── + # Carril CRM -> EFC: los documentos del CRM se resguardan en el expediente de EFC. + # Los nombres son los MISMOS que usa el gateway de Anexo22 contra el mismo EFC: inventar + # otros obligaría a quien opera los dos sistemas a recordar dos juegos de variables para + # exactamente lo mismo. + # EFC_API_URL vacía = integración APAGADA. Todo el enganche es best-effort y hace no-op: + # el CRM sigue funcionando igual, guardando los archivos solo en su MinIO. + EFC_API_URL: str = "" + EFC_API_KEY: str = "" # == CRM_INTEGRATION_API_KEY del lado de EFC + EFC_API_VERIFY_SSL: bool = True + EFC_API_TIMEOUT_MS: int = 8000 # metadatos: resolver, ingest, completar + # Las subidas van aparte: 8 s no alcanzan para un archivo de 25 MB. Debe quedar POR DEBAJO + # del proxy_read_timeout del nginx de EFC (ver core/efc_client.py). + EFC_UPLOAD_TIMEOUT_MS: int = 55000 + # Scaffolding de mTLS: activarlo es configuración, no código. + EFC_MTLS_CA_PATH: str = "" + EFC_MTLS_CERT_PATH: str = "" + EFC_MTLS_KEY_PATH: str = "" + model_config = SettingsConfigDict( env_file=[".env", "../.env"], case_sensitive=True, diff --git a/backend/core/efc_client.py b/backend/core/efc_client.py new file mode 100644 index 0000000..4350360 --- /dev/null +++ b/backend/core/efc_client.py @@ -0,0 +1,301 @@ +"""Cliente HTTP hacia EFC (carril de integración CRM Agentes de Carga -> EFC). + +Clon del cliente del gateway de Anexo22, que es el carril de referencia ya en producción +(``anexo22/backend/api/v1/modules/pedimentos/pedimento_gateway/client.py``), con las rutas +cambiadas a ``.../integrations/crm/...``. No es una reinterpretación: el molde de reintentos, el +corte en 4xx y la forma del error se conservan tal cual. + +**Síncrono a propósito** (``httpx.Client``): el consumidor es el worker de Celery que drena el +outbox, que corre en contexto sync. El único punto async del carril es el proxy de descarga de cara +al usuario, y ése no usa este cliente. + +Todos los endpoints se autentican con el header ``X-Api-Key`` +(``settings.EFC_API_KEY`` == ``CRM_INTEGRATION_API_KEY`` del lado de EFC). +""" +import logging +import time +from typing import Any, Optional + +import httpx + +from core.config import settings + +logger = logging.getLogger(__name__) + +# Rutas de EFC (prefijo /api/v1/, ver config/urls.py del backend de EFC). +_PATH_ORG_BUSCAR = "/api/v1/organization/integrations/crm/organizaciones/" +_PATH_ORG_RESOLVER = "/api/v1/organization/integrations/crm/organizaciones/resolver/" +_PATH_EXPEDIENTE = "/api/v1/customs/integrations/crm/expedientes/" +_PATH_EXPEDIENTE_COMPLETAR = "/api/v1/customs/integrations/crm/expedientes/{folio}/completar/" +_PATH_EXPEDIENTE_DETALLE = "/api/v1/customs/integrations/crm/expedientes/{folio}/" +_PATH_DOCS = "/api/v1/record/integrations/crm/documentos/" +_PATH_DOCS_LIST = "/api/v1/record/integrations/crm/documentos/list/" +_PATH_DOC_DESCARGAR = "/api/v1/record/integrations/crm/documentos/{doc_id}/descargar/" +_PATH_DOC_ELIMINAR = "/api/v1/record/integrations/crm/documentos/{doc_id}/eliminar/" +_PATH_DOC_REEMPLAZAR = "/api/v1/record/integrations/crm/documentos/{doc_id}/reemplazar/" + + +class EfcClientError(Exception): + """Error de comunicación con EFC. + + ``retryable=True`` marca fallos transitorios (5xx/timeout/red) que el worker debe reintentar. + ``status_code``/``code`` exponen la respuesta de EFC para que el worker pueda ramificar + (p. ej. 404 ``expediente_no_encontrado`` → ensure-then-upload). + + El worker decide **por el campo ``retryable``**, nunca parseando el mensaje: un texto de error + cambia con cualquier refactor del otro repo y con él se caería la política de reintentos sin que + nada se vea roto. + """ + + def __init__(self, message: str, status_code: Optional[int] = None, + code: Optional[str] = None, retryable: bool = False): + super().__init__(message) + self.status_code = status_code + self.code = code + self.retryable = retryable + + +class EfcClient: + def __init__( + self, + base_url: Optional[str] = None, + api_key: Optional[str] = None, + timeout_ms: Optional[int] = None, + upload_timeout_ms: Optional[int] = None, + verify_ssl: Optional[bool] = None, + retries: int = 2, + transport: Optional[httpx.BaseTransport] = None, + ): + self.base_url = (base_url if base_url is not None else settings.EFC_API_URL).rstrip("/") + self.api_key = api_key if api_key is not None else settings.EFC_API_KEY + self.timeout_s = max(0.1, float(timeout_ms or settings.EFC_API_TIMEOUT_MS) / 1000.0) + # Timeout aparte para las subidas: los 8 s de los metadatos no alcanzan para un archivo de + # 25 MB. Tiene que quedar POR DEBAJO del proxy_read_timeout del nginx de EFC — si el CRM + # esperara más, vería un 504 opaco sin saber si el documento entró. Fallando primero de este + # lado, el reintento con el mismo efc_document_ref es limpio. + self.upload_timeout_s = max( + 0.1, float(upload_timeout_ms or settings.EFC_UPLOAD_TIMEOUT_MS) / 1000.0 + ) + self.verify_ssl = settings.EFC_API_VERIFY_SSL if verify_ssl is None else verify_ssl + self.retries = max(0, int(retries)) + self.transport = transport + # mTLS (scaffolding pre-prod): si hay CA se usa para verificar; si hay par cert/key se + # presenta como certificado de cliente. Vacío = TLS normal. + self._ca_path = settings.EFC_MTLS_CA_PATH or "" + self._cert_path = settings.EFC_MTLS_CERT_PATH or "" + self._key_path = settings.EFC_MTLS_KEY_PATH or "" + + def _client_kwargs(self, timeout_s: Optional[float] = None) -> dict: + """``verify``/``cert`` para httpx según config mTLS (o TLS normal si no hay mTLS).""" + verify = self._ca_path if self._ca_path else self.verify_ssl + kwargs = { + "timeout": timeout_s or self.timeout_s, + "verify": verify, + "transport": self.transport, + } + if self._cert_path and self._key_path: + kwargs["cert"] = (self._cert_path, self._key_path) + return kwargs + + @property + def is_configured(self) -> bool: + """``False`` = integración deshabilitada (best-effort): sin URL o sin key.""" + return bool(self.base_url and self.api_key) + + # ── HTTP interno ────────────────────────────────────────────────────────── + + def _request(self, method: str, path: str, *, json: Any = None, + params: dict = None, files: dict = None, data: dict = None, + stream: bool = False, timeout_s: Optional[float] = None): + if not self.is_configured: + raise EfcClientError("EFC no configurado (EFC_API_URL/EFC_API_KEY vacíos).", retryable=False) + + url = f"{self.base_url}{path}" + headers = {"X-Api-Key": self.api_key} + last_error: Optional[Exception] = None + + for attempt in range(self.retries + 1): + try: + client = httpx.Client(**self._client_kwargs(timeout_s)) + try: + response = client.request(method, url, headers=headers, json=json, + params=params, files=files, data=data) + except Exception: + client.close() + raise + + if 200 <= response.status_code < 300: + if stream: + # El caller lee response.content y cierra el cliente. + return response, client + client.close() + return response + + # 5xx: transitorio, reintentar. + if response.status_code >= 500 and attempt < self.retries: + client.close() + time.sleep(0.15 * (attempt + 1)) + continue + + # 4xx u otro: no reintentar. Extraer code/mensaje de EFC. + code, message = _parse_error_body(response) + client.close() + raise EfcClientError( + message or f"EFC respondió {response.status_code}", + status_code=response.status_code, + code=code, + retryable=response.status_code >= 500, + ) + + except (httpx.TimeoutException, httpx.NetworkError) as exc: + last_error = exc + if attempt >= self.retries: + break + time.sleep(0.15 * (attempt + 1)) + except EfcClientError: + raise + except Exception as exc: # noqa: BLE001 — cualquier fallo inesperado es no-retryable + raise EfcClientError(str(exc), retryable=False) from exc + + raise EfcClientError( + f"EFC inaccesible tras {self.retries + 1} intentos: {last_error}", + retryable=True, + ) + + # ── Organización ────────────────────────────────────────────────────────── + + def buscar_organizaciones(self, q: str) -> list: + """Búsqueda por texto, para el alta manual desde una pantalla de administración.""" + return self._request("GET", _PATH_ORG_BUSCAR, params={"q": q}).json() + + def resolve_organizacion(self, tenant_slug: str, tenant_name: Optional[str] = None) -> dict: + payload = {"tenant_slug": tenant_slug} + if tenant_name: + payload["tenant_name"] = tenant_name + return self._request("POST", _PATH_ORG_RESOLVER, json=payload).json() + + # ── Expediente (pedimento provisional en EFC) ────────────────────────────── + + def ingest_expediente(self, payload: dict) -> dict: + """Crea el pedimento provisional del expediente. 201 si es nuevo, 200 si ya existía.""" + return self._request("POST", _PATH_EXPEDIENTE, json=payload).json() + + def completar_expediente(self, folio: str, payload: dict) -> dict: + """Completa un provisional con la data aduanera real. Ningún archivo se mueve.""" + path = _PATH_EXPEDIENTE_COMPLETAR.format(folio=folio) + return self._request("POST", path, json=payload).json() + + def get_expediente(self, folio: str, organizacion_id: str) -> dict: + path = _PATH_EXPEDIENTE_DETALLE.format(folio=folio) + return self._request("GET", path, params={"organizacion_id": str(organizacion_id)}).json() + + # ── Documentos ──────────────────────────────────────────────────────────── + + def upload_documento(self, organizacion_id: str, crm_company_id: int, crm_expediente_id: int, + tipo: str, filename: str, content: bytes, + content_type: str = "application/octet-stream", + crm_document_ref: Optional[str] = None) -> dict: + """Sube un documento al expediente. Multipart, nunca base64. + + ``crm_document_ref`` es **el handle de NUESTRO registro de origen**, y es lo que después + permite recuperar el archivo sin guardar de este lado ningún identificador de EFC. EFC lo + guarda junto al documento con un UNIQUE parcial por ``(organizacion, ref)``, así que una + entrega repetida devuelve el documento que ya existía (200) en vez de crear otro (201). + + Es la cuarta capa de idempotencia del carril y la única que garantiza la base: la entrega la + hace un worker con reintentos, así que un timeout ambiguo —EFC commiteó y contestó tarde— + duplicaría el documento sin esto. + """ + files = {"file": (filename, content, content_type)} + data = { + "organizacion_id": str(organizacion_id), + "crm_company_id": str(int(crm_company_id)), + "crm_expediente_id": str(int(crm_expediente_id)), + "tipo": tipo, + } + if crm_document_ref: + data["crm_document_ref"] = crm_document_ref + return self._request( + "POST", _PATH_DOCS, files=files, data=data, timeout_s=self.upload_timeout_s + ).json() + + def list_documentos(self, organizacion_id: str, crm_expediente_id: int, *, + tipo: Optional[str] = None, + crm_document_ref: Optional[str] = None) -> list: + """Documentos del expediente. ``tipo`` y ``crm_document_ref`` son filtros OPCIONALES. + + Preguntar por ``crm_document_ref`` es lo que permite recuperar ``efc_document_id`` si el + cache local se perdió, sin guardar identificadores ajenos como handle. + """ + params = { + "organizacion_id": str(organizacion_id), + "crm_expediente_id": str(int(crm_expediente_id)), + } + if tipo: + params["tipo"] = tipo + if crm_document_ref: + params["crm_document_ref"] = crm_document_ref + return self._request("GET", _PATH_DOCS_LIST, params=params).json() + + def replace_documento(self, organizacion_id: str, doc_id: str, filename: str, content: bytes, + content_type: str = "application/octet-stream") -> dict: + """Sustituye el CONTENIDO de un documento conservando su fila (mismo id, mismo tipo). + + EFC sube primero y borra el viejo al final, así que una subida fallida deja el anterior + intacto y descargable. + """ + files = {"file": (filename, content, content_type)} + data = {"organizacion_id": str(organizacion_id)} + path = _PATH_DOC_REEMPLAZAR.format(doc_id=doc_id) + return self._request( + "PUT", path, files=files, data=data, timeout_s=self.upload_timeout_s + ).json() + + def download_documento(self, organizacion_id: str, doc_id: str) -> tuple[bytes, str]: + path = _PATH_DOC_DESCARGAR.format(doc_id=doc_id) + params = {"organizacion_id": str(organizacion_id)} + response, client = self._request("GET", path, params=params, stream=True) + try: + content = response.content + filename = _filename_from_response(response, default=str(doc_id)) + finally: + client.close() + return content, filename + + def download_url(self, doc_id: str) -> str: + """URL absoluta del endpoint de descarga de EFC, para el proxy async de la fase 7. + + El proxy no puede usar ``download_documento``: éste es síncrono y bufferiza el archivo + entero. Lo que necesita es la URL y el header, y hace su propio streaming. + """ + return f"{self.base_url}{_PATH_DOC_DESCARGAR.format(doc_id=doc_id)}" + + @property + def auth_headers(self) -> dict: + """El header de autenticación, para el proxy async que no pasa por ``_request``.""" + return {"X-Api-Key": self.api_key} + + +def _parse_error_body(response) -> tuple[Optional[str], Optional[str]]: + """Extrae ``(code, message)`` del cuerpo de error estructurado de EFC + (``{"error": {"code", "message"}}``) sin reventar si no es JSON.""" + try: + body = response.json() + except Exception: + return None, None + err = body.get("error") if isinstance(body, dict) else None + if isinstance(err, dict): + return err.get("code"), err.get("message") + return None, None + + +def _filename_from_response(response, default: str) -> str: + disp = response.headers.get("Content-Disposition", "") + if "filename=" in disp: + return disp.split("filename=")[-1].strip().strip('"') or default + return default + + +# Instancia por defecto (lee settings). Los tests inyectan su propio transport construyendo +# EfcClient(transport=httpx.MockTransport(...)). +efc_client = EfcClient() diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index df57785..bf92efa 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -30,6 +30,7 @@ import api.v1.modules.crm.activities.models # noqa: E402,F401 import api.v1.modules.crm.addresses.models # noqa: E402,F401 import api.v1.modules.crm.contacts.models # noqa: E402,F401 import api.v1.modules.crm.documents.models # noqa: E402,F401 +import api.v1.modules.crm.expediente_gateway.models # noqa: E402,F401 import api.v1.modules.crm.expedientes.models # noqa: E402,F401 import api.v1.modules.crm.leads.models # noqa: E402,F401 import api.v1.modules.crm.opportunities.models # noqa: E402,F401 diff --git a/backend/tests/test_doc_types_paridad.py b/backend/tests/test_doc_types_paridad.py new file mode 100644 index 0000000..0e798e2 --- /dev/null +++ b/backend/tests/test_doc_types_paridad.py @@ -0,0 +1,85 @@ +"""Paridad del catálogo de tipos de documento entre el CRM y EFC. + +``EFC_DOC_TYPES`` del CRM tiene que ser **exactamente** el juego de claves de +``TIPOS_DOCUMENTO_CRM`` de ``api/record/views_integrations_crm.py`` en EFC. La lista está duplicada +a mano en dos repos con despliegue independiente, y este archivo es lo único que la mantiene +honesta: si alguien agrega un tipo de un solo lado, esto se pone rojo **antes** de que un documento +se rechace en producción con ``tipo_invalido``. + +El juego esperado va escrito **literal** aquí y no derivado de ``doc_types.py``, porque una prueba +que se lo pregunte al mismo módulo que valida no prueba nada: pasaría con cualquier cambio. +""" + +from api.v1.modules.crm.expedientes.doc_types import EFC_DOC_TYPES, is_valid_doc_type + +# Copia literal de las claves de TIPOS_DOCUMENTO_CRM (EFC, fase 3 del ticket T2026-08-046). +# Al cambiar EFC, se cambia AQUÍ y el rojo obliga a mirar los dos lados. +CLAVES_EN_EFC = { + # crm.documents — DOC_TYPES de frontend/src/lib/api/crm/format.ts + "constancia_fiscal", + "acta_constitutiva", + "identificacion", + "comprobante_domicilio", + "contrato", + "presentacion", + "certificacion", + "licencia", + "convenio", + "tarifario", + # ops.shipment_documents — SHIPMENT_DOC_TYPES del mismo archivo + "MBL", + "HBL", + "MAWB", + "HAWB", + "CMR", + "factura_comercial", + "packing_list", + "carta_encomienda", + "carta_garantia", + "certificado_permiso", + # fin.invoices + "factura_venta", + # 'otro' existe en AMBAS listas del CRM y significa lo mismo: una sola entrada + "otro", +} + + +def test_el_catalogo_del_crm_es_identico_al_de_efc(): + faltan_en_crm = CLAVES_EN_EFC - EFC_DOC_TYPES + sobran_en_crm = EFC_DOC_TYPES - CLAVES_EN_EFC + assert not faltan_en_crm, f"EFC acepta tipos que el CRM no conoce: {sorted(faltan_en_crm)}" + assert not sobran_en_crm, ( + f"El CRM mandaría tipos que EFC va a rechazar con tipo_invalido: {sorted(sobran_en_crm)}" + ) + + +def test_son_exactamente_veintidos(): + """El número está en el ticket. Si cambia, es un cambio de contrato entre dos repos.""" + assert len(EFC_DOC_TYPES) == 22 + + +def test_no_hay_duplicados_entre_las_tres_fuentes(): + """``otro`` está en las dos listas del CRM y debe colapsar a UNA entrada. + + Un ``frozenset`` lo colapsa solo; la prueba está para que una futura refactorización a lista o + a tupla no reintroduzca el duplicado en silencio. + """ + from api.v1.modules.crm.expedientes import doc_types + + todas = ( + doc_types._TIPOS_DOCUMENTOS_CLIENTE + + doc_types._TIPOS_DOCUMENTOS_EMBARQUE + + doc_types._TIPOS_FACTURACION + + doc_types._TIPOS_COMUNES + ) + assert len(todas) == len(set(todas)) + + +def test_un_tipo_fuera_del_catalogo_se_rechaza(): + """El CRM valida ANTES de gastar un viaje de red, y evita que un typo cree un DocumentType + basura en el catálogo GLOBAL de EFC, que comparten todas las organizaciones.""" + assert is_valid_doc_type("MBL") is True + assert is_valid_doc_type("mbl") is False # sensible a mayúsculas, como el catálogo de EFC + assert is_valid_doc_type("factura_de_venta") is False # typo de 'factura_venta' + assert is_valid_doc_type("") is False + assert is_valid_doc_type(None) is False diff --git a/backend/tests/test_efc_client.py b/backend/tests/test_efc_client.py new file mode 100644 index 0000000..e159b56 --- /dev/null +++ b/backend/tests/test_efc_client.py @@ -0,0 +1,248 @@ +"""Pruebas del cliente HTTP hacia EFC. + +**Existen porque el carril de referencia no las tiene.** Verificado: en el gateway de Anexo22 no hay +ni una prueba de ``EfcClient._request``, así que su bucle de reintentos, su backoff, su corte en 4xx +y su header nunca se ejercitan. Ese hueco no se clona. + +Todo va contra ``httpx.MockTransport`` por el parámetro ``transport``, que existe justamente para +esto: **ninguna de estas pruebas toca la red**. +""" + +import httpx +import pytest + +from core.efc_client import EfcClient, EfcClientError + +BASE = "https://efc.example.test" +KEY = "llave-de-prueba" + + +def _client(handler, **kwargs) -> EfcClient: + return EfcClient( + base_url=kwargs.pop("base_url", BASE), + api_key=kwargs.pop("api_key", KEY), + timeout_ms=kwargs.pop("timeout_ms", 500), + upload_timeout_ms=kwargs.pop("upload_timeout_ms", 500), + verify_ssl=False, + transport=httpx.MockTransport(handler), + **kwargs, + ) + + +def test_reintenta_un_500_y_devuelve_el_exito(): + intentos = {"n": 0} + + def handler(request): + intentos["n"] += 1 + if intentos["n"] == 1: + return httpx.Response(500, json={"detail": "boom"}) + return httpx.Response(200, json={"id": "org-1"}) + + resp = _client(handler).resolve_organizacion("temex") + assert resp == {"id": "org-1"} + assert intentos["n"] == 2 + + +def test_un_500_permanente_hace_exactamente_tres_intentos_y_es_retryable(): + """``retries = 2`` significa 3 intentos: el original + 2. Ni 2 ni 4.""" + intentos = {"n": 0} + + def handler(request): + intentos["n"] += 1 + return httpx.Response(500, json={"detail": "boom"}) + + with pytest.raises(EfcClientError) as exc: + _client(handler).resolve_organizacion("temex") + + assert intentos["n"] == 3 + assert exc.value.retryable is True + + +def test_un_timeout_permanente_hace_tres_intentos_y_es_retryable(): + intentos = {"n": 0} + + def handler(request): + intentos["n"] += 1 + raise httpx.ConnectTimeout("se acabó el tiempo", request=request) + + with pytest.raises(EfcClientError) as exc: + _client(handler).resolve_organizacion("temex") + + assert intentos["n"] == 3 + assert exc.value.retryable is True + + +def test_un_400_no_se_reintenta_y_extrae_el_code_del_cuerpo(): + """El corte en 4xx es lo que evita machacar a EFC con una petición que nunca va a pasar. + + Y el ``code`` extraído es lo que permite al worker ramificar **por campo**, nunca parseando el + texto del mensaje: un texto cambia con cualquier refactor del otro repo. + """ + intentos = {"n": 0} + + def handler(request): + intentos["n"] += 1 + return httpx.Response( + 400, + json={"error": {"code": "espacio_insuficiente", "message": "La licencia no tiene espacio"}}, + ) + + with pytest.raises(EfcClientError) as exc: + _client(handler).resolve_organizacion("temex") + + assert intentos["n"] == 1 + assert exc.value.status_code == 400 + assert exc.value.code == "espacio_insuficiente" + assert exc.value.retryable is False + assert "La licencia no tiene espacio" in str(exc.value) + + +@pytest.mark.parametrize("status_code", [401, 403]) +def test_401_y_403_no_se_reintentan(status_code): + """Una key mal configurada no mejora insistiendo: reintentarla solo gasta cuota y llena logs.""" + intentos = {"n": 0} + + def handler(request): + intentos["n"] += 1 + return httpx.Response(status_code) + + with pytest.raises(EfcClientError) as exc: + _client(handler).resolve_organizacion("temex") + + assert intentos["n"] == 1 + assert exc.value.retryable is False + + +def test_un_cuerpo_de_error_que_no_es_json_no_revienta(): + def handler(request): + return httpx.Response(400, text="502 Bad Gateway") + + with pytest.raises(EfcClientError) as exc: + _client(handler).resolve_organizacion("temex") + + assert exc.value.code is None + assert exc.value.status_code == 400 + + +def test_toda_llamada_manda_el_header_x_api_key(): + visto = {} + + def handler(request): + visto["key"] = request.headers.get("X-Api-Key") + return httpx.Response(200, json={"id": "org-1"}) + + _client(handler).resolve_organizacion("temex") + assert visto["key"] == KEY + + +def test_sin_url_configurada_no_toca_la_red_y_el_error_no_es_retryable(): + """``is_configured is False`` es lo que hace que todo el carril sea best-effort. + + Si esto tocara la red, cada operación del CRM con EFC apagado pagaría un timeout. + """ + llamado = {"n": 0} + + def handler(request): + llamado["n"] += 1 + return httpx.Response(200, json={}) + + client = _client(handler, base_url="") + assert client.is_configured is False + + with pytest.raises(EfcClientError) as exc: + client.resolve_organizacion("temex") + + assert llamado["n"] == 0 + assert exc.value.retryable is False + + +def test_sin_api_key_tampoco_esta_configurado(): + def handler(request): + return httpx.Response(200, json={}) + + assert _client(handler, api_key="").is_configured is False + + +def test_la_base_url_con_y_sin_barra_final_dan_la_misma_url(): + urls = [] + + def handler(request): + urls.append(str(request.url)) + return httpx.Response(200, json={"id": "org-1"}) + + _client(handler, base_url=BASE).resolve_organizacion("temex") + _client(handler, base_url=BASE + "/").resolve_organizacion("temex") + + assert urls[0] == urls[1] + assert "//organization" not in urls[0] + + +def test_la_subida_va_multipart_y_lleva_el_crm_document_ref(): + """El ref es la tercera capa de idempotencia: EFC devuelve 200 con el que ya existía.""" + visto = {} + + def handler(request): + visto["content_type"] = request.headers.get("Content-Type", "") + visto["body"] = request.content + return httpx.Response(201, json={"id": "doc-1"}) + + resp = _client(handler).upload_documento( + "org-1", 1, 42, "MBL", "guia.pdf", b"%PDF-1.4 contenido", "application/pdf", + crm_document_ref="SHPDOC-1-4471", + ) + + assert resp == {"id": "doc-1"} + assert visto["content_type"].startswith("multipart/form-data") + assert b"SHPDOC-1-4471" in visto["body"] + assert b"%PDF-1.4 contenido" in visto["body"] + # Nada de base64: el archivo viaja crudo dentro del multipart. + assert b"base64" not in visto["body"] + + +def test_la_subida_usa_el_timeout_largo_y_los_metadatos_el_corto(): + """Los 8 s de los metadatos no alcanzan para un archivo de 25 MB, y un timeout de subida + demasiado largo haría que el CRM vea un 504 opaco de nginx sin saber si el documento entró.""" + client = _client(handler=lambda r: httpx.Response(200, json={}), timeout_ms=8000, upload_timeout_ms=55000) + assert client.timeout_s == 8.0 + assert client.upload_timeout_s == 55.0 + assert client.upload_timeout_s > client.timeout_s + + +def test_ensure_then_upload_puede_ramificar_por_el_code_del_404(): + """El 404 del expediente tiene que llegar al worker con su ``code`` y su ``status_code``. + + Es lo que dispara el ensure-then-upload; sin el code, el worker tendría que adivinar de qué es + el 404 y crearía provisionales por cualquier ausencia. + """ + def handler(request): + return httpx.Response( + 404, json={"error": {"code": "expediente_no_encontrado", "message": "no está"}} + ) + + with pytest.raises(EfcClientError) as exc: + _client(handler).upload_documento("org-1", 1, 42, "MBL", "g.pdf", b"x") + + assert exc.value.status_code == 404 + assert exc.value.code == "expediente_no_encontrado" + assert exc.value.retryable is False + + +def test_la_descarga_devuelve_contenido_y_nombre_del_content_disposition(): + def handler(request): + return httpx.Response( + 200, + content=b"contenido binario", + headers={"Content-Disposition": 'attachment; filename="factura.pdf"'}, + ) + + contenido, nombre = _client(handler).download_documento("org-1", "doc-1") + assert contenido == b"contenido binario" + assert nombre == "factura.pdf" + + +def test_la_descarga_sin_content_disposition_cae_al_id_del_documento(): + def handler(request): + return httpx.Response(200, content=b"x") + + _contenido, nombre = _client(handler).download_documento("org-1", "doc-9") + assert nombre == "doc-9" diff --git a/backend/tests/test_efc_entrega_documento.py b/backend/tests/test_efc_entrega_documento.py new file mode 100644 index 0000000..51ba9fa --- /dev/null +++ b/backend/tests/test_efc_entrega_documento.py @@ -0,0 +1,350 @@ +"""Pruebas de la entrega de un archivo al expediente de EFC. + +Cubren las tres cosas que hacen que este carril no pierda ni duplique archivos: el +**ensure-then-upload** cuando el provisional todavía no existe allá, el **corte directo** +(``delete_local``) que borra la copia local solo al confirmar, y la idempotencia por +``crm_document_ref`` cuando un timeout ambiguo hace reintentar. +""" + +import pytest + +from api.v1.modules.crm.expediente_gateway import service as gateway +from api.v1.modules.crm.expediente_gateway.models import ( + FILE_KIND_DOCUMENTO, + MAX_ATTEMPTS, + SOURCE_CRM_DOCUMENTS, + STATUS_FAILED, + STATUS_PENDING, + STATUS_SENT, + EfcFileOutbox, +) +from api.v1.modules.crm.expedientes import service as expedientes_service +from api.v1.modules.crm.documents.models import Document +from api.v1.modules.crm.service_requests import service as sr_service +from api.v1.modules.crm.service_requests.dto import ServiceRequestCreate +from core.efc_client import EfcClientError +from tests.conftest import COMPANY_ID, TENANT_ID + +CONTENIDO = b"%PDF-1.4 guia madre" +S3_KEY = "tenants/1/companies/1/expedientes/1/guia.pdf" + + +class _ClienteFalso: + """Doble del cliente de EFC. Registra qué se le pidió, para poder afirmarlo.""" + + is_configured = True + + def __init__(self, *, upload_falla_con=None, falla_solo_la_primera=True): + self.uploads = [] + self.ingests = [] + self._upload_falla_con = upload_falla_con + self._falla_solo_la_primera = falla_solo_la_primera + + def ingest_expediente(self, payload): + self.ingests.append(payload) + return {"status": "created", "efc": {"pedimento_id": "ped-1"}} + + def upload_documento(self, org_id, company_id, expediente_id, tipo, filename, content, + content_type="application/octet-stream", crm_document_ref=None): + primera = not self.uploads + self.uploads.append({ + "org_id": org_id, "company_id": company_id, "expediente_id": expediente_id, + "tipo": tipo, "filename": filename, "content": content, + "content_type": content_type, "crm_document_ref": crm_document_ref, + }) + if self._upload_falla_con is not None and (primera or not self._falla_solo_la_primera): + raise self._upload_falla_con + return {"id": f"doc-{len(self.uploads)}"} + + +@pytest.fixture() +def entorno(db, monkeypatch): + """EFC encendido, storage simulado y organización ya resuelta.""" + from core.config import settings + + monkeypatch.setattr(settings, "EFC_API_URL", "https://efc.example.test/", raising=False) + monkeypatch.setattr(gateway, "_dispatch_delivery", lambda *a, **k: None) + monkeypatch.setattr(gateway, "_dispatch_file_delivery", lambda *a, **k: None) + monkeypatch.setattr(gateway, "_tenant_slug", lambda tid: ("temex", "TEMEX")) + monkeypatch.setattr(gateway, "_resolve_org_id", lambda c, t: "org-1") + + borrados = [] + import core.storage_s3 as storage + + monkeypatch.setattr(storage, "get_object_bytes", lambda key: CONTENIDO) + monkeypatch.setattr(storage, "delete_object_if_exists", lambda key: borrados.append(key)) + + solicitud = sr_service.create_service_request( + db, ServiceRequestCreate(operation_type="importacion"), TENANT_ID, COMPANY_ID, "user-1" + ) + expediente = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID) + return {"db": db, "expediente": expediente, "borrados": borrados} + + +def _documento_local(db, expediente) -> Document: + doc = Document( + doc_type="MBL", + name="guia.pdf", + file_key=S3_KEY, + content_type="application/pdf", + size_bytes=len(CONTENIDO), + expediente_id=expediente.id, + efc_sync_state="PENDING", + tenant_id=TENANT_ID, + company_id=COMPANY_ID, + ) + db.add(doc) + db.flush() + doc.efc_document_ref = f"CRMDOC-{COMPANY_ID}-{doc.id}" + db.commit() + return doc + + +def _fila(db, expediente, documento, **kwargs) -> EfcFileOutbox: + row = EfcFileOutbox( + kind=FILE_KIND_DOCUMENTO, + s3_key=S3_KEY, + file_name="guia.pdf", + content_type="application/pdf", + efc_tipo="MBL", + source_table=SOURCE_CRM_DOCUMENTS, + source_id=documento.id, + crm_document_ref=documento.efc_document_ref, + expediente_ref=expediente.id, + delete_local=kwargs.pop("delete_local", True), + status=kwargs.pop("status", STATUS_PENDING), + tenant_id=TENANT_ID, + company_id=COMPANY_ID, + **kwargs, + ) + db.add(row) + db.commit() + return row + + +# ── Camino feliz ───────────────────────────────────────────────────────────── + +def test_entrega_feliz_marca_la_fila_y_el_documento(entorno): + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento) + cliente = _ClienteFalso() + + gateway.deliver_file_row(db, row, cliente) + + assert row.status == STATUS_SENT + assert row.efc_document_id == "doc-1" + assert row.sent_at is not None + assert documento.efc_sync_state == "SYNCED" + assert documento.efc_document_id == "doc-1" + assert documento.efc_synced_at is not None + + +def test_la_subida_lleva_el_crm_document_ref_y_el_contenido_leido_de_minio(entorno): + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento) + cliente = _ClienteFalso() + + gateway.deliver_file_row(db, row, cliente) + + assert len(cliente.uploads) == 1 + subida = cliente.uploads[0] + assert subida["crm_document_ref"] == documento.efc_document_ref + assert subida["content"] == CONTENIDO + assert subida["tipo"] == "MBL" + assert subida["expediente_id"] == expediente.id + + +# ── Ensure-then-upload ─────────────────────────────────────────────────────── + +def test_un_404_de_expediente_crea_el_provisional_y_reintenta_una_vez(entorno): + """La creación del provisional y la subida son colas distintas: la subida puede llegar antes. + + Sin esto, el primer documento de cada expediente fallaría y esperaría al barrido. + """ + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento) + cliente = _ClienteFalso( + upload_falla_con=EfcClientError("no está", status_code=404, code="expediente_no_encontrado") + ) + + gateway.deliver_file_row(db, row, cliente) + + assert len(cliente.ingests) == 1 + assert cliente.ingests[0]["folio"] == expediente.folio + assert cliente.ingests[0]["storage_token"] == expediente.efc_storage_token + assert len(cliente.uploads) == 2 # el que falló + UNO de reintento + assert row.status == STATUS_SENT + + +def test_un_404_con_OTRO_code_no_dispara_el_ensure(entorno): + """El ensure se dispara por el ``code``, no por el 404 a secas. + + Si se disparara por cualquier 404, un documento no encontrado crearía provisionales espurios. + """ + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento) + cliente = _ClienteFalso( + upload_falla_con=EfcClientError("otro", status_code=404, code="documento_no_encontrado"), + falla_solo_la_primera=False, + ) + + gateway.deliver_file_row(db, row, cliente) + + assert cliente.ingests == [] + assert len(cliente.uploads) == 1 + assert row.attempts == 1 + + +def test_el_ensure_reintenta_UNA_vez_y_no_entra_en_bucle(entorno): + """Si el reintento vuelve a dar 404, se registra el fallo. No se reintenta indefinidamente.""" + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento) + cliente = _ClienteFalso( + upload_falla_con=EfcClientError("no está", status_code=404, code="expediente_no_encontrado"), + falla_solo_la_primera=False, + ) + + gateway.deliver_file_row(db, row, cliente) + + assert len(cliente.ingests) == 1 + assert len(cliente.uploads) == 2 + assert row.attempts == 1 + assert row.status == STATUS_FAILED # un 404 no es retryable + + +# ── Corte directo (delete_local) ───────────────────────────────────────────── + +def test_al_confirmar_se_borra_la_copia_local(entorno): + """«EFC es la fuente única» se cumple así: el objeto local se borra AL CONFIRMAR, no antes.""" + db, expediente, borrados = entorno["db"], entorno["expediente"], entorno["borrados"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento, delete_local=True) + + gateway.deliver_file_row(db, row, _ClienteFalso()) + + assert borrados == [S3_KEY] + # La key local se limpia: dejarla apuntaría a un objeto que ya no existe y la descarga se + # ramificaría por el camino equivocado. + assert documento.file_key is None + + +def test_con_delete_local_en_false_no_se_borra_nada(entorno): + db, expediente, borrados = entorno["db"], entorno["expediente"], entorno["borrados"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento, delete_local=False) + + gateway.deliver_file_row(db, row, _ClienteFalso()) + + assert borrados == [] + assert documento.file_key == S3_KEY + assert row.status == STATUS_SENT + + +def test_si_el_borrado_local_falla_la_entrega_sigue_siendo_valida(entorno, monkeypatch): + """El archivo YA está en EFC. No poder borrar la copia local no invalida la entrega, y volver a + intentarla subiría el mismo documento otra vez.""" + db, expediente = entorno["db"], entorno["expediente"] + import core.storage_s3 as storage + + def _revienta(key): + raise RuntimeError("MinIO no responde") + + monkeypatch.setattr(storage, "delete_object_if_exists", _revienta) + + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento, delete_local=True) + + gateway.deliver_file_row(db, row, _ClienteFalso()) + + assert row.status == STATUS_SENT + assert documento.efc_sync_state == "SYNCED" + + +def test_el_borrado_local_ocurre_ANTES_de_marcar_enviada_pero_no_antes_de_subir(entorno): + """Nunca se borra el original antes de confirmar la subida: si se borrara primero y la subida + fallara, el archivo se habría perdido.""" + db, expediente, borrados = entorno["db"], entorno["expediente"], entorno["borrados"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento, delete_local=True) + cliente = _ClienteFalso( + upload_falla_con=EfcClientError("500", status_code=500, retryable=True), + falla_solo_la_primera=False, + ) + + gateway.deliver_file_row(db, row, cliente) + + assert borrados == [] # la subida falló: el original sigue ahí + assert row.status == STATUS_PENDING + assert documento.file_key == S3_KEY + + +# ── Fallos ─────────────────────────────────────────────────────────────────── + +def test_un_fallo_de_efc_no_propaga_y_se_refleja_en_el_documento(entorno): + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento, attempts=MAX_ATTEMPTS - 1) + cliente = _ClienteFalso( + upload_falla_con=EfcClientError("EFC caído", status_code=503, retryable=True), + falla_solo_la_primera=False, + ) + + gateway.deliver_file_row(db, row, cliente) # no lanza + + assert row.status == STATUS_FAILED + assert documento.efc_sync_state == "FAILED" + assert "EFC caído" in documento.efc_error_detail + assert documento.efc_attempts == MAX_ATTEMPTS + + +def test_una_fila_ya_enviada_no_vuelve_a_subir_el_archivo(entorno): + """Segunda guarda de idempotencia. Un re-despacho tras un timeout ambiguo no duplica.""" + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento, status=STATUS_SENT) + cliente = _ClienteFalso() + + gateway.deliver_file_row(db, row, cliente) + + assert cliente.uploads == [] + + +def test_un_expediente_borrado_deja_la_fila_reintentable(entorno): + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento) + row.expediente_ref = 999999 + db.commit() + + gateway.deliver_file_row(db, row, _ClienteFalso()) + + assert row.status == STATUS_PENDING # retryable: el expediente puede reaparecer + assert row.attempts == 1 + + +def test_un_reintento_tras_timeout_manda_el_mismo_ref_y_no_duplica(entorno): + """El ``crm_document_ref`` es estable entre reintentos: EFC devuelve 200 con el que ya existía. + + Es la tercera capa de idempotencia y la que cubre el timeout ambiguo —EFC commiteó y contestó + tarde—, donde el CRM no puede saber si el documento entró. + """ + db, expediente = entorno["db"], entorno["expediente"] + documento = _documento_local(db, expediente) + row = _fila(db, expediente, documento) + + primer_cliente = _ClienteFalso( + upload_falla_con=EfcClientError("timeout", retryable=True), falla_solo_la_primera=False + ) + gateway.deliver_file_row(db, row, primer_cliente) + assert row.status == STATUS_PENDING + + segundo_cliente = _ClienteFalso() + gateway.deliver_file_row(db, row, segundo_cliente) + + assert primer_cliente.uploads[0]["crm_document_ref"] == segundo_cliente.uploads[0]["crm_document_ref"] + assert row.status == STATUS_SENT diff --git a/backend/tests/test_efc_outbox.py b/backend/tests/test_efc_outbox.py new file mode 100644 index 0000000..f87c97f --- /dev/null +++ b/backend/tests/test_efc_outbox.py @@ -0,0 +1,386 @@ +"""Pruebas de la máquina de reintentos del outbox hacia EFC. + +Lo que se fija aquí es la capa 2 de las tres del carril: el worker **nunca lanza**, registra el +fallo en la propia fila, y decide reintentar o rendirse por el campo ``retryable`` —nunca parseando +el texto del error—. +""" + +import pytest + +from api.v1.modules.crm.expediente_gateway import service as gateway +from api.v1.modules.crm.expediente_gateway.models import ( + FILE_KIND_DOCUMENTO, + KIND_EXPEDIENTE, + MAX_ATTEMPTS, + SOURCE_CRM_DOCUMENTS, + SOURCE_OPS_SHIPMENT_DOCUMENTS, + STATUS_FAILED, + STATUS_PENDING, + STATUS_SENT, + EfcFileOutbox, + EfcSyncOutbox, +) +from api.v1.modules.crm.expedientes import service as expedientes_service +from api.v1.modules.crm.service_requests import service as sr_service +from api.v1.modules.crm.service_requests.dto import ServiceRequestCreate +from core.efc_client import EfcClientError +from tests.conftest import COMPANY_ID, TENANT_ID + +OTRO_TENANT = 99 + + +@pytest.fixture() +def efc_encendido(monkeypatch): + """Enciende la integración y evita que el encolado toque el broker o el tenant real.""" + from core.config import settings + + monkeypatch.setattr(settings, "EFC_API_URL", "https://efc.example.test/", raising=False) + monkeypatch.setattr(gateway, "_dispatch_delivery", lambda *a, **k: None) + monkeypatch.setattr(gateway, "_dispatch_file_delivery", lambda *a, **k: None) + monkeypatch.setattr(gateway, "_tenant_slug", lambda tid: ("temex", "TEMEX")) + return settings + + +@pytest.fixture() +def efc_apagado(monkeypatch): + from core.config import settings + + monkeypatch.setattr(settings, "EFC_API_URL", "", raising=False) + return settings + + +def _expediente(db): + solicitud = sr_service.create_service_request( + db, ServiceRequestCreate(operation_type="importacion"), TENANT_ID, COMPANY_ID, "user-1" + ) + return expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID) + + +def _fila_sync(db, expediente, **kwargs): + row = EfcSyncOutbox( + kind=kwargs.pop("kind", KIND_EXPEDIENTE), + payload=kwargs.pop("payload", {"folio": expediente.folio}), + expediente_ref=expediente.id, + status=kwargs.pop("status", STATUS_PENDING), + tenant_id=kwargs.pop("tenant_id", TENANT_ID), + company_id=kwargs.pop("company_id", COMPANY_ID), + **kwargs, + ) + db.add(row) + db.commit() + return row + + +def _fila_archivo(db, expediente, **kwargs): + row = EfcFileOutbox( + kind=kwargs.pop("kind", FILE_KIND_DOCUMENTO), + s3_key=kwargs.pop("s3_key", "tenants/1/companies/1/expedientes/1/guia.pdf"), + file_name=kwargs.pop("file_name", "guia.pdf"), + content_type="application/pdf", + efc_tipo=kwargs.pop("efc_tipo", "MBL"), + source_table=kwargs.pop("source_table", SOURCE_CRM_DOCUMENTS), + source_id=kwargs.pop("source_id", 1), + crm_document_ref=kwargs.pop("crm_document_ref", "CRMDOC-1-1"), + expediente_ref=expediente.id, + status=kwargs.pop("status", STATUS_PENDING), + tenant_id=kwargs.pop("tenant_id", TENANT_ID), + company_id=kwargs.pop("company_id", COMPANY_ID), + **kwargs, + ) + db.add(row) + db.commit() + return row + + +# ── _register_failure ──────────────────────────────────────────────────────── + +def test_un_fallo_retryable_suma_un_intento_y_deja_la_fila_pendiente(db, efc_encendido): + expediente = _expediente(db) + row = _fila_sync(db, expediente) + + gateway._register_failure(db, row, EfcClientError("EFC no responde", retryable=True), True) + + assert row.attempts == 1 + assert row.status == STATUS_PENDING + assert "EFC no responde" in row.last_error + + +def test_un_fallo_no_retryable_marca_failed_de_inmediato(db, efc_encendido): + """Un 400 no mejora insistiendo: reintentarlo ocho veces solo retrasa que alguien lo vea.""" + expediente = _expediente(db) + row = _fila_sync(db, expediente) + + gateway._register_failure(db, row, EfcClientError("tipo inválido", retryable=False), False) + + assert row.attempts == 1 + assert row.status == STATUS_FAILED + + +def test_al_llegar_a_max_attempts_la_fila_queda_failed(db, efc_encendido): + expediente = _expediente(db) + row = _fila_sync(db, expediente, attempts=MAX_ATTEMPTS - 1) + + gateway._register_failure(db, row, EfcClientError("otra vez", retryable=True), True) + + assert row.attempts == MAX_ATTEMPTS + assert row.status == STATUS_FAILED + + +def test_el_ultimo_error_se_trunca_a_2000_caracteres(db, efc_encendido): + """``last_error`` es Text, pero un traceback de 5000 caracteres por fila llena la tabla de ruido.""" + expediente = _expediente(db) + row = _fila_sync(db, expediente) + + gateway._register_failure(db, row, Exception("x" * 5000), True) + + assert len(row.last_error) == 2000 + + +# ── deliver_row: no propaga ────────────────────────────────────────────────── + +class _ClienteQueRevienta: + is_configured = True + + def __init__(self, exc): + self._exc = exc + self.llamadas = 0 + + def ingest_expediente(self, payload): + self.llamadas += 1 + raise self._exc + + def completar_expediente(self, folio, payload): + self.llamadas += 1 + raise self._exc + + +def test_deliver_row_no_propaga_la_excepcion_de_efc(db, efc_encendido, monkeypatch): + """Si esto propagara, un EFC caído mataría al worker y se perdería la cola entera.""" + expediente = _expediente(db) + row = _fila_sync(db, expediente) + monkeypatch.setattr(gateway, "_resolve_org_id", lambda c, t: "org-1") + + gateway.deliver_row(db, row, _ClienteQueRevienta(EfcClientError("caído", retryable=True))) + + assert row.status == STATUS_PENDING + assert row.attempts == 1 + + +def test_deliver_row_tampoco_propaga_una_excepcion_inesperada(db, efc_encendido, monkeypatch): + expediente = _expediente(db) + row = _fila_sync(db, expediente) + monkeypatch.setattr(gateway, "_resolve_org_id", lambda c, t: "org-1") + + gateway.deliver_row(db, row, _ClienteQueRevienta(RuntimeError("algo raro"))) + + assert row.attempts == 1 + # Una excepción inesperada se trata como transitoria: no se sabe que sea permanente. + assert row.status == STATUS_PENDING + + +def test_una_fila_ya_enviada_no_vuelve_a_llamar_a_efc(db, efc_encendido): + """Segunda guarda de idempotencia. Sin ella, un re-despacho duplicaría el expediente en EFC.""" + expediente = _expediente(db) + row = _fila_sync(db, expediente, status=STATUS_SENT) + cliente = _ClienteQueRevienta(EfcClientError("no debería llamarse")) + + gateway.deliver_row(db, row, cliente) + + assert cliente.llamadas == 0 + assert row.status == STATUS_SENT + + +# ── _ya_entregado: la ambigüedad de las dos secuencias ─────────────────────── + +def test_no_se_encola_dos_veces_el_mismo_archivo(db, efc_encendido): + expediente = _expediente(db) + _fila_archivo(db, expediente, source_id=7, status=STATUS_SENT) + + assert gateway._ya_entregado(db, SOURCE_CRM_DOCUMENTS, 7, FILE_KIND_DOCUMENTO) is True + + row = gateway.enqueue_file_best_effort( + db, kind=FILE_KIND_DOCUMENTO, s3_key="k", file_name="f.pdf", content_type=None, + efc_tipo="MBL", source_table=SOURCE_CRM_DOCUMENTS, source_id=7, + crm_document_ref="CRMDOC-1-7", expediente_ref=expediente.id, + tenant_id=TENANT_ID, company_id=COMPANY_ID, + ) + assert row is None + + +def test_el_mismo_id_en_otra_tabla_de_origen_SI_se_encola(db, efc_encendido): + """La prueba de la ambigüedad de las dos secuencias. + + ``crm.documents.id = 7`` y ``ops.shipment_documents.id = 7`` son documentos DISTINTOS. Sin + ``source_table`` en la guarda, entregar el primero haría que el segundo se saltara para + siempre — y nadie vería un error. + """ + expediente = _expediente(db) + _fila_archivo(db, expediente, source_id=7, source_table=SOURCE_CRM_DOCUMENTS, status=STATUS_SENT) + + assert gateway._ya_entregado(db, SOURCE_OPS_SHIPMENT_DOCUMENTS, 7, FILE_KIND_DOCUMENTO) is False + + row = gateway.enqueue_file_best_effort( + db, kind=FILE_KIND_DOCUMENTO, s3_key="k", file_name="f.pdf", content_type=None, + efc_tipo="MBL", source_table=SOURCE_OPS_SHIPMENT_DOCUMENTS, source_id=7, + crm_document_ref="SHPDOC-1-7", expediente_ref=expediente.id, + tenant_id=TENANT_ID, company_id=COMPANY_ID, + ) + assert row is not None + assert row.source_table == SOURCE_OPS_SHIPMENT_DOCUMENTS + + +# ── retry ──────────────────────────────────────────────────────────────────── + +def test_retry_resetea_la_fila_y_la_re_despacha(db, efc_encendido, monkeypatch): + despachos = [] + monkeypatch.setattr( + gateway, "_dispatch_file_delivery", lambda oid, t, c: despachos.append((oid, t, c)) + ) + expediente = _expediente(db) + row = _fila_archivo(db, expediente, status=STATUS_FAILED, attempts=MAX_ATTEMPTS, + last_error="se acabaron los intentos") + + ok = gateway.retry_outbox_row(db, row.id, TENANT_ID, COMPANY_ID, "file") + + assert ok is True + assert row.status == STATUS_PENDING + assert row.attempts == 0 + assert row.last_error is None + assert despachos == [(row.id, TENANT_ID, COMPANY_ID)] + + +def test_retry_de_otro_tenant_devuelve_false(db, efc_encendido): + """Devuelve False y el llamador lo traduce a 404: un 200 le haría creer al frontend que se + reencoló algo que ni siquiera es suyo.""" + expediente = _expediente(db) + row = _fila_archivo(db, expediente, status=STATUS_FAILED) + + assert gateway.retry_outbox_row(db, row.id, OTRO_TENANT, COMPANY_ID, "file") is False + assert row.status == STATUS_FAILED # intacta + + +def test_retry_de_una_fila_inexistente_devuelve_false(db, efc_encendido): + assert gateway.retry_outbox_row(db, 999999, TENANT_ID, COMPANY_ID, "file") is False + + +# ── métricas y listado ─────────────────────────────────────────────────────── + +def test_las_metricas_cuentan_por_status_sumando_las_dos_tablas(db, efc_encendido): + expediente = _expediente(db) + _fila_sync(db, expediente, status=STATUS_SENT) + _fila_archivo(db, expediente, source_id=1, status=STATUS_PENDING) + _fila_archivo(db, expediente, source_id=2, status=STATUS_FAILED) + _fila_archivo(db, expediente, source_id=3, status=STATUS_FAILED) + + metricas = gateway.outbox_metrics(db, TENANT_ID, COMPANY_ID) + + # El alta del expediente encoló su propia fila pendiente al crearse la solicitud. + assert metricas["failed"] == 2 + assert metricas["sent"] == 1 + assert metricas["pending"] >= 1 + + +def test_las_metricas_no_ven_las_filas_de_otro_tenant(db, efc_encendido): + expediente = _expediente(db) + _fila_archivo(db, expediente, source_id=5, status=STATUS_FAILED, tenant_id=OTRO_TENANT) + + assert gateway.outbox_metrics(db, TENANT_ID, COMPANY_ID)["failed"] == 0 + + +def test_el_listado_marca_de_que_tabla_viene_cada_fila(db, efc_encendido): + expediente = _expediente(db) + _fila_archivo(db, expediente, source_id=1) + + filas = gateway.list_outbox(db, TENANT_ID, COMPANY_ID) + tablas = {f["tabla"] for f in filas} + assert tablas == {"sync", "file"} + + solo_archivos = gateway.list_outbox(db, TENANT_ID, COMPANY_ID, tipo="file") + assert {f["tabla"] for f in solo_archivos} == {"file"} + + +# ── huecos ─────────────────────────────────────────────────────────────────── + +def test_find_expediente_gaps_encuentra_los_que_no_tienen_fila(db, efc_apagado): + """Con EFC apagado no se encola nada, así que todos los expedientes son huecos. + + Es exactamente el caso que el barrido cubre: lo creado ANTES de activar la integración. + """ + _expediente(db) + _expediente(db) + + huecos = gateway.find_expediente_gaps(db) + assert len(huecos) == 2 + + +def test_un_expediente_con_fila_failed_NO_es_un_hueco(db, efc_encendido): + """Un ``failed`` existe como fila: es visible en el tablero y reintentable a mano. + + Tratarlo como hueco lo re-encolaría en cada barrido y escondería el fallo. + """ + expediente = _expediente(db) + fila = db.query(EfcSyncOutbox).filter(EfcSyncOutbox.expediente_ref == expediente.id).first() + assert fila is not None + fila.status = STATUS_FAILED + db.commit() + + assert gateway.find_expediente_gaps(db) == [] + + +# ── best-effort ────────────────────────────────────────────────────────────── + +def test_con_efc_apagado_no_se_encola_nada(db, efc_apagado): + """``EFC_API_URL`` vacía apaga el carril entero. El CRM sigue funcionando igual.""" + expediente = _expediente(db) + + assert db.query(EfcSyncOutbox).count() == 0 + + fila = gateway.enqueue_file_best_effort( + db, kind=FILE_KIND_DOCUMENTO, s3_key="k", file_name="f.pdf", content_type=None, + efc_tipo="MBL", source_table=SOURCE_CRM_DOCUMENTS, source_id=1, + crm_document_ref="CRMDOC-1-1", expediente_ref=expediente.id, + tenant_id=TENANT_ID, company_id=COMPANY_ID, + ) + assert fila is None + assert db.query(EfcFileOutbox).count() == 0 + + +def test_con_efc_encendido_crear_una_solicitud_encola_su_expediente(db, efc_encendido): + expediente = _expediente(db) + + filas = db.query(EfcSyncOutbox).filter(EfcSyncOutbox.expediente_ref == expediente.id).all() + assert len(filas) == 1 + assert filas[0].kind == KIND_EXPEDIENTE + assert filas[0].status == STATUS_PENDING + assert filas[0].payload["folio"] == expediente.folio + assert filas[0].payload["storage_token"] == expediente.efc_storage_token + + +def test_si_el_encolado_revienta_la_operacion_local_no_se_rompe(db, efc_encendido, monkeypatch): + """La integración NUNCA puede tumbar el alta de una solicitud del usuario.""" + def _revienta(*a, **k): + raise RuntimeError("la tabla del outbox no existe") + + monkeypatch.setattr(gateway, "_expediente_ya_encolado", _revienta) + + solicitud = sr_service.create_service_request( + db, ServiceRequestCreate(operation_type="importacion"), TENANT_ID, COMPANY_ID, "user-1" + ) + + assert solicitud.id is not None + assert expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID) is not None + + +def test_no_se_encola_dos_veces_el_mismo_expediente(db, efc_encendido): + """Primera guarda: ``ensure`` es idempotente y no debe generar una segunda réplica.""" + expediente = _expediente(db) + solicitud_id = expediente.service_request_id + + expedientes_service.ensure_expediente(db, solicitud_id, TENANT_ID, COMPANY_ID, "user-1") + expedientes_service.ensure_expediente(db, solicitud_id, TENANT_ID, COMPANY_ID, "user-1") + + filas = db.query(EfcSyncOutbox).filter( + EfcSyncOutbox.expediente_ref == expediente.id, + EfcSyncOutbox.kind == KIND_EXPEDIENTE, + ).all() + assert len(filas) == 1 diff --git a/backend/tests/test_gateway_rutas.py b/backend/tests/test_gateway_rutas.py new file mode 100644 index 0000000..40494f4 --- /dev/null +++ b/backend/tests/test_gateway_rutas.py @@ -0,0 +1,134 @@ +"""Contrato del tablero de ops del carril CRM -> EFC. + +Lo que se fija aquí es lo que el frontend espera recibir: el 404 del reintento sobre una fila que no +existe (y **no** un 200 silencioso), la forma exacta de la respuesta de éxito, y el aislamiento por +tenant/company. +""" + +import pytest +from fastapi import HTTPException + +from api.v1.modules.crm.expediente_gateway import routes +from api.v1.modules.crm.expediente_gateway import service as gateway +from api.v1.modules.crm.expediente_gateway.models import ( + FILE_KIND_DOCUMENTO, + SOURCE_CRM_DOCUMENTS, + STATUS_FAILED, + STATUS_PENDING, + STATUS_SENT, + EfcFileOutbox, +) +from api.v1.modules.crm.expedientes import service as expedientes_service +from api.v1.modules.crm.service_requests import service as sr_service +from api.v1.modules.crm.service_requests.dto import ServiceRequestCreate +from tests.conftest import COMPANY_ID, TENANT_ID + +OTRO_TENANT = 99 +OTRA_COMPANY = 77 +USUARIO = {"tenant_id": TENANT_ID, "sub": "user-1"} + + +@pytest.fixture() +def entorno(db, monkeypatch): + from core.config import settings + + monkeypatch.setattr(settings, "EFC_API_URL", "https://efc.example.test/", raising=False) + monkeypatch.setattr(gateway, "_dispatch_delivery", lambda *a, **k: None) + monkeypatch.setattr(gateway, "_dispatch_file_delivery", lambda *a, **k: None) + monkeypatch.setattr(gateway, "_tenant_slug", lambda tid: ("temex", "TEMEX")) + + solicitud = sr_service.create_service_request( + db, ServiceRequestCreate(operation_type="importacion"), TENANT_ID, COMPANY_ID, "user-1" + ) + expediente = expedientes_service.find_by_service_request(db, solicitud.id, TENANT_ID, COMPANY_ID) + return {"db": db, "expediente": expediente} + + +def _fila_archivo(db, expediente, **kwargs) -> EfcFileOutbox: + row = EfcFileOutbox( + kind=FILE_KIND_DOCUMENTO, + s3_key="k", + file_name="guia.pdf", + content_type="application/pdf", + efc_tipo="MBL", + source_table=SOURCE_CRM_DOCUMENTS, + source_id=kwargs.pop("source_id", 1), + crm_document_ref="CRMDOC-1-1", + expediente_ref=expediente.id, + status=kwargs.pop("status", STATUS_PENDING), + tenant_id=kwargs.pop("tenant_id", TENANT_ID), + company_id=kwargs.pop("company_id", COMPANY_ID), + **kwargs, + ) + db.add(row) + db.commit() + return row + + +def test_retry_de_una_fila_inexistente_da_404_con_mensaje_especifico(entorno): + """**Es contrato con el frontend.** Un 200 le haría pintar «reencolado» cuando no hay nada que + entregar, y el usuario esperaría un badge que nunca va a cambiar.""" + with pytest.raises(HTTPException) as exc: + routes.retry_outbox(999999, COMPANY_ID, "file", USUARIO, entorno["db"]) + + assert exc.value.status_code == 404 + assert exc.value.detail == "Fila de outbox no encontrada" + + +def test_retry_exitoso_devuelve_requeued_con_el_id(entorno): + row = _fila_archivo(entorno["db"], entorno["expediente"], status=STATUS_FAILED) + + resp = routes.retry_outbox(row.id, COMPANY_ID, "file", USUARIO, entorno["db"]) + + assert resp == {"status": "requeued", "id": row.id} + assert row.status == STATUS_PENDING + + +def test_retry_de_una_fila_de_otra_company_da_404(entorno): + """No se filtra la existencia: para ese usuario la fila simplemente no existe.""" + row = _fila_archivo(entorno["db"], entorno["expediente"], company_id=OTRA_COMPANY) + + with pytest.raises(HTTPException) as exc: + routes.retry_outbox(row.id, COMPANY_ID, "file", USUARIO, entorno["db"]) + assert exc.value.status_code == 404 + + +def test_metrics_cuenta_por_status(entorno): + db, expediente = entorno["db"], entorno["expediente"] + _fila_archivo(db, expediente, source_id=1, status=STATUS_FAILED) + _fila_archivo(db, expediente, source_id=2, status=STATUS_SENT) + + metricas = routes.metrics(COMPANY_ID, USUARIO, db) + + assert set(metricas) == {"pending", "sent", "failed"} + assert metricas["failed"] == 1 + assert metricas["sent"] == 1 + + +def test_metrics_no_ve_otro_tenant(entorno): + db, expediente = entorno["db"], entorno["expediente"] + _fila_archivo(db, expediente, source_id=3, status=STATUS_FAILED, tenant_id=OTRO_TENANT) + + assert routes.metrics(COMPANY_ID, USUARIO, db)["failed"] == 0 + + +def test_el_listado_solo_devuelve_lo_del_tenant_y_la_company(entorno): + db, expediente = entorno["db"], entorno["expediente"] + _fila_archivo(db, expediente, source_id=1) + _fila_archivo(db, expediente, source_id=2, tenant_id=OTRO_TENANT) + _fila_archivo(db, expediente, source_id=3, company_id=OTRA_COMPANY) + + filas = routes.list_outbox(COMPANY_ID, "file", None, 100, USUARIO, db) + + assert len(filas) == 1 + assert filas[0]["source_id"] == 1 + + +def test_el_listado_filtra_por_status(entorno): + db, expediente = entorno["db"], entorno["expediente"] + _fila_archivo(db, expediente, source_id=1, status=STATUS_FAILED) + _fila_archivo(db, expediente, source_id=2, status=STATUS_SENT) + + fallidas = routes.list_outbox(COMPANY_ID, "file", STATUS_FAILED, 100, USUARIO, db) + + assert [f["source_id"] for f in fallidas] == [1]