Sincronizar Conversaciones

Autor: Iker Acevedo¶

Fecha creación: 2026-08-04

Estado: producción

Lambda: ApiLambdaSincronizarConversacionesPancake¶

Accionador: Step Functions (Task, dentro de un Map — 1 invocación por página)

AOT: No

Posición en el pipeline: 2 de 3 (ver Orquestación)

¿Qué hace?¶

Trae todas las conversaciones nuevas o actualizadas de 1 página de Pancake (WhatsApp/Facebook) y las guarda en PancakeConversaciones (MySQL) con UPSERT. No trae mensajes — solo la metadata de la conversación (quién escribió, cuándo se actualizó, tipo de canal).

Usa un watermark por página (MAX(FechaActualizacionPancake) de lo ya guardado) para pedirle a Pancake solo lo que cambió desde la última corrida — nunca re-trae todo el historial completo en cada ejecución.

Request¶

Recibe el page_id de la página a sincronizar (lo inyecta el Map del Step Function, 1 por iteración):

{ "PageId": "waba_954021881132494" }

Response¶

{
  "PageId": "waba_954021881132494",
  "TotalSincronizadas": 62,
  "Error": false,
  "MensajeError": null
}

Flujo de datos¶

flowchart TD
    A["Step Function Map\n(1 invocación por página)"] --> B["Function.cs"]
    B --> C["SincronizarConversacionesUseCase"]
    C --> D["PaginaTokenRepository\n(Mongo: token de página + token de usuario)"]
    C --> E["ConversacionRepository.ObtenerWaterMarkAsync\n(MySQL: MAX FechaActualizacionPancake)"]
    E --> F["since = max(watermark, ahora - MAXIMO_DIAS_SINCRONIZAR)"]
    F --> G{"Llamar Pancake API\nObtenerConversacionesAsync"}
    G -- "token vencido (PancakeApiException)" --> H["GenerarTokenPaginaAsync\n(regenerar token)"]
    H --> G
    G -- OK --> I["ConversacionMapper.AEntidad\n(por cada conversación del lote)"]
    I --> J[("MySQL: PancakeConversaciones\nUPSERT por lote")]
    J --> K{"¿Vinieron 60 filas\n(página llena)?"}
    K -- Sí, sigue paginando --> G
    K -- No, fin --> L["Respuesta: SalidaSincronizacion\n{ PageId, TotalSincronizadas, Error }"]

Fuente de este diagrama: carpeta de diagramas.

Flujo interno (detalle en código)¶

FunctionHandler (Function.cs)
  -> SincronizarConversacionesUseCase.EjecutarAsync(pageId, ahoraUtc)
       1. IPaginaTokenRepository.ObtenerAsync(pageId)      -> Mongo: token de pagina + token de usuario (cuenta madre)
       2. IConversacionRepository.ObtenerWaterMarkAsync    -> MySQL: MAX(FechaActualizacionPancake) ya guardado
       3. since = max(watermark, ahoraUtc - MAXIMO_DIAS_SINCRONIZAR)   // nunca trae mas historico que el tope
       4. Loop paginado (60 por pagina) contra Pancake:
            IPancakeConversacionesClient.ObtenerConversacionesAsync(pageId, token, since, until, cursor)
              -> si el token esta vencido (PancakeApiException): regenera con GenerarTokenPaginaAsync y reintenta
            -> ConversacionMapper.AEntidad(dto, pageId, ahoraUtc)  por cada conversacion
            -> IConversacionRepository.GuardarLoteAsync            -> MySQL: PancakeConversaciones (UPSERT por lote)
       5. Repite hasta que Pancake devuelve menos de 60 (fin de pagina) o vacio
  -> [RESULT] SalidaSincronizacion { PageId, TotalSincronizadas, Error }

Arquitectura Clean Architecture¶

ApiLambdaSincronizarConversacionesPancake/
├── Function.cs
├── Dominio/
│   ├── Entidades/
│   │   └── Conversacion.cs          ← { ConversacionId, PageId, NumeroTelefono, NombreCliente, Tipo, Visto, FechaActualizacionPancake, DatosCrudos, FechaSincronizacion }
│   └── PancakeApiException.cs
├── Aplicacion/
│   ├── CasosUso/
│   │   └── SincronizarConversacionesUseCase.cs
│   ├── ConversacionMapper.cs         ← DTO Pancake -> Entidad de dominio
│   ├── DTO/
│   │   ├── ConversacionPancakeDto.cs
│   │   ├── EntradaSincronizacion.cs
│   │   └── SalidaSincronizacion.cs
│   └── Interfaces/
│       ├── IConversacionRepository.cs
│       └── IPancakeConversacionesClient.cs
└── Infraestructura/
    ├── Repositorio/
    │   ├── ConversacionRepository.cs   ← MySqlConnector, UPSERT por lote
    │   └── PaginaTokenRepository.cs    ← Mongo, token + regeneracion
    └── Servicios/
        └── PancakeConversacionesClient.cs  ← HttpClient contra Pancake

Tabla destino (MySQL)¶

PancakeConversaciones — UPSERT (ON DUPLICATE KEY UPDATE) por ConversacionId:

Variables de entorno¶

Configuración Lambda¶

Historial de cambios¶

Observaciones¶

• El watermark (ObtenerWaterMarkAsync) hace que corridas repetidas sean baratas: si nada cambió en Pancake desde la última vez, since queda igual y la 1ª página de resultados vuelve vacía casi de inmediato.
• Esta lambda corre 1 página a la vez, secuencial internamente (no tiene el paralelismo de ApiLambdaSincronizarMensajes) — el paralelismo entre páginas lo da el Map del Step Function (MaxConcurrency: 8), no esta lambda.
• Deuda pendiente: no persiste el token regenerado de vuelta a Mongo (a diferencia de ApiLambdaSincronizarMensajes, que sí lo hace) — como corre 1 sola invocación por página sin concurrencia interna, no sufre la condición de carrera que sí afectaba a la otra lambda, pero cada corrida nueva puede tener que regenerar el token de nuevo si Mongo quedó con el viejo.