Syntage · CIEC extraction status (credenciales / extracciones)

Cloud Run con API REST que extrae el estado de las credenciales CIEC desde Syntage, las cruza con la tabla taxpayers de BigQuery, y persiste el estado diario de credenciales y el historial de extracciones en syntage.

¿Llegaste hasta acá desde…?

Si ves…	Eso sale de…	La lógica está en…
Estado diario de una credencial CIEC (valid/invalid) por RFC	`solvento-data-prod.syntage.daily_credential_status`	Cloud Run `syntage-ciec-extraction-status` (repo) → `GET /extract-status`
Historial de una extracción (finished / failed)	`solvento-data-prod.syntage.extractions_history`	Cloud Run `syntage-ciec-extraction-status` → `POST /save-syntage-extraction` (webhook `extraction.updated`) y `GET /search-extractions` (barrido cada 3h)
El cruce `rfc → taxpayer_id` que enriquece las credenciales	lectura de `solvento-data-prod.svc_master_public.taxpayers`	El servicio lee esta tabla (no la escribe)

Resumen

Campo	Valor
Origen	API Syntage `/entities` (por RFC) y `/extractions` (paginado); eventos `extraction.updated`; lectura de `svc_master_public.taxpayers`
Patrón	event-driven (Cloud Run con API REST; webhook por evento + extracción bajo demanda)
Destino	`solvento-data-prod.syntage.daily_credential_status` y `solvento-data-prod.syntage.extractions_history`
Frecuencia	Diaria `0 5 * * ` (job `extract-syntage-credential-status-daily`, `/extract-status`) + cada 3h `0 /3 * * *` (job `extract-syntage-extractions`, `/search-extractions`)
Criticidad	🟢 Baja
Dueño	Martin
Repos	`syntage-ciec-extraction-status`

Diagrama de linaje

graph LR
    API[["🌐 API Syntage<br/>/entities (1 call por RFC)<br/>/extractions (paginado)"]]
    EVT(["📨 evento<br/>extraction.updated"])
    TAX[("svc_master_public.taxpayers<br/><i>rfc → taxpayer_id</i>")]

    subgraph ing["Cloud Run · API REST"]
        CR["syntage-ciec-extraction-status<br/><i>/extract-status · /save-syntage-extraction<br/>/search-extractions · /health</i>"]
        CRED[("solvento-data-prod<br/>syntage.daily_credential_status<br/>MERGE rfc+día")]
        HIST[("syntage.extractions_history")]
    end

    CONS["📊 Reportes / consultas"]

    API --> CR
    EVT --> CR
    TAX --> CR
    CR --> CRED
    CR --> HIST
    CRED --> CONS

    classDef src fill:#fde68a,stroke:#b45309,color:#000;
    classDef store fill:#bfdbfe,stroke:#1e40af,color:#000;
    class API,EVT src;
    class TAX,CRED,HIST store;

Paso a paso

Cloud Run `syntage-ciec-extraction-status` (repo, `syntage_ciec_extraction_status_gcp.py`)

App Flask (gunicorn) que expone una API REST. Lee de un dataset (BIGQUERY_READ_DATASET=svc_master_public) y escribe en otro (BIGQUERY_WRITE_DATASET=syntage).

GET /extract-status (read_bigquery_table) — hace SELECT DISTINCT rfc, taxpayer_id FROM svc_master_public.taxpayers y después recorre el resultado fila por fila, haciendo una llamada HTTP a Syntage por cada RFC (GET /entities?taxpayer.id={rfc}, timeout 30s, secuencial, sin retry ni paralelismo). Usa el taxpayer_id de BigQuery, no el de la API. Acumula de a 100 respuestas y hace MERGE por (rfc, día) contra syntage.daily_credential_status usando CURRENT_DATE().
⚠️ Si la llamada a Syntage falla para un RFC, call_syntage_api devuelve None y ese RFC se saltea en silencio: el job termina en 200 con filas faltantes, no falla. Un día con pocas filas en daily_credential_status apunta acá.
⚠️ Es el endpoint lento del servicio: el tiempo de corrida escala lineal con la cantidad de RFC en taxpayers.
POST /save-syntage-extraction — recibe un evento extraction.updated; solo procesa status finished o failed (y, para finished, ciertas condiciones sobre changes.finishedAt/changes.status), extrae los datos y hace un MERGE por evento (dedupe por extraction_id) contra syntage.extractions_history.
GET /search-extractions — barrido por rango de fechas (start_date/end_date, default últimas 3h1min). Parte el rango en bloques de 7 días y por bloque pagina GET /extractions (itemsPerPage=200) hasta que no hay hydra:next. Además de devolverlas, las persiste en syntage.extractions_history con el mismo MERGE por extraction_id. Escribe en lotes de EXTRACTIONS_BATCH_SIZE (default 400).
⚠️ En la práctica es el único cargador de la tabla, no una red de seguridad. Verificado 2026-07-28: de las 12.153 filas ingestadas en los 30 días previos, el 100% tiene ingested_at en un tick de 3h del scheduler — o sea que el webhook /save-syntage-extraction no aportó ni una fila. Consecuencia: cada corrida fallida = pérdida total de esa ventana de 3h, sin recuperación automática. Revisar si el webhook está configurado en Syntage.
GET /health — health check.

Nota: aunque el docstring del módulo describe el servicio como "lee datos de BigQuery y los expone vía API REST", el código actual también escribe en BigQuery (daily_credential_status, extractions_history) y consume la API de Syntage. La lectura de BigQuery se limita al cruce con taxpayers.

⚠️ Deuda técnica conocida — los MERGE se construyen concatenando SQL. Tanto save_extractions_batch_to_bigquery como merge_credential_status_batch arman la query como un SELECT ... UNION ALL por fila, con los valores interpolados como literales y escapados a mano (escape_sql_string, que solo duplica comillas simples). Dos consecuencias: 1. El tamaño de la query escala con los datos y choca contra el límite de 1024K caracteres de BigQuery (ver "Si falla"). ~69% de cada fila es el COALESCE(SAFE.PARSE_TIMESTAMP(...)) ×3 formatos que genera sql_timestamp_literal para cada uno de los 5 timestamps: 237 chars por timestamp, 1185 por fila. 2. Los datos vienen de una API externa y terminan dentro del texto del SQL. El escapado manual es frágil.

El fix de fondo es cargar el lote a una tabla staging (load_table_from_json) y hacer el MERGE contra ella, o parametrizar con ARRAY<STRUCT<...>> + UNNEST(). En ambos casos la query queda de tamaño fijo. Pendiente.

⚠️ El MERGE de extractions_history es INSERT-only (WHEN NOT MATCHED THEN INSERT, sin WHEN MATCHED), y el barrido /search-extractions no filtra por status — trae lo que la API devuelva. Resultado: si una extracción se captura mientras está en un status intermedio, esa fila queda congelada para siempre; cuando termina, el extraction_id ya existe y el MERGE la saltea. Medido 2026-07-28: 9.173 filas con status no terminal y finished_at NULL (8.597 running, 525 waiting, 51 pending), todas posteriores al 2025-10-24. Un re-barrido no las corrige.

Notar el contraste con daily_credential_status, cuyo MERGE sí tiene WHEN MATCHED THEN UPDATE y por eso se auto-corrige. Para arreglar extractions_history hay que agregar un WHEN MATCHED AND target.finished_at IS NULL THEN UPDATE (o restringir el barrido a status terminales, como ya hace el webhook).

Tablas que alimenta

Tabla	Contenido	Job que la carga
`solvento-data-prod.syntage.daily_credential_status`	Estado diario de credenciales CIEC por RFC (+ taxpayer_id)	Cloud Run `syntage-ciec-extraction-status` (`GET /extract-status`, MERGE rfc+día)
`solvento-data-prod.syntage.extractions_history`	Historial de extracciones (finished / failed)	Cloud Run `syntage-ciec-extraction-status` (`POST /save-syntage-extraction` webhook + `GET /search-extractions` cada 3h; MERGE por `extraction_id`)

Consumo aguas abajo

Auditado el 2026-07-28 con INFORMATION_SCHEMA.JOBS_BY_PROJECT (ventana de 179 días, retención máxima) + metadata de Metabase. Ningún modelo dbt/dataform lee ninguna de las dos tablas: el consumo es 100% Metabase directo sobre syntage.

`daily_credential_status` — tabla viva 🟢

~1.772 SELECT de Metabase en 60 días. 11 cards, en la colección 880 "Risk Dashboards":

Card	Nombre	Dashboard
5634	Invalid CIEC	618 "Invalid CIEC" · 892 "Cuentas Vivas CIEC"
6102	Cuentas vivas CIEC	892 "Cuentas Vivas CIEC"
1618	tax_compliance	— (el de mayor uso: view_count 13.444)
6323	[Prequal Service] PRODUCTION Daily prequal	automatización Make (escenario 4857116)
6327	[Prequal Service] RE-SCAN monthly	Make, mensual
6324 · 6325	[Prequal Service] Bulk V2 DRAFT · funnel waterfall	—
5540 · 5552 · 6238 · 5981	0_revenues_recent_months · Sales Downward Trend · Con extension-chuck · Ultimo ciec status	dashboard contenedor no mapeado

Falsos positivos descartados: dashboard 578 "CIEC Segmentation V1" (card 5202 usa fact_syntage_invoices, no estas tablas) y los 6 cards de HubSpot que aparecen al buscar "credential".

`extractions_history` — sin consumidores 🔴

292 MERGE de escritura en 60 días contra 8 SELECT en 179 días, todos ad-hoc y ninguno recurrente (uno del 2026-06-17 desde el editor de Metabase que nunca se guardó como card; uno del 2026-05-27 de j.tapias). view_count de la tabla en Metabase = 2. Ningún card contiene sus columnas exclusivas (extractor, taxpayer_rfc, extraction_id).

Implicancia: una caída de /search-extractions no tiene impacto downstream. Antes de invertir en endurecer ese endpoint, evaluar deprecar la tabla o pausar la ingesta.

Si falla 🛠️

El impacto depende de cuál de los dos flujos falló — no son equivalentes:

Flujo	Tabla	Impacto downstream
`/extract-status` (diario 5am)	`daily_credential_status`	🟠 Real. 11 cards, dashboards 618 y 892, y las automatizaciones Make del Prequal Service leen esta tabla. Si falla, el estado CIEC del día no existe
`/search-extractions` (cada 3h)	`extractions_history`	🟢 Ninguno. La tabla no tiene consumidores (ver "Consumo aguas abajo"). Solo se pierde historial interno

Criticidad global baja: no bloquea el pipeline analítico.

Reconciliar `extractions_history` después de una caída

El MERGE es idempotente por extraction_id, así que re-invocar el endpoint es seguro (no duplica). Identificar las ventanas faltantes y re-correrlas:

-- Ventanas de 3h sin cobertura. Correlacionar con los logs: una ventana vacía
-- puede ser actividad cero real (típico entre 03:00 y 09:00 UTC) o una corrida caída.
-- La ventana [T, T+3) la carga la corrida de T+3.
WITH ventanas AS (
  SELECT v AS ini, TIMESTAMP_ADD(v, INTERVAL 3 HOUR) AS fin
  FROM UNNEST(GENERATE_TIMESTAMP_ARRAY(
    TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 DAY),
    CURRENT_TIMESTAMP(), INTERVAL 3 HOUR)) AS v
)
SELECT FORMAT_TIMESTAMP('%Y-%m-%dT%H:%M:%S', w.ini) AS start_date,
       FORMAT_TIMESTAMP('%Y-%m-%dT%H:%M:%S', w.fin) AS end_date,
       COUNT(e.extraction_id) AS filas
FROM ventanas w
LEFT JOIN `solvento-data-prod.syntage.extractions_history` e
  ON e.started_at >= w.ini AND e.started_at < w.fin
GROUP BY 1,2 HAVING filas = 0 ORDER BY 1;

Después, por cada ventana faltante:

TOKEN=$(gcloud auth print-identity-token)
URL=$(gcloud run services describe syntage-ciec-extraction-status-prod \
        --region us-central1 --format='value(status.url)')
curl -H "Authorization: Bearer $TOKEN" \
  "$URL/search-extractions?start_date=2026-07-27T15:00:00&end_date=2026-07-27T18:00:00"

⚠️ Una ventana por invocación. Rangos largos disparan el problema de tamaño de query (mitigado con lotes de 400) y pueden pasarse del timeout de Cloud Run, ya que el barrido pagina de forma secuencial.

⚠️ La reconciliación no arregla las filas con status no terminal (ver la nota del MERGE INSERT-only): si la fila ya existe con running, el re-barrido la saltea. Eso requiere el fix del WHEN MATCHED, o un backfill dirigido.

Síntoma	Dónde mirar	Acción
`❌ Error guardando lote acumulado: 400 ... The query is too large`	`save_extractions_batch_to_bigquery` — el MERGE de `/search-extractions` superó los 1024K chars	Bajar `EXTRACTIONS_BATCH_SIZE` (default 400 ≈ 690K, 73% en el peor caso; 600 ya excede). Ver la nota de deuda técnica arriba. Ocurrió el 2026-07-28 con el valor anterior de 800 (1043.57K)
`extractions_history` con huecos en una ventana de 3h	¿Falló la corrida de `extract-syntage-extractions` de ese horario?	El MERGE es idempotente por `extraction_id`: re-invocar `GET /search-extractions?start_date=...&end_date=...` con el rango faltante. Ojo: rangos largos generan más filas por lote — el batching las corta de a 400, pero el barrido es secuencial y puede pasarse del timeout de Cloud Run
No se actualiza el estado de credenciales	Logs Cloud Run `syntage-ciec-extraction-status`	Re-invocar `GET /extract-status` (idempotente: MERGE por rfc+día)
`daily_credential_status` con menos filas que RFC en `taxpayers`, pero el job dio 200	Logs: `Error consultando Syntage API para RFC ...`	Los RFC cuya llamada a Syntage falló se saltean en silencio (`call_syntage_api` → `None`). Re-invocar el endpoint; si persiste, revisar salud de la API de Syntage
Una extracción no aparece en el historial	¿El evento tenía status `finished`/`failed` y cumplía las condiciones de `changes`?	Solo se persisten esos casos; verificar el payload del evento
`taxpayer_id` en NULL para muchos RFC	Tabla `svc_master_public.taxpayers`	El RFC no está en `taxpayers`; el servicio igual guarda con `taxpayer_id` NULL

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

🔍 Este servicio no tiene alertas sobre extractions_history. El dashboard "Data Processes" (dashboard_gcp/main.py) cubre solo daily_credential_status; el comentario del código lo dice explícitamente: "extractions_history (misma Cloud Run, otro endpoint/cadencia) queda fuera por ahora". Por eso la caída del 2026-07-27 y 2026-07-28 pasó dos días sin detectarse. Además, el logging del branch master usa StreamHandler() sin stream → escribe a stderr en texto plano, así que Cloud Run marca todos los logs como severity=ERROR: el panel de "logs >=WARNING" es ruido y no sirve para alertar.

Enlaces

📦 Servicio: syntage-ciec-extraction-status — syntage_ciec_extraction_status_gcp.py
📊 Estado en vivo: dashboard "Data Processes" (Cloud Monitoring)

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