"""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 ..cases.models import Case
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: Case) -> 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: Case) -> 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
        # Sin folio o sin token no hay nada que replicar: EFC exige los dos y responde
        # {'storage_token': ['This field may not be null.']}, que NO es un fallo transitorio.
        # Encolarlo de todos modos quemaría los 8 intentos para acabar en `failed`, ensuciando
        # el tablero de ops con algo que ningún reintento puede arreglar.
        #
        # Pasa de verdad en dos casos: expedientes nacidos antes de que existiera el carril
        # (los rellena la migración c5d6e7f8a9b0) y aquellos cuyo token no cabe en los 25
        # caracteres de `pedimento_app`. Se avisa en WARNING porque es una omisión silenciosa:
        # el expediente vive en el CRM y sus documentos nunca llegarán a EFC.
        if not expediente.reference or not expediente.efc_storage_token:
            logger.warning(
                "expediente_gateway: expediente id=%s SIN replicar — folio=%r token=%r. "
                "No se encola: EFC rechaza ambos nulos y el reintento no lo arregla.",
                expediente.id, expediente.reference, expediente.efc_storage_token,
            )
            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.reference,
            "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: Case, 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.reference,
                "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(Case).filter(Case.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(Case).filter(Case.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.reference,
                    "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 == Case.id)
    return (
        db.query(Case)
        .filter(
            Case.deleted_at.is_(None),
            ~ya_encolado,
            # Mismo criterio que el encolado: lo que le falta folio o token no es un hueco
            # recuperable, es algo que EFC rechazaría siempre. Sin este filtro la
            # reconciliación los reencola cada 5 minutos para verlos fallar de nuevo.
            Case.reference.isnot(None),
            Case.efc_storage_token.isnot(None),
        )
        .order_by(Case.id.desc())
        .limit(limit)
        .all()
    )
