Syntage · Entities sync (catálogo rfc ↔ entity_id)

Sincroniza el catálogo de entidades de Syntage (GET /entities) a una tabla de mapeo en BigQuery, manteniendo fresco entity_id ↔ rfc ↔ credential.status para que otros procesos resuelvan el entityId a partir del rfc.

¿Llegaste hasta acá desde…?

Si ves…	Eso sale de…	La lógica está en…
Un mapeo `rfc → entity_id` o el `credential_status` de una entidad	`solvento-data-prod.syntage.entities`	Ingesta — Cloud Run `syntage-entities-sync` (repo)
"¿Corrió el sync de entidades hoy / hasta qué watermark?"	`solvento-data-prod.syntage.entities_sync_control`	Tabla de control (watermark del incremental)
Un proceso downstream que no encuentra el `entity_id` de un RFC nuevo	`solvento-data-prod.syntage.entities`	Falta correr el sync incremental; el catálogo es la dependencia

Resumen

Campo	Valor
Origen	API Syntage `GET /entities` (paginación offset, `itemsPerPage=1000`)
Patrón	api-pull-programado (Cloud Run agendado por Cloud Scheduler)
Destino	`solvento-data-prod.syntage.entities` (+ `entities_sync_control`)
Frecuencia	Diaria `0 5 * * *` (job `syntage-entities-sync-daily`, modo incremental, OIDC); `full` para backfill / re-sync completo
Criticidad	🟡 Media (es dependencia de otros procesos)
Dueño	Martin
Repos	`syntage-entities-sync`

Diagrama de linaje

graph LR
    API[["🌐 API Syntage<br/>GET /entities"]]
    SCH(["⏰ Cloud Scheduler<br/>syntage-entities-sync-daily<br/><i>0 5 * * * · incremental</i>"])

    subgraph ing["Ingesta · Cloud Run programado"]
        CR["syntage-entities-sync<br/><i>full | incremental · MERGE por entity_id</i>"]
        ENT[("solvento-data-prod<br/>syntage.entities<br/>dimensión · upsert")]
        CTRL[("syntage.entities_<br/>sync_control<br/>watermark")]
    end

    DOWN["Procesos downstream<br/><i>resuelven rfc → entity_id</i><br/>(p. ej. moratorios)"]

    SCH --> CR
    API --> CR
    CR --> ENT
    CR --> CTRL
    ENT --> DOWN

    classDef src fill:#fde68a,stroke:#b45309,color:#000;
    classDef store fill:#bfdbfe,stroke:#1e40af,color:#000;
    class API,SCH src;
    class ENT,CTRL store;

Paso a paso

Ingesta — Cloud Run `syntage-entities-sync` (repo)

Resuelve el modo. incremental (default, corrida diaria): trae solo entidades con updatedAt posterior al último sync exitoso, usando el filtro updatedAt[after] de la API y aplicando un solapamiento (WATERMARK_OVERLAP_MINUTES, default 60m) para no perder registros de borde. El watermark sale del MAX(watermark) de las corridas con status='success' en entities_sync_control. full: barre todas las entidades con paginación offset (~3 páginas para ~2.5k entidades), para la primera carga o un re-sync completo.
Consulta GET /entities con paginación (ITEMS_PER_PAGE=1000) y manejo de rate limit (lee X-RateLimit-Remaining/X-RateLimit-Reset, pausa preventiva con RATE_LIMIT_MIN_REMAINING=5, backoff ante 429 con MAX_RETRIES=5). El sync es liviano, muy por debajo del límite de 60 req/min.
Escribe por MERGE (upsert) sobre entity_id en syntage.entities (el catálogo es una dimensión, no un log: cada entidad existe una sola vez y se actualiza in-place; load_dts se setea en cada upsert). Registra una fila por corrida en syntage.entities_sync_control (sync_id, mode, watermark, entities_fetched, entities_upserted, started_at, finished_at, status).

Endpoints: POST /sync (body opcional {"mode":"full"|"incremental"}), POST / (alias de /sync para Scheduler simple), GET /health.

Orden recomendado: conviene correrlo antes del delta de intereses moratorios (Scheduler 0 5 * * *), para que una entidad nueva que facturó por primera vez hoy ya tenga su entity_id resuelto.

Tablas que alimenta

Tabla	Contenido	Job que la carga
`solvento-data-prod.syntage.entities`	Catálogo `entity_id ↔ rfc ↔ credential_status` (+ tipo, nombre, fechas)	Cloud Run `syntage-entities-sync` (MERGE)
`solvento-data-prod.syntage.entities_sync_control`	Una fila por corrida con el watermark del incremental	Cloud Run `syntage-entities-sync`

Consumo aguas abajo

Procesos de ingesta que necesitan el entityId de Syntage a partir del rfc (los webhooks de invoices traen rfc, pero los endpoints de insights se consultan por entityId). Ejemplo: el delta diario de intereses moratorios (syntage-moratory-interest).

Si falla 🛠️

Impacto: el catálogo entities queda desactualizado; los procesos downstream que dependen del mapeo rfc → entity_id no resolverán entidades nuevas. Criticidad media (no bloquea producto, pero es dependencia).

Síntoma	Dónde mirar	Acción
Una entidad/RFC nuevo no resuelve `entity_id` aguas abajo	`syntage.entities`; `entities_sync_control` (¿corrió el incremental?)	Re-invocar `POST /sync {"mode":"incremental"}`
El catálogo parece muy desfasado	`entities_sync_control` (`watermark`, `status`)	Correr un `POST /sync {"mode":"full"}` para re-sincronizar todo
Errores 429 / rate limit	`X-RateLimit-*` en logs Cloud Run	El servicio ya hace pausa/backoff; el sync es liviano, revisar si la API está degradada

Escalación: Martin (Cloud Run de ingesta).

Enlaces

📦 Ingesta: syntage-entities-sync
📊 Estado en vivo: dashboard "Data Processes" (Cloud Monitoring)

Última actualización: 2026-06-21 · Si cambia el flujo, actualizar esta ficha en el mismo MR.