sap-intercompany-worker/docs/FLUJO.md

Flujo de un mensaje, de punta a punta

Este documento traza el camino completo de una notificación de SAP —desde que sale del Webhook Messenger hasta que queda replicada en una subsidiaria— nombrando el archivo y la función concreta que interviene en cada paso. Complementa a ARQUITECTURA.md (conceptos y reglas de negocio) con la mecánica real de código.

Dos proyectos intervienen: sap-webhook-api (recibe el webhook, lo guarda) y sap-intercompany-worker (lo procesa y replica) — separados a propósito por el timeout de 10s de SAP, ver README.md de cada uno.

Diagrama de secuencia

sequenceDiagram
    participant SAP as SAP (Webhook Messenger)
    participant Ctrl as WebhooksController<br/>(sap-webhook-api)
    participant Repo as webhook-events.repository<br/>(sap-webhook-api)
    participant DB as Postgres<br/>webhook_events
    participant Idx as index.ts<br/>(worker, runOnce)
    participant Cfg as replicationConfig.ts
    participant Eng as replicateEvents.ts<br/>processPendingEvents
    participant Mod as scenarioTransform.ts
    participant Scn as scenarios/*.ts
    participant SL as serviceLayer.ts<br/>ServiceLayerClient
    participant Origen as SAP Service Layer<br/>(base origen)
    participant Destino as SAP Service Layer<br/>(base destino)
    participant Att as Postgres<br/>replication_attempts

    SAP->>Ctrl: POST /webhooks/sap (CloudEvent)
    Ctrl->>Ctrl: receiveSapEvent()
    Ctrl->>Repo: saveWebhookEventToInbox(notification)
    Repo->>DB: INSERT status='pending'
    Ctrl-->>SAP: 200 OK

    Note over Idx: cada POLL_INTERVAL_SECONDS (o --once)
    Idx->>Cfg: loadReplicationConfig(prisma)
    Cfg-->>Idx: databases, scenarios, destinations, valueMappings
    Idx->>Eng: processPendingEvents(prisma, cfg)

    Eng->>DB: findMany({status:'pending'})
    DB-->>Eng: eventos pendientes

    loop por cada escenario activo del businessObject
        Eng->>Eng: extractSourceCompanyDb(event.payload)
        Note right of Eng: ¿source coincide con<br/>la base origen del escenario?
        Eng->>SL: new ServiceLayerClient(origen).get(key)
        SL->>Origen: Login + GET /{obj}(key)
        Origen-->>SL: registro completo
        SL-->>Eng: sourceRecord

        Eng->>Mod: loadScenarioModule(code)
        Mod-->>Eng: { filter, transform }
        Eng->>Mod: buildScenarioFilterContext(origen, ...)
        Eng->>Scn: filter(sourceRecord, ctx)
        opt el filtro necesita un dato que no viene en el payload
            Scn->>SL: ctx.get(path) — reusa la sesión del origen
            SL->>Origen: GET (ej. BusinessPartners del CardCode)
            Origen-->>SL: registro relacionado
            SL-->>Scn: respuesta
        end
        Scn-->>Eng: true/false

        alt filter() == true
            alt destinationMode == fixed
                Note right of Eng: destinos = scenario_destinations
            else destinationMode == payload
                Note right of Eng: destino = sourceRecord[routingField]<br/>(un solo destino, resuelto del mensaje)
            end

            loop por cada destino resuelto (uno o varios)
                Eng->>Eng: replicateToDestination(...)
                Eng->>Mod: buildScenarioTransformContext(dest, scenarioId, valueMappings)
                Eng->>Scn: transform(sourceRecord, ctx)
                Scn->>Mod: ctx.mapValue(field, sourceValue)
                Mod-->>Scn: targetValue | undefined
                Scn-->>Eng: payload transformado

                Eng->>SL: new ServiceLayerClient(destino)
                Eng->>SL: resolveDestinationRecord(...)
                SL->>Destino: GET por keyField o por $filter=matchField
                Destino-->>SL: existe / no existe

                alt existe
                    SL->>Destino: PATCH /{destinationBusinessObject}(destKey)
                else no existe
                    SL->>Destino: POST /{destinationBusinessObject}
                end
                Destino-->>SL: HTTP status + body
                SL-->>Eng: resultado

                Eng->>Att: replicationAttempt.create({verbo, status, error, payloadSent, resultKey})
            end
        end
    end

    Eng->>DB: webhookEvent.update({status: done|partial|failed|filtered})

Funciones principales por archivo

sap-webhook-api

Archivo	Función	Qué hace
src/webhooks/webhooks.controller.ts	receiveSapEvent()	Recibe el POST /webhooks/sap, valida HMAC si aplica, guarda el .json local y llama al repositorio.
src/webhooks/webhook-events.repository.ts	saveWebhookEventToInbox()	Inserta la fila en webhook_events (status='pending').

sap-intercompany-worker

Archivo	Función	Qué hace
src/index.ts	runOnce()	Orquesta un tick: carga config, corre el motor. Se llama por cron o una vez con --once.
src/replicationConfig.ts	loadReplicationConfig()	Trae de Postgres las bases, escenarios, destinos y mapeos de valor activos — una sola vez por tick.
src/replicateEvents.ts	processPendingEvents()	El motor: recorre eventos pendientes × escenarios activos (datos maestros y transacciones). Resuelve el/los destino(s) según destinationMode — lista fija (scenario_destinations) en fixed, o sourceRecord[routingField] en payload — y calcula destinationBusinessObject (mismo objeto que el origen salvo que el escenario declare uno distinto).
	replicateToDestination()	Arma el payload para UN destino puntual, resuelve PATCH/POST y registra el intento — la misma función sirve para fixed y payload, ya no importa de dónde salió el código de destino.
	extractSourceCompanyDb()	Saca el CompanyDB del campo source del CloudEvent — protección contra loops.
	resolveDestinationRecord()	Decide si el registro existe en el destino: GET directo (portable) o por $filter sobre matchField (udfLink).
	formatKeySegment() / escapeODataString()	Arman la URL OData según keyType (con o sin comillas) y escapan valores.
	cleanRecord()	Quita metadata OData (@odata.context, etc.) del registro leído.
src/scenarioTransform.ts	loadScenarioModule()	Importa dinámicamente src/scenarios/<code>.ts y valida que exporte filter/transform.
	scenarioFileSlug()	Deriva el nombre de archivo (minúsculas) a partir del code (mayúsculas).
	buildScenarioFilterContext()	Arma el ctx (con get) que recibe filter — permite un GET extra contra la base origen.
	buildScenarioTransformContext()	Arma el ctx (con mapValue) que recibe transform.
	applyScenarioTransform()	Ejecuta transform y valida que devuelva un objeto JSON.
src/scenarios/<code>.ts	filter()	Decide si el escenario aplica a ESE registro (una vez por evento). Puede ser async y usar ctx.get(path) si necesita un dato que no viene en el payload.
	transform()	Arma el payload final para UN destino (una vez por destino), usando ctx.mapValue para los mapeos de campo.
src/serviceLayer.ts	ServiceLayerClient.login/get/post/patch()	Cliente mínimo del Service Layer — login lazy, sin cachear sesión entre corridas.

Dónde termina cada mensaje

• webhook_events.status: pending → processing → done / partial / failed / filtered (ver la tabla de reglas en ARQUITECTURA.md).
• replication_attempts: una fila por cada destino intentado dentro de cada escenario que aplicó — incluye el payload enviado (éxito y falla) y, para keyStrategy=udfLink, la clave real del destino en result_key.