CRM Adapter — HubSpot

El adapter es una pieza del patrón orquestación del CSB: el bus decide cuándo sincronizar, encola la tarea y reacciona al resultado. El adapter solo sabe hablar con HubSpot.

Cómo encaja con el CSB

1. Un workflow del CSB (saga order-to-hubspot) crea una tarea en la cola de Cloud Tasks del adapter con el SyncPayloadDto como cuerpo.
2. Cloud Tasks invoca POST /sync con un token OIDC firmado por la cuenta de servicio del CSB (csb-run-sa@…), que tiene roles/run.invoker sobre el servicio.
3. El adapter transforma, llama a HubSpot y publica el resultado en el topic csb-crm-results-<env>.
4. El CSB consume la suscripción csb-crm-results-<env>-csb-sub y continúa el flujo.

Bootstrap (main.ts): ValidationPipe global con transform y whitelist (campos desconocidos se descartan), filtro global de excepciones, Swagger bajo /docs/api cuando la documentación está habilitada, escucha en PORT (8080).

1. Cloud Tasks → POST /syncCuerpo SyncPayloadDto. Headers inyectados por la cola: Authorization: Bearer <oidc>, X-Task-Queue, X-Retry-Count, X-Workflow-Execution.
2. ValidaciónValidationPipe valida envelope y taskMetadata. Si falla: 400 y la tarea reintenta (ver errores).
3. TransformaciónHubSpotTransformer.transform() lee envelope.data y arma properties. Sin email lanza error → 500.
4. Crear contactoPOST https://api.hubapi.com/crm/v3/objects/contacts con Bearer token, timeout 15 s. Éxito → action: created.
5. Si HubSpot responde 409 (ya existe)POST /crm/v3/objects/contacts/search filtrando email EQ → toma el primer id → PATCH /crm/v3/objects/contacts/{id}. Éxito → action: updated.
6. Publicar resultadoResultPublisherService.publishSuccess() publica el CloudEvent en PUBSUB_RESULT_TOPIC. Si la publicación falla, la tarea completa falla (500) y Cloud Tasks reintenta.
7. Respuesta 200 { "status": "ok" }Cloud Tasks marca la tarea como completada. Todo el proceso deja logs con event_id y duration_ms.

Todos responden JSON. La referencia interactiva con esquemas generados desde los DTOs está en Swagger UI; el documento OpenAPI 3.1 en /docs/openapi.json.

{
  "envelope": {
    "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",   // UUID v4 del evento original
    "source": "//bebbia/orders-service",
    "type": "com.bebbia.order.placed.v1",
    "specversion": "1.0",
    "entityid": "ORD-000123",                          // ordering key
    "datacontenttype": "application/json",               // opcional
    "time": "2026-09-09T15:04:05.000Z",                  // opcional
    "data": {
      "email": "ana.lopez@example.com",               // requerido por el transformer
      "first_name": "Ana",
      "last_name": "López",
      "phone": "+52 55 1234 5678",
      "company": "Bebbia"
    }
  },
  "taskMetadata": {
    "retryCount": 0,                                   // requerido, >= 0
    "queueName": "csb-hubspot-adapter-queue-dev",      // opcional
    "taskName": "projects/.../tasks/..."              // opcional
  }
}

{ "status": "ok" }

curl -s -X POST http://localhost:8080/sync \
  -H "Content-Type: application/json" \
  -d '{"envelope":{"id":"11111111-1111-4111-8111-111111111111","source":"//bebbia/orders-service","type":"com.bebbia.order.placed.v1","specversion":"1.0","entityid":"ORD-1","data":{"email":"ana@example.com","first_name":"Ana"}},"taskMetadata":{"retryCount":0}}'

{
  "status": "ok",
  "timestamp": "2026-09-09T15:04:05.000Z",
  "version": "0.1.0",
  "component": "csb-crm-adapter",
  "checks": {
    "hubspot_connectivity": "ok",
    "secret_manager": "ok"
  }
}
// status: ok | degraded — siempre HTTP 200

{ "status": "ok", "timestamp": "2026-09-09T15:04:05.000Z" }

Entrada — SyncPayloadDto (Cloud Tasks)

Mapeo a HubSpot — propiedades del contacto

Salida — CloudEvent de resultado (Pub/Sub)

Publicado en el topic PUBSUB_RESULT_TOPIC (Terraform: csb-crm-results-<env>) como mensaje JSON con atributos ce_type, ce_source, ce_specversion.

{
  "specversion": "1.0",
  "type": "hubspot.contact.synced.v1",
  "source": "//csb-crm-adapter/rtp-transversal-dev",
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890-result",   // <originalEventId>-result
  "time": "2026-09-09T15:04:06.120Z",
  "datacontenttype": "application/json",
  "data": {
    "originalEventId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "email": "ana.lopez@example.com",
    "hubspotId": "901234567",
    "action": "created",                                  // created | updated
    "status": "SUCCESS"
  }
}

Solo se publican éxitos. Un fallo no genera evento: la tarea responde 500 y Cloud Tasks reintenta; agotados los reintentos, la cola descarta la tarea (no hay DLQ configurada en este adapter).

Formato de error (HttpExceptionFilter)

{
  "statusCode": 500,
  "timestamp": "2026-09-09T15:04:05.000Z",
  "path": "/sync",
  "error": { "message": "Internal server error" }   // en 400: objeto de ValidationPipe con "message": [...]
}

Qué produce cada código

• 400Envelope o taskMetadata inválidos (campos faltantes, tipos incorrectos). Cloud Tasks reintenta igual: corregir el productor.
• 500Missing required field: email; HubSpot no-2xx distinto de 409 (auth 401, rate limit 429, 400 por propiedades inexistentes); contacto no encontrado tras 409; timeout 15 s; fallo al publicar en Pub/Sub. El mensaje real queda en logs, no en la respuesta.
• 200Contacto creado o actualizado y evento publicado.

Política de reintentos (Cloud Tasks, Terraform)

max_attempts

5

backoff

10 s → 300 s, max_doublings 4

max_retry_duration

3600 s

rate

10 dispatches/s · 5 concurrentes

Códigos de error declarados en ErrorResponseDto (HUBSPOT_TIMEOUT, HUBSPOT_AUTH_FAILED, HUBSPOT_RATE_LIMITED, SCHEMA_VALIDATION_FAILED) describen la intención del contrato; hoy el filtro global no los emite en la respuesta y el log usa error_code: HUBSPOT_SYNC_ERROR.

De extremo a extremo: qué sistema dispara, por qué puerta entra al Core Service Bus, cómo el bus localiza e invoca a este adapter, qué flujos lo usan hoy, qué hace con el resultado y qué le falta a cada pieza. Fuente: rtp-csb en dev (seed, workflows/, infra/envs/dev) al 10 sep 2026.

Regla de la arquitectura: solo el CSB le pega a HubSpot. Ningún sistema externo llama al adapter; publican un evento o llaman al WebHook del CSB, y es un Cloud Workflow (saga) quien decide cuándo y con qué invocar a este adapter.

El CSB tiene tres puertas de entrada. Las dos primeras son para sistemas; la tercera para personas. Un externo (tienda Bebbia, Data Mesh, Portal de Inventario, CommerceTools) nunca conoce la URL del adapter: conoce un topic o un WebHook del CSB.

Ejemplo 1 · publicar el evento que dispara order-to-hubspot-saga

Contrato Avro order.placed.v1 (schemas/order.placed.v1.avsc). El envelope CloudEvents lo agrega el CSB; el productor solo publica el payload de dominio.

{
  "event_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "order_id": "ORD-000123",
  "customer_id": "CUST-000045",
  "items": [
    {
      "sku_code": "BEB-PURIF-01",
      "product_name": "Purificador Bebbia",
      "quantity": 1,
      "unit_price": "1499.00",
      "total_price": "1499.00"
    }
  ],
  "total_amount": "1499.00",
  "currency": "MXN",
  "placed_at": 1788000000000,
  "channel": "WEB",
  "correlation_id": "corr-7f3a"
}

gcloud pubsub topics publish order.placed.v1 \
  --project=rtp-transversal-dev \
  --ordering-key=ORD-000123 \
  --message="$(cat order-placed.json)"

Ejemplo 2 · llamar un WebHook de flujo (patrón de la tienda Bebbia)

BODY='{"requestId":"REQ-0001","susid":"SUS-100045","storeId":"bebbia-web", ...}'
SIG=$(printf '%s' "$BODY" | openssl dgst -sha256 -hmac "$WEBHOOKS_SIGNING_KEY" | awk '{print $2}')

curl -s -X POST "https://csb-dev.rotoplas.com/api/flows/$FLOW_ID/webhook" \
  -H "Content-Type: application/json" \
  -H "X-CSB-Signature: $SIG" \
  -d "$BODY"
# executeFlow es síncrono: la respuesta ya trae el resultado del workflow

1. El adapter publica sus datos de contacto en su propio TerraformOutputs en rtp-crm-adapter/infra/envs/dev: service_url, queue_name (csb-hubspot-adapter-queue-dev), result_topic_id (csb-crm-results-dev). Estado en gs://rtp-transversal-dev-tfstate/crm-adapter/dev.
2. El CSB los lee por remote state, no los hardcodeartp-csb/infra/envs/dev/main.tf (commit 40352ce): data "terraform_remote_state" "crm_adapter" → local.adapters.crm = { url, queue, tasks_sa = null, result_topic }. Output adapter_endpoints muestra lo resuelto.
3. La URL entra a csb-api y al job de seed como variable de entornomodule.compute.extra_env.CRM_ADAPTER_URL se inyecta en el Cloud Run csb-api y en el job csb-api-migrate.
4. El seed registra el adapter en el registry del CSBapps/api/src/database/seed.ts: ApiDefinition HubSpot CRM v3 API (baseUrl = CRM_ADAPTER_URL), Connector hubspot-order-connector (mapping order_id→dealname, total_amount→amount, customer_id→associatedContactId; throttle 10/s, 15 concurrentes, 3 intentos) y Adapter csb-hubspot-adapter con cloudRunUrl = CRM_ADAPTER_URL, probe GET /health. Visible en csb-dev.rotoplas.com/registry/adapters.
5. Identidad: quién puede invocar al adapterEste adapter usa el modelo inverso: su Terraform otorga roles/run.invoker a csb-run-sa@rtp-transversal-dev.iam.gserviceaccount.com. Cloud Workflows y Cloud Tasks del CSB deben firmar el OIDC con esa SA; si el workflow corre con otra SA, la llamada recibe 403.
6. Los flujos referencian el adapter por id y el workflow lo llamaEn el grafo del flujo el nodo adapter apunta a csb-hubspot-adapter (refId). El YAML del workflow lee HUBSPOT_ADAPTER_URL con sys.get_env y hace http.post con auth: { type: OIDC } y header X-Workflow-Execution. Los pasos del editor con auth: 'OIDC' hacen lo mismo.
7. Cloud Tasks cuando el trabajo es diferidoContrato HUBSPOT_ADAPTER_QUEUE en packages/shared/src/contracts/cloud-tasks.contracts.ts: cola csb-hubspot-queue (Terraform del CSB, 100/s, 10 concurrentes, 5 intentos), destino POST {HUBSPOT_ADAPTER_URL}/sync, cuerpo SyncPayloadDto, headers Authorization: Bearer <oidc>, X-Task-Queue, X-Retry-Count, X-Workflow-Execution. deployFlow crea además una cola que-<flow>-<adapter> por cada edge enqueue del editor.
8. El resultado vuelve por Pub/SubEl adapter publica hubspot.contact.synced.v1 en csb-crm-results-dev; su Terraform crea la suscripción csb-crm-results-dev-csb-sub para que el CSB la consuma.

Secuencia real de order-to-hubspot-saga (como está escrita)

1. Eventarc → Cloud WorkflowEl topic order.placed.v1 dispara el workflow con args.envelope y args.taskMetadata.
2. validate_payloadExige envelope.id, envelope.data.order_id, envelope.data.customer_id; si falta algo lanza VALIDATION_ERROR.
3. resolve_hubspot_contactPOST {HUBSPOT_ADAPTER_URL}/contacts/resolve con { customerId, email }; espera hubspotContactId.
4. transform_to_hubspotPOST /deals/transform con { envelope, hubspotContactId, preview: true }; espera data con el deal.
5. sync_to_hubspotPOST /deals/sync con { envelope, dealPayload, hubspotContactId } y header X-Workflow-Execution; espera dealId.
6. handle_success / handle_failureÉxito: retorna { status: SUCCESS, orderId, hubspotDealId, hubspotContactId } (no publica en Pub/Sub). Fallo: publica el envelope en order.placed.v1.dlq y retorna FAILED.

Lo que este adapter sí sabe hacer hoy: POST /sync

Cuerpo exacto que espera (el que envía Cloud Tasks según el contrato HUBSPOT_ADAPTER_QUEUE):

{
  "envelope": {
    "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "source": "//bebbia/orders-service",
    "type": "com.bebbia.order.placed.v1",
    "specversion": "1.0",
    "entityid": "ORD-000123",
    "data": {
      "email": "ana.lopez@example.com",
      "first_name": "Ana",
      "last_name": "López",
      "phone": "+52 55 1234 5678",
      "company": "Bebbia"
    }
  },
  "taskMetadata": {
    "retryCount": 0,
    "queueName": "csb-hubspot-queue"
  }
}

El flujo order-fulfillment-demo envía en cambio { "envelope": { "id", "entityid", "correlationid", "data": { "customer_id", "email" } } }: sin source, type, specversion ni taskMetadata, por lo que el ValidationPipe responde 400. El paso de deal ni siquiera trae email.

Un sistema externo nunca llama al adapter. Tiene dos maneras de usar lo que el adapter hace y una de enterarse del resultado:

Data Mesh hoy es destino, no productor: CSB-01 le hace POST /v1/domains/inventory/assets/{serialNumber}/status-events (ApiDefinition Core Data Mesh Rotoplas, token BEARER, secreto csb-dev-data-mesh-credential) y usa su lastEventId como candado de idempotencia. Para que consuma resultados de HubSpot aplica la tercera fila.

1. Confirmar outputs del adapter

   cd rtp-crm-adapter/infra/envs/dev && terraform output
   # service_url, queue_name, result_topic_id, crm_adapter_sa

2. Confirmar que el CSB los resuelve

   cd rtp-csb/infra/envs/dev && terraform output -json adapter_endpoints | jq .crm

3. Alinear la identidadLa SA con run.invoker en este adapter es csb-run-sa@…; la de csb-api es sa-csb-api-dev@… y la de cada workflow la crea el módulo csb-workflow (gcloud workflows describe order-to-hubspot-saga --format='value(serviceAccount)'). Otorgar run.invoker a las SAs reales en rtp-crm-adapter/infra/envs/dev/terraform.tfvars (o convertir csb_invoker_sa_email en lista).
4. Inyectar la URL en los workflowsAgregar user_env_vars = { HUBSPOT_ADAPTER_URL = local.adapters.crm.url } al módulo csb-workflow para order-to-hubspot-saga, order-cancellation-notify y order-fulfillment-orchestrator.
5. Sembrar el registryEl job csb-api-migrate corre el seed con CRM_ADAPTER_URL. En local: CRM_ADAPTER_URL=http://localhost:8082 npm run seed en apps/api. Revisar en /registry/adapters que csb-hubspot-adapter tenga la URL y el probe en verde.
6. Exponer en el adapter lo que los flujos llamanImplementar POST /contacts/resolve, POST /deals/transform y POST /deals/sync (ya en NEXT_STEPS.md) o cambiar los pasos de los workflows para usar /sync con SyncPayloadDto completo.
7. Construir o ajustar el flujo en el editor/flows/:id: nodo topic → workflow (pasos call con auth: OIDC hacia {{CRM}}/sync) → nodo adapter (csb-hubspot-adapter) → topic de resultado. Preview GCP muestra la cola que-<flow>-csb-hubspot-adapter; Deploy la crea.
8. Ejecutar y observarPOST /api/provisioning/flows/{id}/execute con el inputTemplate; ver /monitoring (executions, tasks, dlq) y en el adapter los logs sync_request_received → sync_completed.
9. Cerrar el ciclo del resultadoDecidir un solo topic de resultado (csb-crm-results-dev con CloudEvent hubspot.contact.synced.v1, o hubspot.order.synced.v1 del seed) y crear el consumer o la suscripción del CSB que lo lea.

• bloqueanteRutas inexistentes. Los tres workflows Terraform llaman a /contacts/resolve, /deals/transform y /deals/sync; el adapter solo tiene /sync. Toda ejecución termina en CONTACT_RESOLVE_ERROR y el envelope va a order.placed.v1.dlq.
• bloqueanteContrato de /sync. Exige envelope CloudEvents completo (source, type, specversion, entityid) y taskMetadata.retryCount; los pasos del editor (order-fulfillment-demo) envían un envelope mínimo → 400. Además data.email es obligatorio y el paso de deal no lo trae → 500.
• legadoDos adapters HubSpot en paralelo. csb-compute aún declara el Cloud Run hubspot-adapter interno y la suscripción push order.placed.v1 → hubspot-adapter/sync; los workflows leen HUBSPOT_ADAPTER_URL. Hay que apuntar ambos a local.adapters.crm.url y retirar el servicio interno.
• deudaTopic de resultado sin acuerdo. Adapter: csb-crm-results-dev / hubspot.contact.synced.v1. Seed: hubspot.order.synced.v1. Nadie consume ninguno.
• bloqueanteIdentidad. El adapter autoriza a csb-run-sa@…, pero la SA de csb-api es sa-csb-api-dev@… y cada Cloud Workflow corre con su propia SA (módulo csb-workflow). Ninguna de las dos tiene run.invoker aquí: toda llamada OIDC desde el CSB recibe 403.
• bloqueanteVariables de los workflows. Los YAML leen HUBSPOT_ADAPTER_URL con sys.get_env, pero el módulo csb-workflow no declara user_env_vars: el workflow desplegado no conoce la URL del adapter.
• deudaPush a /events. Cuando un flujo del editor conecta un topic directo a un nodo adapter, deployFlow crea una suscripción push a <cloudRunUrl>/events, ruta que no existe en ningún adapter.
• hechoRemote state, CRM_ADAPTER_URL en csb-api y seed, registro en el registry, contrato HUBSPOT_ADAPTER_QUEUE, DLQ y monitoring.

loadHubSpotConfig() opera en dos modos: con USE_SECRET_MANAGER=false lee todo de variables de entorno (local, docker compose); con true (Cloud Run) obtiene el token de Secret Manager usando el sufijo de ambiente derivado de NODE_ENV (production → prod).

Variables de entorno

Secretos (Secret Manager)

La SA del servicio (csb-crm-adapter-<env>@…) tiene secretmanager.secretAccessor solo sobre ese secreto y pubsub.publisher a nivel proyecto.

Toda la infraestructura vive en infra/ con un módulo de cómputo, uno de secretos y una carpeta por ambiente (envs/dev, envs/qa, envs/prod). Estado remoto en GCS. Cloud Build aplica envs/dev en cada push a dev.

Outputs de Terraform

• service_url — registrar en el CSB como CRM_ADAPTER_URL (destino de las tareas).
• queue_name, result_topic_id, crm_adapter_sa.

Trigger: push a dev en GitLab. Pipeline en cloudbuild.yaml, máquina E2_HIGHCPU_8, timeout 30 min. La imagen se etiqueta con $SHORT_SHA y dev-latest.

1. security-audit · lint · testEn paralelo. npm audit --audit-level=high, npm run lint, npm run test:cov con umbral de cobertura global 65 % (líneas y funciones).
2. build · pushDocker multi-stage (deps → builder → runner non-root). Push a Artifact Registry csb-adapters/crm-adapter.
3. terraform init → plan → applyEn infra/envs/dev con image_tag=$SHORT_SHA y csb_invoker_sa_email.
4. deploy --no-trafficNueva revisión sin tráfico.
5. smoke-testEspera (12 × 5 s) a que la revisión quede Ready=True. No hay llamada HTTP: el ingress interno impide que Cloud Build alcance el servicio.
6. promote-trafficupdate-traffic --to-latest solo si el smoke test pasó.

Los logs salen por stdout/stderr y los recoge Cloud Logging. Las excepciones HTTP se escriben como JSON en stderr con severity: ERROR; el resto usa el Logger de NestJS con objetos que incluyen siempre component: "csb-crm-adapter".

Acciones registradas

Consulta útil en Cloud Logging

resource.type="cloud_run_revision"
resource.labels.service_name="rtp-transversal-dev-dev-crm-adapter"
(jsonPayload.action="sync_failed" OR severity>=ERROR)

Métricas nativas de Cloud Run (request count por clase de código, latencias, instancias) están disponibles sin configuración extra. Este repo no declara políticas de alerta propias.

• pendienteEndpoints de la saga order-to-hubspot: POST /contacts/resolve, POST /deals/transform, POST /deals/sync referenciados por el CSB y aún no implementados.
• pendienteConvención de naming (revisión de Hugo, 26 ago 2026): rutas en inglés/REST bajo /crm/…, secretos con prefijo crm-<env>-… y repositorio de Artifact Registry propio crm-adapter. El código actual sigue usando csb- y csb-adapters.
• deudaDos variables para el topic: RESULT_TOPIC se carga en la config pero ResultPublisherService lee PUBSUB_RESULT_TOPIC directamente. Unificar para evitar publicar en un topic equivocado.
• deudaLogger estructurado no conectado: StructuredLoggerService existe pero main.ts no lo registra con app.useLogger(); los objetos pasan por el logger por defecto de NestJS.
• deudaSin dead-letter: tras 5 intentos la tarea se descarta silenciosamente. Evaluar DLQ o publicación de evento de fallo.
• fueraMarketing Hub, webhooks de HubSpot → CSB y Custom Objects son iniciativas separadas.
• hechoMigración desde rtp-csb, Terraform propio, pipeline con canary (no-traffic + Ready check + promote), portal de documentación servido por el servicio.