Syntage · Insights — Sales / Revenue por cliente

Pide a la API de Syntage el insight de sales-revenue por cliente de cada RFC, publica un mensaje por RFC a Pub/Sub, y Dataflow consume y escribe el resultado en BigQuery (insights.sales_revenue_customers_daily_staging).

-staging no es un ambiente

Todo el proceso es productivo. El sufijo -staging (Cloud Run insights-publisher-staging, topic insights-staging-topic-prod, Dataflow insights-sales-revenue-customers-daily-staging, tabla …_staging) indica que la tabla acumula duplicados: cada corrida diaria/mensual y cada webhook vuelve a escribir los meses del RFC. La capa limpia es la vista insights.sales_revenue_customers_daily_v (último webhook_uuid por rfc, date). Para consumo nuevo, leer de …_v.

¿Llegaste hasta acá desde…?

Si ves…	Eso sale de…	La lógica está en…
Sales-revenue en un modelo de dbt (DSCR)	`solvento-data-prod.insights.sales_revenue_customers_daily_v` (dedup de `_staging`)	`dbt-analytics` → `models/staging/syntage/stg_syntage__sales_revenue_customers.sql`
Facturación / ingresos por cliente (sales-revenue) en un dashboard	`solvento-data-prod.insights.sales_revenue_customers_daily_staging`	Ingesta — Cloud Run `insights-publisher-staging` → Pub/Sub → Dataflow (repo `template-insights`)
Un RFC sin datos recientes de sales-revenue	`solvento-data-prod.insights.sales_revenue_customers_daily_staging`	¿El RFC está `valid` en `/credentials` de Syntage? ¿Se publicó el mensaje a Pub/Sub?

Resumen

Campo	Valor
Origen	API Syntage `/insights/{rfc}/sales-revenue` + lista de RFCs de `/credentials`
Patrón	event-driven (Cloud Scheduler diario + webhook `extraction.updated` → Cloud Run → Pub/Sub → Dataflow)
Destino	`solvento-data-prod.insights.sales_revenue_customers_daily_staging`
Frecuencia	Diaria `0 6 2-31 * ` (job `insights-publisher-staging-daily`, 12 meses) + mensual `0 7 1 ` (job `insights-publisher-staging-monthly-5years`, 5 años) + actualizaciones por webhook por RFC. El servicio sin sufijo `insights-publisher` y sus jobs están deprecados* (ya no existen).
Criticidad	🟡 Media
Dueño	Martin
Repos	`template-insights`

Diagrama de linaje

graph LR
    API[["🌐 API Syntage<br/>/credentials · /insights/{rfc}/sales-revenue"]]
    SCH(["⏰ Cloud Scheduler<br/><i>diario · mensual 5 años</i>"])
    EVT(["📨 evento extraction.updated<br/><i>Pub/Sub push → /webhook</i>"])

    subgraph ing["Ingesta · Cloud Run + Pub/Sub + Dataflow"]
        CR["insights-publisher-staging<br/><i>POST / · /rfc · /webhook · /5years</i>"]
        TOPIC(["insights-staging-topic-prod"])
        DF["Dataflow<br/><i>consume y escribe BQ</i>"]
        RAW[("solvento-data-prod<br/>insights.sales_revenue_<br/>customers_daily_staging<br/><i>con duplicados</i>")]
    end
    VIEW[("insights.sales_revenue_<br/>customers_daily_v<br/><i>vista dedup</i>")]
    DBT["dbt-analytics<br/>stg_syntage__sales_revenue_customers"]

    CONS["📊 Métricas / Metabase"]

    SCH --> CR
    EVT --> CR
    API --> CR
    CR --> TOPIC --> DF --> RAW --> CONS
    RAW --> VIEW --> DBT

    classDef src fill:#fde68a,stroke:#b45309,color:#000;
    classDef store fill:#bfdbfe,stroke:#1e40af,color:#000;
    class API,SCH,EVT src;
    class RAW,VIEW store;

Paso a paso

Ingesta — Cloud Run `insights-publisher-staging` (repo `template-insights`, `cloud_run/insights_publisher_gcp.py`)

El servicio es un publisher Flask: obtiene RFCs y publica un mensaje por RFC al topic Pub/Sub insights-staging-topic-prod en prod (default del código insights-topic, configurable vía PUBSUB_TOPIC). No escribe en BigQuery directamente desde este servicio (la limpieza/escritura directa está deshabilitada por diseño; el endpoint /cleanup devuelve 501). El consumidor del topic (Dataflow) es quien materializa la tabla. Endpoints:

POST / — insights diarios. Toma los RFCs valid de GET /credentials de Syntage (paginado, hydra:member), arma la URL del insight sales-revenue con periodicity=monthly, type=customers para los últimos 12 meses (CLEANUP_MONTHS_BACK, default 12) y publica los mensajes en lotes (ThreadPool, hasta 10 workers). Lo dispara Cloud Scheduler diario.
POST /rfc — push de Pub/Sub para un RFC (mensaje base64 con campo rfc); re-publica los insights diarios solo de ese RFC.
POST /webhook — recibe el evento extraction.updated de Syntage (push de Pub/Sub syntage-extractions); ejecuta la carga de 5 años para el RFC del evento, de forma asíncrona (responde 200 inmediato para acknowledge y procesa en thread aparte, evitando timeouts de Pub/Sub).
POST /5years — Cloud Scheduler mensual: carga de 5 años para todos los RFCs.
GET /health — health check.

El único endpoint de insight configurado es sales-revenue (api_name = sales_revenue_customers_daily). El insight devuelve fechas en formato YYYY-MM (mensual).

Consumo

Métricas / Metabase leen de solvento-data-prod.insights.sales_revenue_customers_daily_staging. El nombre dice "staging" pero es productivo; ojo que esa tabla tiene duplicados.
dbt-analytics lee de la vista dedup insights.sales_revenue_customers_daily_v (stg_syntage__sales_revenue_customers, usado en DSCR).

Tablas que alimenta

Tabla	Contenido	Job que la carga
`solvento-data-prod.insights.sales_revenue_customers_daily_staging`	Sales-revenue por cliente, mensual, por RFC — con duplicados (append por corrida/webhook)	Dataflow (suscriptor de `insights-staging-topic-prod`); publicado por Cloud Run `insights-publisher-staging`

Consumo aguas abajo

Métricas / Metabase de facturación e ingresos por cliente.

Si falla 🛠️

Impacto: los insights de sales-revenue quedan con la última carga buena; criticidad media (insight analítico, no bloquea producto).

Síntoma	Dónde mirar	Acción
No se actualiza sales-revenue	Logs Cloud Run `insights-publisher-staging`; ¿corrió el Scheduler diario?	Re-invocar `POST /` (todos) o `POST /rfc` (un RFC)
Un RFC nuevo no aparece	¿Está `valid` en `/credentials` de Syntage?	El publisher solo procesa RFCs `status=valid` que devuelve la API
Llega el mensaje pero la tabla no cambia	Pub/Sub `insights-staging-topic-prod` y el job de Dataflow consumidor	El publisher solo publica; revisar el Dataflow que escribe la tabla
Webhook no dispara la carga de 5 años	Suscripción push `syntage-extractions` → `/webhook`; logs (procesa async)	Verificar que el evento es `extraction.updated` y trae `rfc`

Escalación: SQL/Metabase → Diego · resto → Martin.

Enlaces

📦 Ingesta: template-insights — cloud_run/insights_publisher_gcp.py
📊 Estado en vivo: dashboard "Data Processes" (Cloud Monitoring)

Última actualización: 2026-10-05 (aclarado el sufijo -staging = duplicados, vista dedup _v + consumo dbt-analytics; deprecado insights-publisher sin sufijo; topic real de prod) · Si cambia el flujo, actualizar esta ficha en el mismo MR.