GeorgeandClaude Sonnet 5
ca57403663
Motor de escenarios de replicacion entre bases SAP Business One (Service Layer). Poll periodico sobre webhook_events, escenarios configurables via Prisma (filter + transform por escenario). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
135 lines
7.5 KiB
Markdown
135 lines
7.5 KiB
Markdown
# 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`](./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
|
||
|
||
```mermaid
|
||
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`.
|